From 335164eaf1b558e45a3cf377ed73f756716e05e6 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Thu, 18 Sep 2025 16:23:18 +0100 Subject: [PATCH 1/7] multiple remote assessments --- asset_list/AssetList.py | 21 ++++-- asset_list/app.py | 88 ++++++++++++++++++++++++++ asset_list/mappings/built_form.py | 4 +- asset_list/mappings/heating_systems.py | 7 +- asset_list/mappings/roof.py | 30 +++++++++ asset_list/mappings/walls.py | 12 +++- backend/apis/GoogleSolarApi.py | 1 - 7 files changed, 154 insertions(+), 9 deletions(-) diff --git a/asset_list/AssetList.py b/asset_list/AssetList.py index 9569afe8..dce929ae 100644 --- a/asset_list/AssetList.py +++ b/asset_list/AssetList.py @@ -1783,9 +1783,16 @@ class AssetList: ) ) - not_a_flat = ( - self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] != "flat" - ) + # Determine if the client gave us property type in the first place + if all(self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] == "unknown"): + # Use EPC + not_a_flat = ( + self.standardised_asset_list[self.EPC_API_DATA_NAMES["property-type"]] != "Flat" + ) + else: + not_a_flat = ( + self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] != "flat" + ) solar_roof_meets_criteria = ( self.standardised_asset_list["solar_epc_roof_insulated"] | @@ -3452,7 +3459,13 @@ class AssetList: raise ValueError("No installer column found in master data") measure_mix_col = "MEASURE COMBO" - town_colname = "TOWN" if "TOWN" in master_data.columns else 'Town/Area' + + if "TOWN" in master_data.columns: + town_colname = "TOWN" + elif 'Town/Area' in master_data.columns: + town_colname = 'Town/Area' + else: + town_colname = "Town/City" logger.info("Matching master data to asset list") matched = [] diff --git a/asset_list/app.py b/asset_list/app.py index 01c31f0f..833050fb 100644 --- a/asset_list/app.py +++ b/asset_list/app.py @@ -59,6 +59,74 @@ def app(): Property UPRN """ + # CDS - Sept 2025 + data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/CDS/September 2025 Programme" + data_filename = "Founder Estates CDS.xlsx" + sheet_name = "Combined List" + postcode_column = 'Postcode' + address1_column = None # Is only patchily populated so we create it + address1_method = 'house_number_extraction' + fulladdress_column = "Address" + address_cols_to_concat = [] + missing_postcodes_method = None + landlord_year_built = None + landlord_os_uprn = None + landlord_property_type = "Property Type" + landlord_built_form = None + landlord_wall_construction = None + landlord_roof_construction = None + landlord_heating_system = "Heating Type" + landlord_existing_pv = None + landlord_property_id = "(Do Not Modify) Property" + landlord_sap = None + outcomes_filename = None + outcomes_sheetname = None + outcomes_postcode = None + outcomes_houseno = None + outcomes_id = None + outcomes_address = None + master_filepaths = [] + master_id_colnames = [] + master_to_asset_list_filepath = None + phase = False + ecosurv_landlords = None + asset_list_header = 0 + landlord_block_reference = None + + # Project from Nick + data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/sfr/Sep2025 Project" + data_filename = "AL Test.xlsx" + sheet_name = "Sheet1" + postcode_column = 'postcode' + address1_column = None + address1_method = 'house_number_extraction' + fulladdress_column = "address" + address_cols_to_concat = [] + missing_postcodes_method = None + landlord_year_built = None + landlord_os_uprn = None + landlord_property_type = None + landlord_built_form = None + landlord_wall_construction = None + landlord_roof_construction = None + landlord_heating_system = None + landlord_existing_pv = None + landlord_property_id = "row_id" + landlord_sap = None + outcomes_filename = None + outcomes_sheetname = None + outcomes_postcode = None + outcomes_houseno = None + outcomes_id = None + outcomes_address = None + master_filepaths = [] + master_id_colnames = [] + master_to_asset_list_filepath = None + phase = False + ecosurv_landlords = None + asset_list_header = 0 + landlord_block_reference = None + # Lambeth data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Lambeth" data_filename = "LAMBETH Asset List ( Incomplete).xlsx" @@ -1307,6 +1375,26 @@ def app(): filename = os.path.join(data_folder, ".".join(data_filename.split(".")[:-1])) + " - Standardised.xlsx" # Store the data in two tabs. One for the asset list with the EPC data and the second with the flat data + # Determine inspections priority + # solar_jobs = asset_list.standardised_asset_list[~pd.isnull(asset_list.standardised_asset_list["solar_reason"])][ + # "domna_postcode"].unique() + # asset_list.standardised_asset_list["in_solar_area"] = asset_list.standardised_asset_list["domna_postcode"].isin( + # solar_jobs + # ) + # # Same for cav + # cavity_jobs = asset_list.standardised_asset_list[ + # ~pd.isnull(asset_list.standardised_asset_list["cavity_reason"]) + # ]["domna_postcode"].unique() + # asset_list.standardised_asset_list["in_cavity_area"] = asset_list.standardised_asset_list["domna_postcode"].isin( + # cavity_jobs + # ) + # # We prioritise properties that are in solar areas and cavity areas + # import numpy as np + # asset_list.standardised_asset_list["inspection_priority"] = np.where( + # asset_list.standardised_asset_list["in_solar_area"] | asset_list.standardised_asset_list["in_cavity_area"], + # 1, 2 + # ) + with pd.ExcelWriter(filename) as writer: asset_list.standardised_asset_list.to_excel(writer, sheet_name="Standardised Asset List", index=False) if asset_list.block_analysis_df is not None: diff --git a/asset_list/mappings/built_form.py b/asset_list/mappings/built_form.py index 0dc51129..bdd82883 100644 --- a/asset_list/mappings/built_form.py +++ b/asset_list/mappings/built_form.py @@ -438,6 +438,6 @@ BUILT_FORM_MAPPINGS = { 'Maisonette - Mid Terrace': 'mid-terrace', 'Chalet - Wheelchair': 'unknown', 'Studio Flat': 'unknown', - 'Bungalow - Attached': 'semi-detached' - + 'Bungalow - Attached': 'semi-detached', + 'ND': 'unknown' } diff --git a/asset_list/mappings/heating_systems.py b/asset_list/mappings/heating_systems.py index 424b9b46..4ab8ca72 100644 --- a/asset_list/mappings/heating_systems.py +++ b/asset_list/mappings/heating_systems.py @@ -473,5 +473,10 @@ HEATING_MAPPINGS = { 'Boiler and radiators, oil': 'oil boiler', 'Boiler and radiators, electric': 'electric boiler', 'No system present: electric heaters assumed': 'electric radiators', - 'Boiler and radiators, anthracite': 'solid fuel' + 'Boiler and radiators, anthracite': 'solid fuel', + + 'Heat networks Heat networks (mains gas)': 'communal heating', + 'ND Oil': 'oil fuel', + 'Boiler Biofuel': 'boiler - other fuel' + } diff --git a/asset_list/mappings/roof.py b/asset_list/mappings/roof.py index 60f0473c..8ac926c0 100644 --- a/asset_list/mappings/roof.py +++ b/asset_list/mappings/roof.py @@ -246,4 +246,34 @@ ROOF_CONSTRUCTION_MAPPINGS = { 'Pitched, 150 mm loft insulation': 'pitched insulated', 'Flat, limited insulation (assumed)': 'flat uninsulated', + 'Pitched (no access to loft) 350mm': 'pitched insulated', + 'Pitched (no access to loft) 200mm': 'pitched insulated', + 'Pitched (access to loft) 200mm': 'pitched insulated', + 'Pitched (no access to loft) 250mm': 'pitched insulated', + 'Pitched (access to loft) 100mm': 'pitched insulated', + 'Another dwelling above ND (inferred)': 'another dwelling above', + 'Pitched (no access to loft) N/A': 'pitched no access to loft', + 'Pitched (no access to loft) ND (inferred)': 'pitched no access to loft', + 'Pitched (no access to loft) 150mm': 'pitched insulated', + 'Pitched (access to loft) 400mm+': 'pitched insulated', + 'Pitched (no access to loft) 300mm': 'pitched insulated', + 'Pitched (access to loft) <25mm': 'pitched less than 100mm insulation', + 'Pitched (access to loft) None': 'pitched less than 100mm insulation', + 'Pitched (access to loft) 300mm': 'pitched insulated', + 'Pitched (access to loft) 50mm': 'pitched less than 100mm insulation', + 'Pitched (access to loft) 270mm': 'pitched insulated', + 'Pitched (access to loft) Non-joist': 'pitched access to loft', + 'Pitched (access to loft) 250mm': 'pitched insulated', + 'Another dwelling above N/A': 'another dwelling above', + 'Pitched (access to loft) 150mm': 'pitched insulated', + 'Pitched (access to loft) ND (inferred)': 'pitched access to loft', + 'Pitched (access to loft) 350mm': 'pitched insulated', + 'Pitched (access to loft) NR': 'pitched unknown insulation', + 'Pitched (access to loft) 75mm': 'pitched less than 100mm insulation', + 'Pitched (access to loft) N/A': 'pitched access to loft', + 'ND (inferred) 250mm': 'unknown insulated', + 'Pitched (vaulted ceiling) Non-joist': 'pitched unknown insulation', + 'ND (inferred) ND (inferred)': 'unknown', + 'Flat Non-joist': 'flat insulated', + 'Same dwelling above N/A': 'another dwelling above' } diff --git a/asset_list/mappings/walls.py b/asset_list/mappings/walls.py index 14e4565c..73db586e 100644 --- a/asset_list/mappings/walls.py +++ b/asset_list/mappings/walls.py @@ -342,5 +342,15 @@ WALL_CONSTRUCTION_MAPPINGS = { 'Solid brick, as built, partial insulation (assumed)': 'insulated solid brick', 'Sandstone, as built, no insulation (assumed)': 'uninsulated sandstone or limestone', 'System built, as built, partial insulation (assumed)': 'system built unknown insulation', - 'Timber frame, with external insulation': 'insulated timber frame' + 'Timber frame, with external insulation': 'insulated timber frame', + + 'Cob As-built': 'cob', + 'System built Unknown insulation': 'system built unknown insulation', + 'Solid brick Unknown insulation': 'solid brick unknown insulation', + 'Timber frame Internal': 'insulated timber frame', + 'System built External': 'insulated system built', + 'Stone As-built': 'uninsulated sandstone or limestone', + 'System built As-built': "uninsulated system built", + 'System built Internal': 'insulated system built', + } diff --git a/backend/apis/GoogleSolarApi.py b/backend/apis/GoogleSolarApi.py index 043f41a9..532afec0 100644 --- a/backend/apis/GoogleSolarApi.py +++ b/backend/apis/GoogleSolarApi.py @@ -332,7 +332,6 @@ class GoogleSolarApi: ) if solar_product is None: - logger.info("No suitable solar product found for the configuration with %d panels.", total_panels) continue total_cost = Costs.solar_pv( From d3f941349aa08bbe46f1f28f7e2440dc3894fe24 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Mon, 22 Sep 2025 10:50:05 +0100 Subject: [PATCH 2/7] implementing decent homes wf --- backend/engine/engine.py | 2 +- etl/bill_savings/KwhData.py | 2 +- .../waltham_forest/decent_homes_pilot.py | 442 ++++++++++++++++++ 3 files changed, 444 insertions(+), 2 deletions(-) create mode 100644 etl/customers/waltham_forest/decent_homes_pilot.py diff --git a/backend/engine/engine.py b/backend/engine/engine.py index 2e1ede79..cc17222f 100644 --- a/backend/engine/engine.py +++ b/backend/engine/engine.py @@ -900,7 +900,7 @@ async def model_engine(body: PlanTriggerRequest): r["uplift_project_score"] ) = funding.get_innovation_uplift( measure=r, - starting_sap=p.data["current-energy-efficiency"], + starting_sap=int(p.data["current-energy-efficiency"]), floor_area=p.floor_area, is_cavity=p.walls["is_cavity_wall"], current_wall_uvalue=current_wall_u_value, diff --git a/etl/bill_savings/KwhData.py b/etl/bill_savings/KwhData.py index 24ce9f2c..3291e909 100644 --- a/etl/bill_savings/KwhData.py +++ b/etl/bill_savings/KwhData.py @@ -310,7 +310,7 @@ class KwhData: False: "N", None: "N", "Y": "Y", - "N": "N" + "N": "N", } for v in bools_to_remap: epc[v] = bool_map[epc[v]] diff --git a/etl/customers/waltham_forest/decent_homes_pilot.py b/etl/customers/waltham_forest/decent_homes_pilot.py new file mode 100644 index 00000000..78460f5a --- /dev/null +++ b/etl/customers/waltham_forest/decent_homes_pilot.py @@ -0,0 +1,442 @@ +import json +import os + +import pandas as pd + +from datetime import datetime + + +def years_between(d1, d2): + # precise year difference (accounts for months/days) + return (d1.year - d2.year) - ((d1.month, d1.day) < (d2.month, d2.day)) + + +def get_element(elements, label): + """Safely get an element dict by display label (your JSON keys).""" + return elements.get(label) + + +def adequacy_result_by_text(attr_desc: str): + """ + Generic adequacy parser. + Pass if description clearly says 'Adequate' and not 'Inadequate'. + Fail if it says 'Inadequate' (or equivalent). + Unknown -> 'no_data' + """ + if not attr_desc or not isinstance(attr_desc, str): + return "no_data" + text = attr_desc.strip().lower() + # Common patterns + if "inadequate" in text or "unsatisfactory" in text or "problems" in text: + return "fail" + if "adequate" in text or "standard" in text or "appropriate" in text: + return "pass" + return "no_data" + + +def append_result(decent_homes, variable, result, install_date=None): + decent_homes.append({ + "variable": variable, + "result": result, + "hhsrs_rank": None, + "hhsrs_score": None, + "install_date": install_date + }) + + +# Read in static json, which is transformed by Jun-te's script +folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Waltham Forest/Decent Homes Pilot" +filenames = ["flat 1.json", "house 1.json"] + +houses_waltham_forest_data = pd.read_excel( + os.path.join(folder, "LBWF - Example Asset Data September 2025.xlsx"), + sheet_name="Houses Asset Data" +) +flats_waltham_forest_data = pd.read_excel( + os.path.join(folder, "LBWF - Example Asset Data September 2025.xlsx"), + sheet_name="CHINGFORD ROAD 236-254 Asset Bl" +) + +# Standardised variables which will form the enums in the db +HHSRS_VARIABLES = [ + "damp_and_mould_growth", + "excess_cold", + "excess_heat", + "asbestos_and_mm_fibres", + "biocides", + "carbon_monoxide_and_fuel_combustion_products", + "lead", + "radiation", + "uncombusted_fuel_gas", + "volatile_organic_compounds", + "crowding_and_space", + "entry_by_intruders", + "lighting", + "noise", + "domestic_hygiene_pests_and_refuse", + "food_safety", + "personal_hygiene_sanitation_and_drainage", + "water_supply", + "falls_associated_with_baths", + "falls_on_level_surfaces", + "falls_on_stairs_and_steps", + "falls_between_levels", + "electrical_hazards", + "fire", + "flames_hot_surfaces_and_materials", + "collision_and_entrapment", + "explosions", + "ergonomics", + "structural_collapse_and_falling_elements" +] + +CRITERION_B_VARIABLES = [ + "external_walls_structure", "lintels", "brickwork_spalling", "wall_finish", "roof_structure", "roof_finish", + "chimneys", "windows", "external_doors", "kitchens", "bathrooms", "central_heating_boiler", + "central_heating_distribution_system", "heating_other", "electrical_systems", +] + +CRITERION_C_VARIABLES = [ + "kitchen_facilities", +] + +# Criterion C explicit age limits (different from component lifespans used elsewhere) +CRITERION_C_AGE_LIMITS = { + "kitchen_years_max": 20, + "bathroom_years_max": 30, +} + +# Field labels as they appear in your JSON (based on your code) +LABEL_KITCHEN = "Adequacy of Kitchen and Type in Property" +LABEL_BATHROOM = "Adequacy of Bathroom Location in Property" +LABEL_NOISE = "Adequacy of Noise Insulation in Property" +LABEL_COMMON_CIRC = "Circulation Space in Common Area" # flats only + +STANDARD_HHSRS_MAPPING = {"pass": "TYPRISK", "fail": "MODRISK", "no_data": "TOBEASSESS"} + +# Criterion A - mapping of HHSRS variables to Waltham forest element codes +HHSRS_MAPPING = { + "damp_and_mould_growth": {"HHSRSDAMP": STANDARD_HHSRS_MAPPING}, + "excess_cold": {"HHSRSCOLD": STANDARD_HHSRS_MAPPING}, + "excess_heat": {"HHSRSHEAT": STANDARD_HHSRS_MAPPING}, + "asbestos_and_mm_fibres": {"HHSRSASB": STANDARD_HHSRS_MAPPING}, + "biocides": {"HHSRSBIOC": STANDARD_HHSRS_MAPPING}, + "carbon_monoxide_and_fuel_combustion_products": { + "HHSRSCO": STANDARD_HHSRS_MAPPING, + "HHSRSSO2": STANDARD_HHSRS_MAPPING, + "HHSRSNO2": STANDARD_HHSRS_MAPPING + }, + "lead": {"HHSRSLEAD": STANDARD_HHSRS_MAPPING}, + "radiation": {"HHSRSRADIA": STANDARD_HHSRS_MAPPING}, + "uncombusted_fuel_gas": {"HHSRSFUEL": STANDARD_HHSRS_MAPPING}, + "volatile_organic_compounds": {"HHSRSORGAN": STANDARD_HHSRS_MAPPING}, + "crowding_and_space": {"HHSRSCROWD": STANDARD_HHSRS_MAPPING}, + "entry_by_intruders": {"HHSRSENTRY": STANDARD_HHSRS_MAPPING}, + "lighting": {"HHSRSLIGHT": STANDARD_HHSRS_MAPPING}, + "noise": {"HHSRSNOISE": STANDARD_HHSRS_MAPPING}, + "domestic_hygiene_pests_and_refuse": {"HHSRSDOMES": STANDARD_HHSRS_MAPPING}, + "food_safety": {"HHSRSFOOD": STANDARD_HHSRS_MAPPING}, + "personal_hygiene_sanitation_and_drainage": {"HHSRSPERS": STANDARD_HHSRS_MAPPING}, + "water_supply": {"HHSRSWATER": STANDARD_HHSRS_MAPPING}, + "falls_associated_with_baths": {"HHSRSFBATH": STANDARD_HHSRS_MAPPING}, + "falls_on_level_surfaces": {"HHSRSFLEVE": STANDARD_HHSRS_MAPPING}, + "falls_on_stairs_and_steps": {"HHSRSFSTAI": STANDARD_HHSRS_MAPPING}, + "falls_between_levels": {"HHSRSFBETW": STANDARD_HHSRS_MAPPING}, + "electrical_hazards": {"HHSRSELEC": STANDARD_HHSRS_MAPPING}, + "fire": {"HHSRSFIRE": STANDARD_HHSRS_MAPPING}, + "flames_hot_surfaces_and_materials": {"HHSRSFLAME": STANDARD_HHSRS_MAPPING}, + "collision_and_entrapment": {"HHSRSENTRP": STANDARD_HHSRS_MAPPING, "HHSRSCLOW": STANDARD_HHSRS_MAPPING}, + "explosions": {"HHSRSEXPLO": STANDARD_HHSRS_MAPPING}, + "ergonomics": {"HHSRSPOSI": STANDARD_HHSRS_MAPPING}, + "structural_collapse_and_falling_elements": {"HHSRSSTRUC": STANDARD_HHSRS_MAPPING} +} + +print(houses_waltham_forest_data[ + houses_waltham_forest_data["ELEMENT CODE"] == "INTHTIMP" + ][["ATTRIBUTE CODE", "ATTRIBUTE CODE DESCRIPTION"]].drop_duplicates()) + +print(flats_waltham_forest_data[ + flats_waltham_forest_data["ELEMENT CODE"] == "INTBTHADEQ" + ][["ATTRIBUTE CODE", "ATTRIBUTE CODE DESCRIPTION"]].drop_duplicates()) + +# Criterion B +CRITERION_B_MAPPING = { + # TODO: Needs to be sorted!!! + # "external_walls_structure": { + # "EXTWALLSTR": {"pass": "GOOD", "fail": "POOR", "no_data": "Unknown if Structural Defects in External Area"} + # } + "lintels": { + "EXTLINTELS": {"pass": "GOOD", "fail": "POOR", "no_data": "Unknown Condition of Lintels"} + } +} + +# Criterion C +CRITERION_C_MAPPING = { + # "kitchen_less_than_20_years_old": +} + +COMPONENT_LIFESPANS = { + "kitchen": {"house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30}, + "bathroom": {"house": 50, "flat_below_6_storeys": 50, "flat_above_6_storeys": 50} +} + +# Database design +# creation_date, uprn, variable, result, hhsrs_score (optional, numeric), hhsrs_rank (A-J), install_date (for +# components which expire, e.g. kitchen) + +decent_homes = [] +# Use to capture criterion A, B, C and D. Should be: +# {"uprn": int, "creation_date": datetime, "criterion_a": bool, "criterion_b": bool, "criterion_c": bool, +# "criterion_d": bool, "decent_homes": bool"} +property_decent_homes = [] +for fn in filenames: + with open(os.path.join(folder, fn), "rb") as f: + data = json.load(f) + + from pprint import pprint + + pprint(data["elements"]) + + property_info = data["property_info"] + if property_info["PROP TYPE"] in ["HOU"]: + property_type = "house" + elif property_info["PROP TYPE"] == "FLA": + raise Exception("Implement distrinction between below and above 6 storeys") + property_type = "flat" + else: + raise NotImplementedError("Unknown property type") + + # Criterion A + for hhsrs_variable, mapping in HHSRS_MAPPING.items(): + element_code = list(mapping.keys())[0] + + # Find the data in the JSON within data["elements"] + check_pass = [] + for k, v in data["elements"].items(): + if v["ELEMENT CODE"] == element_code: + # We check the attribute code + # Check if pass + if v["ATTRIBUTE CODE"] == mapping[element_code]["pass"]: + result = "pass" + elif v["ATTRIBUTE CODE"] == mapping[element_code]["fail"]: + result = "fail" + elif v["ATTRIBUTE CODE"] == mapping[element_code]["no_data"]: + result = "no_data" + else: + raise ValueError("Unknown attribute code") + check_pass.append(result) + + # We check if we have a pass, fail or no_data + if all([x == "pass" for x in check_pass]): + hhsrs_result = "pass" + elif any([x == "fail" for x in check_pass]): + hhsrs_result = "fail" + elif any([x == "no_data" for x in check_pass]): + hhsrs_result = "no_data" + else: + raise NotImplementedError("Mixed results not implemented") + decent_homes.append( + {"variable": hhsrs_variable, 'result': hhsrs_result, "hhsrs_rank": None, "hhsrs_score": None, + "install_date": None} + ) + + # Criterion B + + # --- Criterion C --- + today = pd.Timestamp.today().normalize() + + # Guard: property type string already set earlier + is_flat = (property_info["PROP TYPE"] == "FLA") + + # 1) Kitchen age ≤ 20 years + kitchen = get_element(data["elements"], LABEL_KITCHEN) + if kitchen: + kit_install_raw = kitchen.get("INSTALL DATE") + try: + kit_install = pd.to_datetime(kit_install_raw) + kit_age_years = years_between(today.to_pydatetime(), kit_install.to_pydatetime()) + kitchen_age_result = "pass" if kit_age_years <= CRITERION_C_AGE_LIMITS["kitchen_years_max"] else "fail" + # For transparency, store next renewal as install + 20 years (criterion C perspective) + kit_next_due = kit_install + pd.DateOffset(years=CRITERION_C_AGE_LIMITS["kitchen_years_max"]) + except Exception: + kitchen_age_result = "no_data" + kit_next_due = None + else: + kitchen_age_result = "no_data" + kit_next_due = None + append_result(decent_homes, "kitchen_less_than_20_years_old", kitchen_age_result, kit_next_due) + + # 2) Kitchen adequate space/layout + # Prefer explicit codes if you have them, fall back to text in ATTRIBUTE CODE DESCRIPTION + if kitchen: + kit_attr_desc = kitchen.get("ATTRIBUTE CODE DESCRIPTION", "") + # If you prefer codes, you can also branch here on kitchen.get("ATTRIBUTE CODE") == "STDKITADQ" + kitchen_adequacy_result = adequacy_result_by_text(kit_attr_desc) + else: + kitchen_adequacy_result = "no_data" + append_result(decent_homes, "kitchen_adequate_space_and_layout", kitchen_adequacy_result) + + # 3) Bathroom age ≤ 30 years + bath = get_element(data["elements"], LABEL_BATHROOM) + if bath: + bth_install_raw = bath.get("INSTALL DATE") + try: + bth_install = pd.to_datetime(bth_install_raw) + bth_age_years = years_between(today.to_pydatetime(), bth_install.to_pydatetime()) + bathroom_age_result = "pass" if bth_age_years <= CRITERION_C_AGE_LIMITS["bathroom_years_max"] else "fail" + bth_next_due = bth_install + pd.DateOffset(years=CRITERION_C_AGE_LIMITS["bathroom_years_max"]) + except Exception: + bathroom_age_result = "no_data" + bth_next_due = None + else: + bathroom_age_result = "no_data" + bth_next_due = None + append_result(decent_homes, "bathroom_less_than_30_years_old", bathroom_age_result, bth_next_due) + + # 4) Bathroom/WC appropriately located + if bath: + # You already observed codes like STDBTHADQ / ADPBTHADQ as 'pass' + bth_attr_code = bath.get("ATTRIBUTE CODE", "") + bth_attr_desc = bath.get("ATTRIBUTE CODE DESCRIPTION", "") + known_pass_codes = {"STDBTHADQ", "ADPBTHADQ"} + if bth_attr_code in known_pass_codes: + bathroom_location_result = "pass" + else: + # Fallback to text adequacy check + bathroom_location_result = adequacy_result_by_text(bth_attr_desc) + else: + bathroom_location_result = "no_data" + append_result(decent_homes, "bathroom_wc_appropriately_located", bathroom_location_result) + + # 5) Adequate external noise insulation + noise = get_element(data["elements"], LABEL_NOISE) + if noise: + noise_desc = noise.get("ATTRIBUTE CODE DESCRIPTION", "") + noise_result = adequacy_result_by_text(noise_desc) + else: + noise_result = "no_data" + append_result(decent_homes, "adequate_external_noise_insulation", noise_result) + + # 6) Adequate common entrance areas (flats only) + if is_flat: + raise Exception("Pls check this") + common = get_element(data["elements"], LABEL_COMMON_CIRC) + if common: + circ_desc = common.get("ATTRIBUTE CODE DESCRIPTION", "") + common_areas_result = adequacy_result_by_text(circ_desc) + else: + common_areas_result = "no_data" + append_result(decent_homes, "adequate_common_entrance_areas", common_areas_result) + + # ---------------- Criterion D ---------------- + # heating system type + heating = get_element(data["elements"], "Heating Improvement Required in Property") + if heating: + # Example: ATTRIBUTE CODE == "GOOD" means pass, "POOR" means fail + heat_type_code = heating.get("ATTRIBUTE CODE", "") + if heat_type_code in {"NOTAPPLIC"}: + heating_type_result = "pass" + elif heat_type_code in {"WETINSFULL"}: + heating_type_result = "fail" + else: + raise NotImplementedError("No other observed codes yet") + else: + raise NotImplementedError("Heating element missing in dataset") + + append_result(decent_homes, "efficient_heating_system_type", heating_type_result) + + # heating distribution + heating_dist = get_element(data["elements"], "Heating Distribution System in Property") + if heating_dist: + dist_desc = heating_dist.get("ATTRIBUTE CODE DESCRIPTION", "") + heating_dist_result = adequacy_result_by_text(dist_desc) + else: + raise NotImplementedError("Heating distribution element missing in dataset") + + append_result(decent_homes, "efficient_heating_distribution", heating_dist_result) + + # insulation + loft = get_element(data["elements"], "Size in mm of Loft Insulation Thickness in Property") + wall = get_element(data["elements"], "Wall Insulation Improvement in External Area") + heating = get_element(data["elements"], "Heating Improvement Required in Property") + # To determine how much loft insulation is required + + # Loft insulation check (example threshold: ≥ 270mm = pass) + if loft: + # We have a specific code, where further loft insulation is needed + loft_code = loft.get("ATTRIBUTE CODE", "") + if loft_code == "LOFTINSRQD": + loft_result = "fail" + elif loft_code.isnumeric(): + loft_result = "pass" + else: + raise NotImplementedError("Unknown loft insulation code - pls check") + else: + raise NotImplementedError("Loft insulation data missing - pls check") + append_result(decent_homes, "loft_insulation_sufficient", loft_result) + + # Wall insulation check (simple adequacy parser) + if wall: + wall_desc = wall.get("ATTRIBUTE CODE DESCRIPTION", "") + wall_result = adequacy_result_by_text(wall_desc) + else: + raise NotImplementedError("Wall insulation data missing - pls check") + append_result(decent_homes, "wall_insulation_sufficient", wall_result) + + # ---------------- Criterion A overall ---------------- + a_vars = set(HHSRS_MAPPING.keys()) + latest_a_results = {r["variable"]: r["result"] for r in decent_homes if r["variable"] in a_vars} + + if any(v == "fail" for v in latest_a_results.values()): + criterion_a_result = "fail" + elif all(v == "pass" for v in latest_a_results.values()): + criterion_a_result = "pass" + else: + criterion_a_result = "no_data" + + # ---------------- Criterion C overall ---------------- + criterion_c_vars = [ + "kitchen_less_than_20_years_old", + "kitchen_adequate_space_and_layout", + "bathroom_less_than_30_years_old", + "bathroom_wc_appropriately_located", + "adequate_external_noise_insulation", + ] + if is_flat: + criterion_c_vars.append("adequate_common_entrance_areas") + + latest_c_results = {r["variable"]: r["result"] for r in decent_homes if r["variable"] in criterion_c_vars} + + count_fails = sum(1 for v in latest_c_results.values() if v == "fail") + # optionally count no_data too if you want strict interpretation + criterion_c_result = "fail" if count_fails >= 3 else "pass" + + # ---------------- Criterion D overall ---------------- + criterion_d_vars = [ + "efficient_heating_system_type", + "efficient_heating_distribution", + "loft_insulation_sufficient", + "wall_insulation_sufficient", + ] + latest_d_results = {r["variable"]: r["result"] for r in decent_homes if r["variable"] in criterion_d_vars} + + if any(v == "fail" for v in latest_d_results.values()): + criterion_d_result = "fail" + elif all(v == "pass" for v in latest_d_results.values()): + criterion_d_result = "pass" + else: + criterion_d_result = "no_data" + + # ---------------- Append to property_decent_homes ---------------- + property_decent_homes.append({ + "uprn": property_info.get("UPRN"), # update field name if needed + "creation_date": datetime.now().date().isoformat(), + "criterion_a": criterion_a_result, + "criterion_b": None, # not yet implemented + "criterion_c": criterion_c_result, + "criterion_d": criterion_d_result, + "decent_homes": ( + criterion_a_result == "pass" + and criterion_c_result == "pass" + ) + }) From a5ae1669718ac1fd6b17fba13678920534d17ea1 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Mon, 22 Sep 2025 12:32:51 +0100 Subject: [PATCH 3/7] mvp implementation for A, C, D, year mapping for B --- .../waltham_forest/decent_homes_pilot.py | 367 ++++++++++++------ 1 file changed, 254 insertions(+), 113 deletions(-) diff --git a/etl/customers/waltham_forest/decent_homes_pilot.py b/etl/customers/waltham_forest/decent_homes_pilot.py index 78460f5a..b59168fb 100644 --- a/etl/customers/waltham_forest/decent_homes_pilot.py +++ b/etl/customers/waltham_forest/decent_homes_pilot.py @@ -16,26 +16,8 @@ def get_element(elements, label): return elements.get(label) -def adequacy_result_by_text(attr_desc: str): - """ - Generic adequacy parser. - Pass if description clearly says 'Adequate' and not 'Inadequate'. - Fail if it says 'Inadequate' (or equivalent). - Unknown -> 'no_data' - """ - if not attr_desc or not isinstance(attr_desc, str): - return "no_data" - text = attr_desc.strip().lower() - # Common patterns - if "inadequate" in text or "unsatisfactory" in text or "problems" in text: - return "fail" - if "adequate" in text or "standard" in text or "appropriate" in text: - return "pass" - return "no_data" - - -def append_result(decent_homes, variable, result, install_date=None): - decent_homes.append({ +def append_result(decent_homes_meta, variable, result, install_date=None): + decent_homes_meta.append({ "variable": variable, "result": result, "hhsrs_rank": None, @@ -97,7 +79,8 @@ CRITERION_B_VARIABLES = [ ] CRITERION_C_VARIABLES = [ - "kitchen_facilities", + "kitchen_less_than_20_years_old", "kitchen_adequate_space_and_layout", "bathroom_less_than_30_years_old", + "bathroom_wc_appropriately_located", "adequate_external_noise_insulation", "adequate_common_entrance_areas", ] # Criterion C explicit age limits (different from component lifespans used elsewhere) @@ -151,40 +134,163 @@ HHSRS_MAPPING = { "structural_collapse_and_falling_elements": {"HHSRSSTRUC": STANDARD_HHSRS_MAPPING} } -print(houses_waltham_forest_data[ - houses_waltham_forest_data["ELEMENT CODE"] == "INTHTIMP" - ][["ATTRIBUTE CODE", "ATTRIBUTE CODE DESCRIPTION"]].drop_duplicates()) +# print(houses_waltham_forest_data[ +# houses_waltham_forest_data["ELEMENT CODE"] == "INTBTHADEQ" +# ][["ATTRIBUTE CODE", "ATTRIBUTE CODE DESCRIPTION"]].drop_duplicates()) + +# print(flats_waltham_forest_data[ +# flats_waltham_forest_data["ELEMENT CODE"] == "INTBTHADEQ" +# ][["ATTRIBUTE CODE", "ATTRIBUTE CODE DESCRIPTION"]].drop_duplicates()) -print(flats_waltham_forest_data[ - flats_waltham_forest_data["ELEMENT CODE"] == "INTBTHADEQ" - ][["ATTRIBUTE CODE", "ATTRIBUTE CODE DESCRIPTION"]].drop_duplicates()) # Criterion B -CRITERION_B_MAPPING = { - # TODO: Needs to be sorted!!! - # "external_walls_structure": { - # "EXTWALLSTR": {"pass": "GOOD", "fail": "POOR", "no_data": "Unknown if Structural Defects in External Area"} - # } - "lintels": { - "EXTLINTELS": {"pass": "GOOD", "fail": "POOR", "no_data": "Unknown Condition of Lintels"} - } +B_COMPONENT_LABELS = { + # Key components + "wall_structure": [ + "Wall Structure in External Area", + ], + "lintels": [ + "Lintels in External Area", + ], + "brickwork_spalling": [ + "Wall Spalling in External Area", + ], + "wall_finish": [ + "Wall Finish 1 in External Area", + "Wall Finish 2 in External Area", + "External Decorations in External Area", + "Brickwork Pointing in External Area", + ], + "roof_structure": [ + "Roof Structure 1 in External Area", + "Roof Structure 2 in External Area", + "Roof Structure 3 in External Area", + # If you later decide to include ancillary items, add: + # "Fascia / Soffit / Bargeboard in External Area", + # "Gutters in External Area", "Downpipes in External Area", + # "Internal Downpipes in External Area", + # and give them a clear condition rule. + ], + "roof_finish": [ + "Roof Covering 1 in External Area", + "Roof Covering 2 in External Area", + "Roof Covering 3 in External Area", + ], + "chimneys": [ + "Chimneys in External Area", + ], + "windows": [ + "Windows in Property", + "Windows 1 in External Area", + "Windows 2 in External Area", + "Garage and Store Windows in External Area", + "Garage Windows in External Area", + "Store Windows in External Area", + ], + "external_doors": [ + "Type and Location of Front Door in Property", + "Front Door Fire Rating in Property", + "Patio and French Doors 1 in External Area", + "Back and Side Doors 1 in External Area", + "Back and Side Doors 2 in External Area", + "Garage and Store Doors in External Area", + "Garage Door in External Area", + "Store Door in External Area", + ], + "central_heating_boiler": [ + # If the dataset exposes a specific boiler element, put it here. + # For now we only have "Heating Improvement Required in Property" elsewhere (Criterion D), + # which isn't reliable for age. If your JSON later includes a boiler line with INSTALL DATE, + # add its label here. + ], + "heating_other": [ + # e.g., gas fires/storage heaters if present as discrete elements later. + ], + "electrical_systems": [ + # If you have an installation line with dates (e.g. "Electrics Required in Property") + # add it here; we will rely on INSTALL DATE + REMAINING LIFE. + "Electrics Required in Property", + ], + + # Other components + "kitchen": [ + "Adequacy of Kitchen and Type in Property", + ], + "bathroom": [ + "Adequacy of Bathroom Location in Property", + ], + "central_heating_distribution_system": [ + "Heating Distribution System in Property", + ], +} + +KEY_COMPONENTS = { + "wall_structure", "lintels", "brickwork_spalling", "wall_finish", + "roof_structure", "roof_finish", "chimneys", "windows", + "external_doors", "central_heating_boiler", "heating_other", + "electrical_systems", +} +OTHER_COMPONENTS = { + "kitchen", "bathroom", "central_heating_distribution_system", } # Criterion C -CRITERION_C_MAPPING = { - # "kitchen_less_than_20_years_old": -} - COMPONENT_LIFESPANS = { - "kitchen": {"house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30}, - "bathroom": {"house": 50, "flat_below_6_storeys": 50, "flat_above_6_storeys": 50} + # Key components + "wall_structure": { + "house": 80, "flat_below_6_storeys": 80, "flat_above_6_storeys": 80 + }, + "lintels": { + "house": 60, "flat_below_6_storeys": 60, "flat_above_6_storeys": 60 + }, + "brickwork_spalling": { + "house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30 + }, + "wall_finish": { + "house": 60, "flat_below_6_storeys": 60, "flat_above_6_storeys": 30 + }, + "roof_structure": { + "house": 50, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30 + }, + "roof_finish": { + "house": 50, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30 + }, + "chimneys": { + "house": 50, "flat_below_6_storeys": 50, "flat_above_6_storeys": None # N/A + }, + "windows": { + "house": 40, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30 + }, + "external_doors": { + "house": 40, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30 + }, + "central_heating_boiler": { + "house": 15, "flat_below_6_storeys": 15, "flat_above_6_storeys": 15 + }, + "heating_other": { + "house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30 + }, + "electrical_systems": { + "house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30 + }, + + # Other components + "kitchen": { + "house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30 + }, + "bathroom": { + "house": 40, "flat_below_6_storeys": 40, "flat_above_6_storeys": 40 + }, + "central_heating_distribution_system": { + "house": 40, "flat_below_6_storeys": 40, "flat_above_6_storeys": 40 + }, } # Database design # creation_date, uprn, variable, result, hhsrs_score (optional, numeric), hhsrs_rank (A-J), install_date (for # components which expire, e.g. kitchen) -decent_homes = [] +decent_homes_meta = [] # Use to capture criterion A, B, C and D. Should be: # {"uprn": int, "creation_date": datetime, "criterion_a": bool, "criterion_b": bool, "criterion_c": bool, # "criterion_d": bool, "decent_homes": bool"} @@ -193,20 +299,16 @@ for fn in filenames: with open(os.path.join(folder, fn), "rb") as f: data = json.load(f) - from pprint import pprint - - pprint(data["elements"]) - property_info = data["property_info"] if property_info["PROP TYPE"] in ["HOU"]: property_type = "house" elif property_info["PROP TYPE"] == "FLA": raise Exception("Implement distrinction between below and above 6 storeys") - property_type = "flat" + # property_type = "flat" else: raise NotImplementedError("Unknown property type") - # Criterion A + # ---------------- Criterion A ---------------- for hhsrs_variable, mapping in HHSRS_MAPPING.items(): element_code = list(mapping.keys())[0] @@ -235,14 +337,48 @@ for fn in filenames: hhsrs_result = "no_data" else: raise NotImplementedError("Mixed results not implemented") - decent_homes.append( + decent_homes_meta.append( {"variable": hhsrs_variable, 'result': hhsrs_result, "hhsrs_rank": None, "hhsrs_score": None, "install_date": None} ) - # Criterion B + # ---------------- Criterion B ---------------- + # Check each of the components - # --- Criterion C --- + component_pass_or_fail = [] + # TODO: Delete me + component, labels = list(B_COMPONENT_LABELS.items())[1] + for component, labels in B_COMPONENT_LABELS.items(): + # TODO: labels may not need to be multiple variables + for label in labels: + # Grab the label + label_data = get_element(data["elements"], label) + # 1) We check if the component is old + install_date = pd.to_datetime(label_data["INSTALL DATE"]) + if pd.isnull(install_date): + raise ValueError("Missing install date - pls check") + component_lifetime = COMPONENT_LIFESPANS[component][property_type] + # This should be populated, and for the pilot it's okay if this errors if missing - we'll handle accordingly + is_old = years_between(today.to_pydatetime(), install_date.to_pydatetime()) >= component_lifetime + # 2) We check if the component is in poor condition + if pd.isnull(label_data["REMAINING LIFE"]): + raise ValueError("Missing remaining life - pls check") + has_failed = label_data["REMAINING LIFE"] < 0 + # The component needs to have both failed and be old to fail criterion B + component_result = "fail" if is_old and has_failed else "pass" + component_pass_or_fail.append( + { + "component": component, + "label": label, + "install_date": str(install_date), + "remaining_life": label_data["REMAINING LIFE"], + "is_old": is_old, + "has_failed": has_failed, + "result": component_result + } + ) + + # ---------------- Criterion C ---------------- today = pd.Timestamp.today().normalize() # Guard: property type string already set earlier @@ -251,71 +387,67 @@ for fn in filenames: # 1) Kitchen age ≤ 20 years kitchen = get_element(data["elements"], LABEL_KITCHEN) if kitchen: - kit_install_raw = kitchen.get("INSTALL DATE") - try: - kit_install = pd.to_datetime(kit_install_raw) - kit_age_years = years_between(today.to_pydatetime(), kit_install.to_pydatetime()) - kitchen_age_result = "pass" if kit_age_years <= CRITERION_C_AGE_LIMITS["kitchen_years_max"] else "fail" - # For transparency, store next renewal as install + 20 years (criterion C perspective) - kit_next_due = kit_install + pd.DateOffset(years=CRITERION_C_AGE_LIMITS["kitchen_years_max"]) - except Exception: - kitchen_age_result = "no_data" - kit_next_due = None + kit_install_raw = kitchen["INSTALL DATE"] + kit_install = pd.to_datetime(kit_install_raw) + kit_age_years = years_between(today.to_pydatetime(), kit_install.to_pydatetime()) + kitchen_age_result = "pass" if kit_age_years <= CRITERION_C_AGE_LIMITS["kitchen_years_max"] else "fail" + # For transparency, store next renewal as install + 20 years (criterion C perspective) + kit_next_due = kit_install + pd.DateOffset(years=CRITERION_C_AGE_LIMITS["kitchen_years_max"]) else: - kitchen_age_result = "no_data" - kit_next_due = None - append_result(decent_homes, "kitchen_less_than_20_years_old", kitchen_age_result, kit_next_due) + raise NotImplementedError("Kitchen data missing - pls check") + append_result( + decent_homes_meta, "kitchen_less_than_20_years_old", kitchen_age_result, install_date=str(kit_install) + ) # 2) Kitchen adequate space/layout # Prefer explicit codes if you have them, fall back to text in ATTRIBUTE CODE DESCRIPTION if kitchen: - kit_attr_desc = kitchen.get("ATTRIBUTE CODE DESCRIPTION", "") - # If you prefer codes, you can also branch here on kitchen.get("ATTRIBUTE CODE") == "STDKITADQ" - kitchen_adequacy_result = adequacy_result_by_text(kit_attr_desc) + kit_attr_desc = kitchen["ATTRIBUTE CODE"] + if kit_attr_desc == "STDKITADQ": + kitchen_adequacy_result = "pass" + else: + raise NotImplementedError("No other observed codes yet") else: - kitchen_adequacy_result = "no_data" - append_result(decent_homes, "kitchen_adequate_space_and_layout", kitchen_adequacy_result) + raise NotImplementedError("Kitchen data missing - pls check") + append_result(decent_homes_meta, "kitchen_adequate_space_and_layout", kitchen_adequacy_result) # 3) Bathroom age ≤ 30 years bath = get_element(data["elements"], LABEL_BATHROOM) if bath: - bth_install_raw = bath.get("INSTALL DATE") - try: - bth_install = pd.to_datetime(bth_install_raw) - bth_age_years = years_between(today.to_pydatetime(), bth_install.to_pydatetime()) - bathroom_age_result = "pass" if bth_age_years <= CRITERION_C_AGE_LIMITS["bathroom_years_max"] else "fail" - bth_next_due = bth_install + pd.DateOffset(years=CRITERION_C_AGE_LIMITS["bathroom_years_max"]) - except Exception: - bathroom_age_result = "no_data" - bth_next_due = None + bth_install_raw = bath["INSTALL DATE"] + bth_install = pd.to_datetime(bth_install_raw) + bth_age_years = years_between(today.to_pydatetime(), bth_install.to_pydatetime()) + bathroom_age_result = "pass" if bth_age_years <= CRITERION_C_AGE_LIMITS["bathroom_years_max"] else "fail" + bth_next_due = bth_install + pd.DateOffset(years=CRITERION_C_AGE_LIMITS["bathroom_years_max"]) else: - bathroom_age_result = "no_data" - bth_next_due = None - append_result(decent_homes, "bathroom_less_than_30_years_old", bathroom_age_result, bth_next_due) + raise NotImplementedError("Bathroom data missing - pls check") + append_result( + decent_homes_meta, "bathroom_less_than_30_years_old", bathroom_age_result, install_date=str(bth_install) + ) # 4) Bathroom/WC appropriately located if bath: - # You already observed codes like STDBTHADQ / ADPBTHADQ as 'pass' - bth_attr_code = bath.get("ATTRIBUTE CODE", "") - bth_attr_desc = bath.get("ATTRIBUTE CODE DESCRIPTION", "") - known_pass_codes = {"STDBTHADQ", "ADPBTHADQ"} - if bth_attr_code in known_pass_codes: + bth_attr_code = bath["ATTRIBUTE CODE"] + if bth_attr_code in {"STDBTHADQ", "ADPBTHADQ"}: bathroom_location_result = "pass" else: - # Fallback to text adequacy check - bathroom_location_result = adequacy_result_by_text(bth_attr_desc) + raise NotImplementedError("No other observed codes yet") else: - bathroom_location_result = "no_data" - append_result(decent_homes, "bathroom_wc_appropriately_located", bathroom_location_result) + raise NotImplementedError("Bathroom data missing - pls check") + + append_result(decent_homes_meta, "bathroom_wc_appropriately_located", bathroom_location_result) # 5) Adequate external noise insulation noise = get_element(data["elements"], LABEL_NOISE) if noise: - noise_desc = noise.get("ATTRIBUTE CODE DESCRIPTION", "") - noise_result = adequacy_result_by_text(noise_desc) + noise_code = noise["ATTRIBUTE CODE"] + if noise_code in {"ADEQUATE"}: + noise_result = "pass" + else: + raise NotImplementedError("No other observed codes yet") else: - noise_result = "no_data" - append_result(decent_homes, "adequate_external_noise_insulation", noise_result) + raise NotImplementedError("Noise insulation data missing - pls check") + append_result(decent_homes_meta, "adequate_external_noise_insulation", noise_result) # 6) Adequate common entrance areas (flats only) if is_flat: @@ -326,14 +458,13 @@ for fn in filenames: common_areas_result = adequacy_result_by_text(circ_desc) else: common_areas_result = "no_data" - append_result(decent_homes, "adequate_common_entrance_areas", common_areas_result) + append_result(decent_homes_meta, "adequate_common_entrance_areas", common_areas_result) # ---------------- Criterion D ---------------- # heating system type heating = get_element(data["elements"], "Heating Improvement Required in Property") if heating: - # Example: ATTRIBUTE CODE == "GOOD" means pass, "POOR" means fail - heat_type_code = heating.get("ATTRIBUTE CODE", "") + heat_type_code = heating["ATTRIBUTE CODE"] if heat_type_code in {"NOTAPPLIC"}: heating_type_result = "pass" elif heat_type_code in {"WETINSFULL"}: @@ -343,28 +474,33 @@ for fn in filenames: else: raise NotImplementedError("Heating element missing in dataset") - append_result(decent_homes, "efficient_heating_system_type", heating_type_result) + append_result(decent_homes_meta, "efficient_heating_system_type", heating_type_result) # heating distribution heating_dist = get_element(data["elements"], "Heating Distribution System in Property") if heating_dist: - dist_desc = heating_dist.get("ATTRIBUTE CODE DESCRIPTION", "") - heating_dist_result = adequacy_result_by_text(dist_desc) + dist_code = heating_dist["ATTRIBUTE CODE"] + if dist_code == "UNKNOWN": + # For the observed case, there was no heating and wet heating needed to be installed in full so the value + # was unknown + heating_dist_result = "no_data" + else: + raise NotImplementedError("No other observed codes yet") else: raise NotImplementedError("Heating distribution element missing in dataset") - append_result(decent_homes, "efficient_heating_distribution", heating_dist_result) + append_result(decent_homes_meta, "efficient_heating_distribution", heating_dist_result) # insulation loft = get_element(data["elements"], "Size in mm of Loft Insulation Thickness in Property") wall = get_element(data["elements"], "Wall Insulation Improvement in External Area") - heating = get_element(data["elements"], "Heating Improvement Required in Property") # To determine how much loft insulation is required # Loft insulation check (example threshold: ≥ 270mm = pass) if loft: - # We have a specific code, where further loft insulation is needed - loft_code = loft.get("ATTRIBUTE CODE", "") + # We have a specific code, where further loft insulation is needed - It appears the heating type check has + # already been completed in this dataset and so we just need to check the code + loft_code = loft["ATTRIBUTE CODE"] if loft_code == "LOFTINSRQD": loft_result = "fail" elif loft_code.isnumeric(): @@ -373,19 +509,22 @@ for fn in filenames: raise NotImplementedError("Unknown loft insulation code - pls check") else: raise NotImplementedError("Loft insulation data missing - pls check") - append_result(decent_homes, "loft_insulation_sufficient", loft_result) + append_result(decent_homes_meta, "loft_insulation_sufficient", loft_result) - # Wall insulation check (simple adequacy parser) + # Wall insulation check if wall: - wall_desc = wall.get("ATTRIBUTE CODE DESCRIPTION", "") - wall_result = adequacy_result_by_text(wall_desc) + wall_code = wall["ATTRIBUTE CODE"] + if wall_code in {"NONE"}: # Means no insulation improvement required + wall_result = "pass" + else: + raise NotImplementedError("No other observed codes yet") else: raise NotImplementedError("Wall insulation data missing - pls check") - append_result(decent_homes, "wall_insulation_sufficient", wall_result) + append_result(decent_homes_meta, "wall_insulation_sufficient", wall_result) # ---------------- Criterion A overall ---------------- a_vars = set(HHSRS_MAPPING.keys()) - latest_a_results = {r["variable"]: r["result"] for r in decent_homes if r["variable"] in a_vars} + latest_a_results = {r["variable"]: r["result"] for r in decent_homes_meta if r["variable"] in a_vars} if any(v == "fail" for v in latest_a_results.values()): criterion_a_result = "fail" @@ -405,20 +544,21 @@ for fn in filenames: if is_flat: criterion_c_vars.append("adequate_common_entrance_areas") - latest_c_results = {r["variable"]: r["result"] for r in decent_homes if r["variable"] in criterion_c_vars} + latest_c_results = {r["variable"]: r["result"] for r in decent_homes_meta if r["variable"] in criterion_c_vars} count_fails = sum(1 for v in latest_c_results.values() if v == "fail") # optionally count no_data too if you want strict interpretation criterion_c_result = "fail" if count_fails >= 3 else "pass" # ---------------- Criterion D overall ---------------- + # Needs to have both efficient geating and distribution so all should pass criterion_d_vars = [ "efficient_heating_system_type", "efficient_heating_distribution", "loft_insulation_sufficient", "wall_insulation_sufficient", ] - latest_d_results = {r["variable"]: r["result"] for r in decent_homes if r["variable"] in criterion_d_vars} + latest_d_results = {r["variable"]: r["result"] for r in decent_homes_meta if r["variable"] in criterion_d_vars} if any(v == "fail" for v in latest_d_results.values()): criterion_d_result = "fail" @@ -429,7 +569,7 @@ for fn in filenames: # ---------------- Append to property_decent_homes ---------------- property_decent_homes.append({ - "uprn": property_info.get("UPRN"), # update field name if needed + "uprn": property_info.get("UPRN"), # TODO: Need UPRN "creation_date": datetime.now().date().isoformat(), "criterion_a": criterion_a_result, "criterion_b": None, # not yet implemented @@ -438,5 +578,6 @@ for fn in filenames: "decent_homes": ( criterion_a_result == "pass" and criterion_c_result == "pass" + and criterion_d_result == "pass" ) }) From d68ef88b9db7735a55a74732a58dabe5f3ff8463 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Mon, 22 Sep 2025 14:57:58 +0100 Subject: [PATCH 4/7] wip --- .../waltham_forest/decent_homes_pilot.py | 107 ++++++++++++++---- 1 file changed, 85 insertions(+), 22 deletions(-) diff --git a/etl/customers/waltham_forest/decent_homes_pilot.py b/etl/customers/waltham_forest/decent_homes_pilot.py index b59168fb..ba9bb3b7 100644 --- a/etl/customers/waltham_forest/decent_homes_pilot.py +++ b/etl/customers/waltham_forest/decent_homes_pilot.py @@ -5,6 +5,8 @@ import pandas as pd from datetime import datetime +from docutils.nodes import table + def years_between(d1, d2): # precise year difference (accounts for months/days) @@ -16,13 +18,14 @@ def get_element(elements, label): return elements.get(label) -def append_result(decent_homes_meta, variable, result, install_date=None): +def append_result(decent_homes_meta, variable, result, install_date=None, expiry_date=None): decent_homes_meta.append({ "variable": variable, "result": result, "hhsrs_rank": None, "hhsrs_score": None, - "install_date": install_date + "install_date": install_date, + "expiry_date": expiry_date, }) @@ -165,11 +168,13 @@ B_COMPONENT_LABELS = { "Roof Structure 1 in External Area", "Roof Structure 2 in External Area", "Roof Structure 3 in External Area", - # If you later decide to include ancillary items, add: - # "Fascia / Soffit / Bargeboard in External Area", - # "Gutters in External Area", "Downpipes in External Area", - # "Internal Downpipes in External Area", - # and give them a clear condition rule. + "Garage Roof in External Area", + "Garage and Store Roofs in External Area", + "Store Roof in External Area", + "Fascia / Soffit / Bargeboard in External Area", + "Gutters in External Area", + "Downpipes in External Area", + "Internal Downpipes in External Area" ], "roof_finish": [ "Roof Covering 1 in External Area", @@ -198,20 +203,15 @@ B_COMPONENT_LABELS = { "Store Door in External Area", ], "central_heating_boiler": [ - # If the dataset exposes a specific boiler element, put it here. - # For now we only have "Heating Improvement Required in Property" elsewhere (Criterion D), - # which isn't reliable for age. If your JSON later includes a boiler line with INSTALL DATE, - # add its label here. + # TODO ], "heating_other": [ - # e.g., gas fires/storage heaters if present as discrete elements later. + # TODO ], "electrical_systems": [ - # If you have an installation line with dates (e.g. "Electrics Required in Property") - # add it here; we will rely on INSTALL DATE + REMAINING LIFE. + # TODO "Electrics Required in Property", ], - # Other components "kitchen": [ "Adequacy of Kitchen and Type in Property", @@ -287,9 +287,10 @@ COMPONENT_LIFESPANS = { } # Database design -# creation_date, uprn, variable, result, hhsrs_score (optional, numeric), hhsrs_rank (A-J), install_date (for -# components which expire, e.g. kitchen) +# creation_date, uprn, variable, result (pass/fail/nodata), hhsrs_score (optional, numeric), hhsrs_rank (A-J), +# install_date (for components which expire, e.g. kitchen), remaining_life (for components which expire, e.g. kitchen), +# TODO: Add the criterion decent_homes_meta = [] # Use to capture criterion A, B, C and D. Should be: # {"uprn": int, "creation_date": datetime, "criterion_a": bool, "criterion_b": bool, "criterion_c": bool, @@ -303,12 +304,15 @@ for fn in filenames: if property_info["PROP TYPE"] in ["HOU"]: property_type = "house" elif property_info["PROP TYPE"] == "FLA": - raise Exception("Implement distrinction between below and above 6 storeys") + raise NotImplementedError("Implement distrinction between below and above 6 storeys") # property_type = "flat" else: raise NotImplementedError("Unknown property type") # ---------------- Criterion A ---------------- + # TODO: Map out the sub-information + # Critrion A: pass/fail + # If fail, why? for hhsrs_variable, mapping in HHSRS_MAPPING.items(): element_code = list(mapping.keys())[0] @@ -347,19 +351,36 @@ for fn in filenames: component_pass_or_fail = [] # TODO: Delete me - component, labels = list(B_COMPONENT_LABELS.items())[1] + component, labels = list(B_COMPONENT_LABELS.items())[9] + label = labels[0] + # TODO: need to handle the case where there is no survey data at all for a component for component, labels in B_COMPONENT_LABELS.items(): # TODO: labels may not need to be multiple variables for label in labels: # Grab the label label_data = get_element(data["elements"], label) + if label_data["ATTRIBUTE CODE"] in ["UNKNOWN", "NONE", "UNKNOWNG", "UNKNOWNS"]: + # This isn't applicable + component_pass_or_fail.append( + { + "component": component, + "label": label, + "install_date": None, + "remaining_life": None, + "is_old": False, + "has_failed": False, + "result": "pass", + "appliable": False + } + ) + continue # 1) We check if the component is old install_date = pd.to_datetime(label_data["INSTALL DATE"]) if pd.isnull(install_date): raise ValueError("Missing install date - pls check") component_lifetime = COMPONENT_LIFESPANS[component][property_type] # This should be populated, and for the pilot it's okay if this errors if missing - we'll handle accordingly - is_old = years_between(today.to_pydatetime(), install_date.to_pydatetime()) >= component_lifetime + is_old = years_between(today.to_pydatetime(), install_date.to_pydatetime()) > component_lifetime # 2) We check if the component is in poor condition if pd.isnull(label_data["REMAINING LIFE"]): raise ValueError("Missing remaining life - pls check") @@ -369,15 +390,54 @@ for fn in filenames: component_pass_or_fail.append( { "component": component, + "component_type": "key" if component in KEY_COMPONENTS else "other", + "component_sub_description": label_data["ATTRIBUTE CODE DESCRIPTION"], "label": label, "install_date": str(install_date), "remaining_life": label_data["REMAINING LIFE"], "is_old": is_old, "has_failed": has_failed, - "result": component_result + "result": component_result, + "appliable": True } ) + # TODO: We need to check by component + # Example of a pass for a component + # [ + # {"component": "external_walls", "component_type": "key", "descr": "A", "result": "pass"}, + # {"component": "external_walls", "component_type": "key", "descr": "B", "result": "pass"}, + # {"component": "external_walls", "component_type": "key", "descr": "C", "result": "pass"}, + # ] + + # Example of a fail for a component + # [ + # {"component": "external_walls", "component_type": "key", "descr": "A", "result": "pass"}, + # {"component": "external_walls", "component_type": "key", "descr": "B", "result": "fail"}, + # {"component": "external_walls", "component_type": "key", "descr": "C", "result": "pass"}, + # ] + + # Example of a no data for a component + # [ + # {"component": "external_walls", "component_type": "key", "descr": "A", "result": "pass"}, + # {"component": "external_walls", "component_type": "key", "descr": "B", "result": "nodata", "appliable": True}, + # {"component": "external_walls", "component_type": "key", "descr": "C", "result": "pass"}, + # ] + # OR + # Everything is unknown + # [ + # {"component": "external_walls", "component_type": "key", "descr": "A", "result": "pass", "appliable": False}, + # {"component": "external_walls", "component_type": "key", "descr": "B", "result": "pass", "appliable": False}, + # {"component": "external_walls", "component_type": "key", "descr": "C", "result": "pass", "appliable": False}, + # ] + + # Component 1: pass/fail, key: true/False + # Component 2: pass/fail, key: true/False + # Component 3: pass/fail, key: true/False + # Component 4: pass/fail, key: true/False + # Component 4: pass/fail, key: true/False + # -> Decide on outcome. If failure of 1 key component -> fail criterion B, or 2 other components -> fail criterion B + # ---------------- Criterion C ---------------- today = pd.Timestamp.today().normalize() @@ -396,7 +456,8 @@ for fn in filenames: else: raise NotImplementedError("Kitchen data missing - pls check") append_result( - decent_homes_meta, "kitchen_less_than_20_years_old", kitchen_age_result, install_date=str(kit_install) + decent_homes_meta, "kitchen_less_than_20_years_old", kitchen_age_result, + install_date=str(kit_install), expiry_date=str(kit_next_due) ) # 2) Kitchen adequate space/layout @@ -533,6 +594,8 @@ for fn in filenames: else: criterion_a_result = "no_data" + # ---------------- Criterion B overall ---------------- + # ---------------- Criterion C overall ---------------- criterion_c_vars = [ "kitchen_less_than_20_years_old", From a22db51be9ffedd782c38f4120b4cb5e26e23919 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Mon, 22 Sep 2025 22:33:31 +0100 Subject: [PATCH 5/7] pilot implementation --- .../waltham_forest/decent_homes_pilot.py | 319 ++++++++++++------ 1 file changed, 215 insertions(+), 104 deletions(-) diff --git a/etl/customers/waltham_forest/decent_homes_pilot.py b/etl/customers/waltham_forest/decent_homes_pilot.py index ba9bb3b7..33836236 100644 --- a/etl/customers/waltham_forest/decent_homes_pilot.py +++ b/etl/customers/waltham_forest/decent_homes_pilot.py @@ -18,9 +18,11 @@ def get_element(elements, label): return elements.get(label) -def append_result(decent_homes_meta, variable, result, install_date=None, expiry_date=None): +def append_result(decent_homes_meta, criteria, variable, sub_variable, result, install_date=None, expiry_date=None): decent_homes_meta.append({ + "criteria": criteria, "variable": variable, + "sub_variable": sub_variable, "result": result, "hhsrs_rank": None, "hhsrs_score": None, @@ -75,6 +77,44 @@ HHSRS_VARIABLES = [ "structural_collapse_and_falling_elements" ] +ELEMENT_CODE_TO_DESCRIPTION = { + # One-to-one + "HHSRSDAMP": "damp_and_mould_growth", + "HHSRSCOLD": "excess_cold", + "HHSRSHEAT": "excess_heat", + "HHSRSASB": "asbestos_and_mm_fibres", + "HHSRSBIOC": "biocides", + "HHSRSLEAD": "lead", + "HHSRSRADIA": "radiation", + "HHSRSFUEL": "uncombusted_fuel_gas", + "HHSRSORGAN": "volatile_organic_compounds", + "HHSRSCROWD": "crowding_and_space", + "HHSRSENTRY": "entry_by_intruders", + "HHSRSLIGHT": "lighting", + "HHSRSNOISE": "noise", + "HHSRSDOMES": "domestic_hygiene_pests_and_refuse", + "HHSRSFOOD": "food_safety", + "HHSRSPERS": "personal_hygiene_sanitation_and_drainage", + "HHSRSWATER": "water_supply", + "HHSRSFBATH": "falls_associated_with_baths", + "HHSRSFLEVE": "falls_on_level_surfaces", + "HHSRSFSTAI": "falls_on_stairs_and_steps", + "HHSRSFBETW": "falls_between_levels", + "HHSRSELEC": "electrical_hazards", + "HHSRSFIRE": "fire", + "HHSRSFLAME": "flames_hot_surfaces_and_materials", + "HHSRSEXPLO": "explosions", + "HHSRSPOSI": "ergonomics", + "HHSRSSTRUC": "structural_collapse_and_falling_elements", + + # One-to-many expansions + "HHSRSCO": "carbon_monoxide", + "HHSRSSO2": "sulphur_dioxide_and_smoke", + "HHSRSNO2": "nitrogen_dioxide", + "HHSRSENTRP": "collision_and_entrapment", + "HHSRSCLOW": "collision_hazards_and_low_headroom", +} + CRITERION_B_VARIABLES = [ "external_walls_structure", "lintels", "brickwork_spalling", "wall_finish", "roof_structure", "roof_finish", "chimneys", "windows", "external_doors", "kitchens", "bathrooms", "central_heating_boiler", @@ -203,13 +243,16 @@ B_COMPONENT_LABELS = { "Store Door in External Area", ], "central_heating_boiler": [ - # TODO + # "Heating Improvement Required in Property", + "Boiler Fuel in Property", + "Type of Water Heating in Property", ], "heating_other": [ - # TODO + # "Heating Distribution System in Property" + "Boiler Fuel in Property", + "Type of Water Heating in Property", ], "electrical_systems": [ - # TODO "Electrics Required in Property", ], # Other components @@ -300,6 +343,8 @@ for fn in filenames: with open(os.path.join(folder, fn), "rb") as f: data = json.load(f) + today = pd.Timestamp.today().normalize() + property_info = data["property_info"] if property_info["PROP TYPE"] in ["HOU"]: property_type = "house" @@ -310,7 +355,6 @@ for fn in filenames: raise NotImplementedError("Unknown property type") # ---------------- Criterion A ---------------- - # TODO: Map out the sub-information # Critrion A: pass/fail # If fail, why? for hhsrs_variable, mapping in HHSRS_MAPPING.items(): @@ -331,115 +375,97 @@ for fn in filenames: else: raise ValueError("Unknown attribute code") check_pass.append(result) + append_result( + decent_homes_meta, + criteria="A", + variable=hhsrs_variable, + sub_variable=ELEMENT_CODE_TO_DESCRIPTION[element_code], + result=result, + install_date=None, + expiry_date=None, + ) # We check if we have a pass, fail or no_data - if all([x == "pass" for x in check_pass]): - hhsrs_result = "pass" - elif any([x == "fail" for x in check_pass]): - hhsrs_result = "fail" - elif any([x == "no_data" for x in check_pass]): - hhsrs_result = "no_data" - else: - raise NotImplementedError("Mixed results not implemented") - decent_homes_meta.append( - {"variable": hhsrs_variable, 'result': hhsrs_result, "hhsrs_rank": None, "hhsrs_score": None, - "install_date": None} - ) + # if all([x == "pass" for x in check_pass]): + # hhsrs_result = "pass" + # elif any([x == "fail" for x in check_pass]): + # hhsrs_result = "fail" + # elif any([x == "no_data" for x in check_pass]): + # hhsrs_result = "no_data" + # else: + # raise NotImplementedError("Mixed results not implemented") # ---------------- Criterion B ---------------- # Check each of the components - component_pass_or_fail = [] - # TODO: Delete me - component, labels = list(B_COMPONENT_LABELS.items())[9] - label = labels[0] - # TODO: need to handle the case where there is no survey data at all for a component + # ---------------- Criterion B ---------------- + property_boiler = get_element(data["elements"], "Boiler Fuel in Property") + for component, labels in B_COMPONENT_LABELS.items(): - # TODO: labels may not need to be multiple variables for label in labels: - # Grab the label label_data = get_element(data["elements"], label) + + # Handle no-data or not-applicable if label_data["ATTRIBUTE CODE"] in ["UNKNOWN", "NONE", "UNKNOWNG", "UNKNOWNS"]: - # This isn't applicable - component_pass_or_fail.append( - { - "component": component, - "label": label, - "install_date": None, - "remaining_life": None, - "is_old": False, - "has_failed": False, - "result": "pass", - "appliable": False - } - ) + # append_result( + # decent_homes_meta, + # criteria="B", + # variable=component, + # sub_variable=label, + # result="pass", + # install_date=None, + # expiry_date=None, + # ) continue - # 1) We check if the component is old + + # Special skip conditions for heating + no_boiler_condition = ( + property_boiler["ATTRIBUTE CODE"] in ["NONENOCH"] + and component == "central_heating_boiler" + ) + other_heating_condition = ( + label_data["ATTRIBUTE CODE"] in ["NONENOCH"] + and component == "heating_other" + ) + if no_boiler_condition or other_heating_condition: + # append_result( + # decent_homes_meta, + # criteria="B", + # variable=component, + # sub_variable=label, + # result="pass", + # install_date=None, + # expiry_date=None, + # ) + continue + + # Normal case: evaluate install date + lifetime + remaining life install_date = pd.to_datetime(label_data["INSTALL DATE"]) if pd.isnull(install_date): - raise ValueError("Missing install date - pls check") + raise ValueError(f"Missing install date for {component}/{label}") + component_lifetime = COMPONENT_LIFESPANS[component][property_type] - # This should be populated, and for the pilot it's okay if this errors if missing - we'll handle accordingly is_old = years_between(today.to_pydatetime(), install_date.to_pydatetime()) > component_lifetime - # 2) We check if the component is in poor condition + if pd.isnull(label_data["REMAINING LIFE"]): - raise ValueError("Missing remaining life - pls check") + raise ValueError(f"Missing remaining life for {component}/{label}") has_failed = label_data["REMAINING LIFE"] < 0 - # The component needs to have both failed and be old to fail criterion B + + expiry_date = install_date + pd.DateOffset(years=component_lifetime) component_result = "fail" if is_old and has_failed else "pass" - component_pass_or_fail.append( - { - "component": component, - "component_type": "key" if component in KEY_COMPONENTS else "other", - "component_sub_description": label_data["ATTRIBUTE CODE DESCRIPTION"], - "label": label, - "install_date": str(install_date), - "remaining_life": label_data["REMAINING LIFE"], - "is_old": is_old, - "has_failed": has_failed, - "result": component_result, - "appliable": True - } + + # Push into decent_homes_meta + append_result( + decent_homes_meta, + criteria="B", + variable=component, + sub_variable=label, + result=component_result, + install_date=str(install_date), + expiry_date=str(expiry_date), ) - # TODO: We need to check by component - # Example of a pass for a component - # [ - # {"component": "external_walls", "component_type": "key", "descr": "A", "result": "pass"}, - # {"component": "external_walls", "component_type": "key", "descr": "B", "result": "pass"}, - # {"component": "external_walls", "component_type": "key", "descr": "C", "result": "pass"}, - # ] - - # Example of a fail for a component - # [ - # {"component": "external_walls", "component_type": "key", "descr": "A", "result": "pass"}, - # {"component": "external_walls", "component_type": "key", "descr": "B", "result": "fail"}, - # {"component": "external_walls", "component_type": "key", "descr": "C", "result": "pass"}, - # ] - - # Example of a no data for a component - # [ - # {"component": "external_walls", "component_type": "key", "descr": "A", "result": "pass"}, - # {"component": "external_walls", "component_type": "key", "descr": "B", "result": "nodata", "appliable": True}, - # {"component": "external_walls", "component_type": "key", "descr": "C", "result": "pass"}, - # ] - # OR - # Everything is unknown - # [ - # {"component": "external_walls", "component_type": "key", "descr": "A", "result": "pass", "appliable": False}, - # {"component": "external_walls", "component_type": "key", "descr": "B", "result": "pass", "appliable": False}, - # {"component": "external_walls", "component_type": "key", "descr": "C", "result": "pass", "appliable": False}, - # ] - - # Component 1: pass/fail, key: true/False - # Component 2: pass/fail, key: true/False - # Component 3: pass/fail, key: true/False - # Component 4: pass/fail, key: true/False - # Component 4: pass/fail, key: true/False - # -> Decide on outcome. If failure of 1 key component -> fail criterion B, or 2 other components -> fail criterion B - # ---------------- Criterion C ---------------- - today = pd.Timestamp.today().normalize() # Guard: property type string already set earlier is_flat = (property_info["PROP TYPE"] == "FLA") @@ -456,8 +482,13 @@ for fn in filenames: else: raise NotImplementedError("Kitchen data missing - pls check") append_result( - decent_homes_meta, "kitchen_less_than_20_years_old", kitchen_age_result, - install_date=str(kit_install), expiry_date=str(kit_next_due) + decent_homes_meta, + criteria="C", + variable="kitchen_less_than_20_years_old", + sub_variable="kitchen_less_than_20_years_old", + result=kitchen_age_result, + install_date=str(kit_install), + expiry_date=str(kit_next_due) ) # 2) Kitchen adequate space/layout @@ -470,7 +501,13 @@ for fn in filenames: raise NotImplementedError("No other observed codes yet") else: raise NotImplementedError("Kitchen data missing - pls check") - append_result(decent_homes_meta, "kitchen_adequate_space_and_layout", kitchen_adequacy_result) + append_result( + decent_homes_meta, + criteria="C", + variable="kitchen_adequate_space_and_layout", + sub_variable="kitchen_adequate_space_and_layout", + result=kitchen_adequacy_result, + ) # 3) Bathroom age ≤ 30 years bath = get_element(data["elements"], LABEL_BATHROOM) @@ -483,7 +520,13 @@ for fn in filenames: else: raise NotImplementedError("Bathroom data missing - pls check") append_result( - decent_homes_meta, "bathroom_less_than_30_years_old", bathroom_age_result, install_date=str(bth_install) + decent_homes_meta, + criteria="C", + variable="bathroom_less_than_30_years_old", + sub_variable="bathroom_less_than_30_years_old", + result=bathroom_age_result, + install_date=str(bth_install), + expiry_date=bth_next_due ) # 4) Bathroom/WC appropriately located @@ -496,7 +539,13 @@ for fn in filenames: else: raise NotImplementedError("Bathroom data missing - pls check") - append_result(decent_homes_meta, "bathroom_wc_appropriately_located", bathroom_location_result) + append_result( + decent_homes_meta, + criteria="C", + variable="bathroom_wc_appropriately_located", + sub_variable="bathroom_wc_appropriately_located", + result=bathroom_location_result + ) # 5) Adequate external noise insulation noise = get_element(data["elements"], LABEL_NOISE) @@ -508,7 +557,13 @@ for fn in filenames: raise NotImplementedError("No other observed codes yet") else: raise NotImplementedError("Noise insulation data missing - pls check") - append_result(decent_homes_meta, "adequate_external_noise_insulation", noise_result) + append_result( + decent_homes_meta, + criteria="C", + variable="adequate_external_noise_insulation", + sub_variable="adequate_external_noise_insulation", + result=noise_result + ) # 6) Adequate common entrance areas (flats only) if is_flat: @@ -535,7 +590,13 @@ for fn in filenames: else: raise NotImplementedError("Heating element missing in dataset") - append_result(decent_homes_meta, "efficient_heating_system_type", heating_type_result) + append_result( + decent_homes_meta, + criteria="D", + variable="efficient_heating_system_type", + sub_variable="efficient_heating_system_type", + result=heating_type_result + ) # heating distribution heating_dist = get_element(data["elements"], "Heating Distribution System in Property") @@ -550,7 +611,13 @@ for fn in filenames: else: raise NotImplementedError("Heating distribution element missing in dataset") - append_result(decent_homes_meta, "efficient_heating_distribution", heating_dist_result) + append_result( + decent_homes_meta, + criteria="D", + variable="efficient_heating_distribution", + sub_variable="efficient_heating_distribution", + result=heating_dist_result + ) # insulation loft = get_element(data["elements"], "Size in mm of Loft Insulation Thickness in Property") @@ -570,7 +637,13 @@ for fn in filenames: raise NotImplementedError("Unknown loft insulation code - pls check") else: raise NotImplementedError("Loft insulation data missing - pls check") - append_result(decent_homes_meta, "loft_insulation_sufficient", loft_result) + append_result( + decent_homes_meta, + criteria="D", + variable="loft_insulation_sufficient", + sub_variable="loft_insulation_sufficient", + result=loft_result + ) # Wall insulation check if wall: @@ -581,7 +654,13 @@ for fn in filenames: raise NotImplementedError("No other observed codes yet") else: raise NotImplementedError("Wall insulation data missing - pls check") - append_result(decent_homes_meta, "wall_insulation_sufficient", wall_result) + append_result( + decent_homes_meta, + criteria="D", + variable="wall_insulation_sufficient", + sub_variable="wall_insulation_sufficient", + result=wall_result + ) # ---------------- Criterion A overall ---------------- a_vars = set(HHSRS_MAPPING.keys()) @@ -596,6 +675,38 @@ for fn in filenames: # ---------------- Criterion B overall ---------------- + component_results = {} + + for component in B_COMPONENT_LABELS.keys(): + comp_rows = [r for r in decent_homes_meta if + r["criteria"] == "B" and r["variable"] == component and r["sub_variable"] is not None] + comp_sub_results = [r["result"] for r in comp_rows] + + if not comp_sub_results: # no rows at all + comp_result = "no_data" + elif any(r == "fail" for r in comp_sub_results): + comp_result = "fail" + elif all(r == "pass" for r in comp_sub_results if r != "no_data"): + comp_result = "pass" + elif all(r == "no_data" for r in comp_sub_results): + comp_result = "no_data" + else: + comp_result = "no_data" + + component_results[component] = comp_result + + key_fails = [c for c, r in component_results.items() if c in KEY_COMPONENTS and r == "fail"] + other_fails = [c for c, r in component_results.items() if c in OTHER_COMPONENTS and r == "fail"] + + if key_fails: + criterion_b_result = "fail" + elif len(other_fails) >= 2: + criterion_b_result = "fail" + elif all(r == "no_data" for r in component_results.values()): + criterion_b_result = "no_data" + else: + criterion_b_result = "pass" + # ---------------- Criterion C overall ---------------- criterion_c_vars = [ "kitchen_less_than_20_years_old", @@ -635,7 +746,7 @@ for fn in filenames: "uprn": property_info.get("UPRN"), # TODO: Need UPRN "creation_date": datetime.now().date().isoformat(), "criterion_a": criterion_a_result, - "criterion_b": None, # not yet implemented + "criterion_b": criterion_b_result, "criterion_c": criterion_c_result, "criterion_d": criterion_d_result, "decent_homes": ( From e410e8d9c862e1d46ec3270399af97a833f970dd Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Mon, 22 Sep 2025 22:36:47 +0100 Subject: [PATCH 6/7] minor tidy --- etl/customers/waltham_forest/decent_homes_pilot.py | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/etl/customers/waltham_forest/decent_homes_pilot.py b/etl/customers/waltham_forest/decent_homes_pilot.py index 33836236..0c7ea98f 100644 --- a/etl/customers/waltham_forest/decent_homes_pilot.py +++ b/etl/customers/waltham_forest/decent_homes_pilot.py @@ -1,12 +1,8 @@ import json import os - import pandas as pd - from datetime import datetime -from docutils.nodes import table - def years_between(d1, d2): # precise year difference (accounts for months/days) @@ -35,15 +31,6 @@ def append_result(decent_homes_meta, criteria, variable, sub_variable, result, i folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Waltham Forest/Decent Homes Pilot" filenames = ["flat 1.json", "house 1.json"] -houses_waltham_forest_data = pd.read_excel( - os.path.join(folder, "LBWF - Example Asset Data September 2025.xlsx"), - sheet_name="Houses Asset Data" -) -flats_waltham_forest_data = pd.read_excel( - os.path.join(folder, "LBWF - Example Asset Data September 2025.xlsx"), - sheet_name="CHINGFORD ROAD 236-254 Asset Bl" -) - # Standardised variables which will form the enums in the db HHSRS_VARIABLES = [ "damp_and_mould_growth", From 754644a8574827f4cf318b5fc2e6faa495fda2fb Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Wed, 24 Sep 2025 00:26:46 +0100 Subject: [PATCH 7/7] minor bug fix with solar --- backend/apis/GoogleSolarApi.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/backend/apis/GoogleSolarApi.py b/backend/apis/GoogleSolarApi.py index 532afec0..a8982061 100644 --- a/backend/apis/GoogleSolarApi.py +++ b/backend/apis/GoogleSolarApi.py @@ -854,18 +854,21 @@ class GoogleSolarApi: ): continue + solar_api_client = cls(api_key=google_solar_api_key, solar_materials=solar_materials) + if unit["longitude"] is None or unit["latitude"] is None: # At this point, we've checked that solar PV is valid, and so we provide some defaults property_instance.set_solar_panel_configuration( solar_panel_configuration={ "insights_data": None, - "panel_performance": cls.default_panel_performance(property_instance=property_instance), + "panel_performance": solar_api_client.default_panel_performance( + property_instance=property_instance + ), "unit_share_of_energy": 1 }, ) continue - solar_api_client = cls(api_key=google_solar_api_key, solar_materials=solar_materials) solar_api_client.get( longitude=unit["longitude"], latitude=unit["latitude"],