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": (