handling funky edge case where old property has been surveyed as a new buld

.idea/Model.iml

@@ -7,7 +7,7 @@
       <sourceFolder url="file://$MODULE_DIR$/open_uprn" isTestSource="false" />
       <sourceFolder url="file://$MODULE_DIR$/recommendations" isTestSource="false" />
     </content>
-    <orderEntry type="jdk" jdkName="AssetList" jdkType="Python SDK" />
+    <orderEntry type="jdk" jdkName="Fastapi-backend" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
   <component name="PyNamespacePackagesService">

asset_list/AssetList.py

@@ -456,6 +456,12 @@ class AssetList:
                 self.standardised_asset_list[self.landlord_built_form] = (
                     self.standardised_asset_list["Archetype"].copy()
                 )
+            else:
+                # We use the EPC data as our property type and built form
+                self.landlord_property_type = self.STANDARD_PROPERTY_TYPE
+                self.landlord_built_form = self.STANDARD_BUILT_FORM
+                self.standardised_asset_list[self.landlord_property_type] = None
+                self.standardised_asset_list[self.landlord_built_form] = None
 
         # Handle the case where the property type column is the same as the built type
         if self.landlord_property_type == self.landlord_built_form:

asset_list/app.py

@@ -89,37 +89,6 @@ def app():
     # - We want: fully insulated property (all wall types), EPC D or below (floors should be solid)
     # - Or the insulation required is loft/cavity (floors should be solid)
 
-    # Sandwell
-    data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Sandwell"
-    data_filename = "Sandwell BC - Full Asset List MAIN.xlsx"
-    sheet_name = "Sheet1"
-    postcode_column = 'Post-Code'
-    fulladdress_column = "Address"
-    address1_column = None
-    address1_method = "house_number_extraction"
-    address_cols_to_concat = []
-    missing_postcodes_method = None
-    landlord_year_built = "Build-Date"
-    landlord_os_uprn = None
-    landlord_property_type = None
-    landlord_built_form = None
-    landlord_wall_construction = "ConstructionTypeName"
-    landlord_roof_construction = None
-    landlord_heating_system = "Heat Type"
-    landlord_existing_pv = None
-    landlord_property_id = "Place-Ref"
-    landlord_sap = None
-    outcomes_filename = None
-    outcomes_sheetname = None
-    outcomes_postcode = None
-    outcomes_houseno = None
-    outcomes_id = None
-    outcomes_address = None
-    master_filepaths = []
-    master_to_asset_list_filepath = None
-    phase = True
-    ecosurv_landlords = None
-
     # Torus
     data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Torus/Phase 1"
     data_filename = "Torus Property Asset List - Phase 1.xlsx"
@@ -150,33 +119,6 @@ def app():
     master_to_asset_list_filepath = None
     phase = True
 
-    # Ealing - houses
-    data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Ealing"
-    data_filename = "Ealing_rechecked_cleaned_05042025.csv"
-    sheet_name = None
-    postcode_column = 'Postcode'
-    fulladdress_column = "Address"
-    address1_column = None
-    address1_method = "house_number_extraction"
-    address_cols_to_concat = []
-    missing_postcodes_method = None
-    landlord_year_built = "Year Built"
-    landlord_os_uprn = None
-    landlord_property_type = "Property Type Code"
-    landlord_built_form = None
-    landlord_wall_construction = None
-    landlord_heating_system = None
-    landlord_existing_pv = None
-    landlord_property_id = "Property ref"
-    outcomes_filename = None
-    outcomes_sheetname = None
-    outcomes_postcode = None
-    outcomes_houseno = None
-    outcomes_id = None
-    outcomes_address = None
-    master_filepaths = []
-    master_to_asset_list_filepath = None
-
     # Southern Midlands
     data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Southern/Midlands Properties - Apr 2025"
     data_filename = "Southern Housing Midlands Property List - combined.xlsx"
@@ -204,67 +146,6 @@ def app():
     master_filepaths = []
     master_to_asset_list_filepath = None
 
-    # Live West (2018 Asset list)
-    data_folder = (
-        "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Livewest/Programme Update - March 2025/2018 Asset List"
-    )
-    data_filename = "LIVEWEST  STOCK - 23rd October 2018.xlsx"
-    sheet_name = "Assets"
-    postcode_column = 'Postcode'
-    fulladdress_column = "Address"
-    address1_column = None
-    address1_method = "house_number_extraction"
-    address_cols_to_concat = []
-    missing_postcodes_method = None
-    landlord_year_built = "Build Year"
-    landlord_os_uprn = None
-    landlord_property_type = "Property Archetype"
-    landlord_built_form = None
-    landlord_wall_construction = None
-    landlord_heating_system = "Heating Fuel Type"
-    landlord_existing_pv = None
-    landlord_property_id = "Uprn - DO NOT DELETE"
-    outcomes_filename = "RT - LiveWest.xlsx"
-    outcomes_sheetname = "Feedback"
-    outcomes_postcode = "Poscode"
-    outcomes_houseno = "No."
-    outcomes_id = "UPRN"
-    master_filepaths = [
-        "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Livewest/Programme Update - March 2025/Rolling Master "
-        "- redacted for analysis/CAVITY-Table 1.csv"
-    ]
-    master_to_asset_list_filepath = None
-
-    # Live West (South West asset list)
-    data_folder = ("/Users/khalimconn-kowlessar/Documents/hestia/Customers/Livewest/Programme Update - March "
-                   "2025/Livewest Asset List (Original) - csv")
-    data_filename = "Report-Table 1.csv"
-    sheet_name = None
-    postcode_column = 'Postcode'
-    fulladdress_column = "T1_Address"
-    address1_column = None
-    address1_method = "house_number_extraction"
-    address_cols_to_concat = []
-    missing_postcodes_method = None
-    landlord_year_built = "Build Yr"
-    landlord_os_uprn = None
-    landlord_property_type = "T1_AssetType"
-    landlord_built_form = "T1_AssetType"
-    landlord_wall_construction = "Wall Type Cavity"
-    landlord_heating_system = "Heating Fuel"
-    landlord_existing_pv = None
-    landlord_property_id = "T1_UPRN"
-    outcomes_filename = "RT - LiveWest.xlsx"
-    outcomes_sheetname = "Feedback"
-    outcomes_postcode = "Poscode"
-    outcomes_houseno = "No."
-    outcomes_id = "UPRN"
-    master_filepaths = [
-        "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Livewest/Programme Update - March 2025/Rolling Master "
-        "- redacted for analysis/CAVITY-Table 1.csv"
-    ]
-    master_to_asset_list_filepath = None
-
     # PFP London
     data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Places For People/London"
     data_filename = "PFP AREAS SURROUNDING LONDON - JAY, RUTH & LANE.xlsx"
@@ -702,6 +583,9 @@ def app():
         epc_data.append(csv_data)
 
     epc_df = pd.concat(epc_data)
+    if "estimated" not in epc_df.columns:
+        epc_df["estimated"] = False
+
     epc_df["estimated"] = epc_df["estimated"].fillna(False)
 
     # We expand out the recommendations

asset_list/mappings/built_form.py

@@ -209,5 +209,8 @@ BUILT_FORM_MAPPINGS = {
     'Bungalow Semi Detach': 'semi-detached',
     '4 Ext. Wall Flat': 'unknown',
     '6 Ext. Wall Flat': 'unknown',
-    '5 Ext. Wall Flat': 'unknown'
+    '5 Ext. Wall Flat': 'unknown',
+    'Unknown': 'unknown',
+    'Enclosed mid-terrace': 'mid-terrace',
+    'Enclosed end-terrace': 'end-terrace'
 }

asset_list/mappings/property_type.py

@@ -193,5 +193,6 @@ PROPERTY_MAPPING = {
     'Maisonette 3 Ext. Wall': 'maisonette',
     'Maisonette 2 Ext. Wall': 'maisonette',
     '5 Ext. Wall Flat': 'flat',
-    'Bungalow Semi Detached': 'bungalow'
+    'Bungalow Semi Detached': 'bungalow',
+    'COMINT': 'unknown'
 }

asset_list/mappings/walls.py

@@ -211,6 +211,13 @@ WALL_CONSTRUCTION_MAPPINGS = {
     'Integer': 'system built',
     'Cornish': 'system built',
     'Rwate': 'system built',
-    'Hill Presweld Steel': 'system built'
+    'Hill Presweld Steel': 'system built',
+
+    'Cavity Filled Cavity': 'filled cavity',
+    'Cavity Unknown': 'cavity unknown insulation',
+    'Cavity Filled Cavity (internal)': 'filled cavity',
+    '': 'unknown',
+    'Cavity Internal Insulation': 'filled cavity',
+    'Cavity As Built': "uninsulated cavity"
 
 }

backend/engine/engine.py

@@ -411,7 +411,7 @@ def get_funding_data():
 
 
 async def model_engine(body: PlanTriggerRequest):
-    logger.info("Model Engine triggered with body: %s", body.model_dump_json())
+    logger.info("Model Engine triggered with body: %s", json.loads(body.model_dump_json()))
 
     logger.info("Connecting to db")
     session = sessionmaker(bind=db_engine)()

etl/customers/l_and_g/risk_matrix.py

@@ -16,48 +16,56 @@ def app():
     ]
 
     pricing_matrix = {
-        "cavity_wall_insulation": 14.5,
+        "Cavity wall insulation": 14.5,
         "ventilation": 350,
-        "room_roof_insulation": 210,
+        "Room Roof Insulation": 210,
         "Loft insulation": 15,
-        "internal_wall_insulation": 215,
+        "Internal wall insulation": 215,
-        "external_wall_insulation": 298.35,
+        "External wall insulation": 298.35,
-        "low_energy_lighting": 35,  # per light
+        "Solid wall insulation": 215,
-        "flat_roof_insulation": 195,
+        "LEDs": 35,  # per light
-        "double_glazing": 1140,
+        "Flat Roof Insulation": 195,
+        "Double Glazing": 1140,
         "secondary_glazing": 970,
-        "air_source_heat_pump": 16500,
+        "5kw ASHP feeding heating & Hot water (dual tariff)": 14738,
-        "solar_pv": 6200,
+        "11.2kw ASHP feeding heating & Hot water (dual tariff)": 16541,
-        "high_heat_retention_storage": 1000,  # per heater
+        "3 kWp Solar PV": 4552.32,
+        "4 kWp Solar PV": 4892.8,
+        "4.3 kWp Solar PV": 4961.44,
+        "4.8 kWp Solar PV": 5414,
+        '5 kWp Solar PV': 5509.71,
+        '5.5 kWp Solar PV': 5631.92,
+        "HHRSH (dual tariff)": 1000,  # per heater
+        "Suspended floor insulation": 75
     }
     dwelling_types = [
         "Semi Detached House",
-        "Detached House",
+        "Detached house",
-        "Mid Terrace House",
+        "Mid Terrace house",
         "Mid Floor Flat",
         "Top Floor Flat",
         "Ground Floor Flat"
     ]
     num_floors_map = {
         "Semi-detached house": 2,
-        "Detached House": 2,
+        "Detached house": 2,
-        "Mid Terrace House": 2,
+        "Mid Terrace house": 2,
         "Mid Floor Flat": 1,
         "Top Floor Flat": 1,
         "Ground Floor Flat": 1
     }
     built_form_map = {
         "Semi-detached house": "Semi-Detached",
-        "Detached House": "Detached",
+        "Detached house": "Detached",
-        "Mid Terrace House": "Mid Terrace",
+        "Mid Terrace house": "Mid Terrace",
         "Mid Floor Flat": "Semi-Detached",
         "Top Floor Flat": "Semi-Detached",
         "Ground Floor Flat": "Semi-Detached"
     }
     lighting_count = {
         "Semi-detached house": 15,
-        "Detached House": 19,
+        "Detached house": 19,
-        "Mid Terrace House": 12,
+        "Mid Terrace house": 12,
         "Mid Floor Flat": 10,
         "Top Floor Flat": 10,
         "Ground Floor Flat": 10
@@ -239,13 +247,42 @@ def app():
             floor_area=row["area"],
             number_habitable_rooms=n_rooms
         )
-        measure = row["Measure added"]
+        cost_upper_bound = None
-        unit_cost = pricing_matrix[measure]
+        if pd.isnull(row["Measure added"]):
+            unit_cost = None
+        else:
+            measure = row["Measure added"]
+            unit_cost = pricing_matrix[measure]
 
         if pd.isnull(row["Measure added"]):
             cost = None
         elif row["Measure added"] == "Loft insulation":
             cost = unit_cost * ground_floor_area
+        elif row["Measure added"] in ["Cavity wall insulation", "Internal wall insulation"]:
+            cost = unit_cost * external_wall_area + pricing_matrix["ventilation"] * 3
+        elif row["Measure added"] == "Solid wall insulation":
+            cost = unit_cost * external_wall_area + pricing_matrix["ventilation"] * 3
+            cost_upper_bound = pricing_matrix["External wall insulation"] * external_wall_area + pricing_matrix[
+                "ventilation"] * 3
+        elif row["Measure added"] == "Double Glazing":
+            cost = unit_cost * n_windows
+        elif row["Measure added"] == "LEDs":
+            cost = unit_cost * lighting_count[row["Property Type"]]
+        elif row["Measure added"] in [
+            '5kw ASHP feeding heating & Hot water (dual tariff)',
+            '11.2kw ASHP feeding heating & Hot water (dual tariff)',
+            "3 kWp Solar PV",
+            '4 kWp Solar PV',
+            "4.3 kWp Solar PV",
+            '4.8 kWp Solar PV',
+            '5 kWp Solar PV',
+            '5.5 kWp Solar PV'
+        ]:
+            cost = unit_cost
+        elif row["Measure added"] == "HHRSH (dual tariff)":
+            cost = unit_cost * (n_rooms + 1)
+        elif row["Measure added"] == "Suspended floor insulation":
+            cost = unit_cost * ground_floor_area
         else:
             raise Exception()
 
@@ -254,6 +291,33 @@ def app():
                 "row_id": row["row_id"],
                 "epc": epc,
                 "sap": sap,
-                "cost": cost
+                "cost": cost,
+                "cost upper bound": cost_upper_bound
             }
         )
+
+    cost_data = pd.DataFrame(cost_data)
+
+    risk_matrix = pd.merge(
+        epr_data,
+        cost_data,
+        on="row_id",
+    )
+
+    risk_matrix["contingency"] = risk_matrix["cost"] * contingency
+    risk_matrix["upper bound coningency"] = risk_matrix["cost upper bound"] * contingency
+
+    pricing_df = pd.DataFrame(
+        [
+            {
+                "Measure": k,
+                "Unit Cost": v
+            }
+            for k, v in pricing_matrix.items()
+        ]
+    )
+
+    with pd.ExcelWriter(
+        "/Users/khalimconn-kowlessar/Documents/hestia/Customers/L&G/Risk Matrix/risk_matrix.xlsx") as writer:
+        risk_matrix.to_excel(writer, sheet_name="Risk Matrix", index=False)
+        pricing_df.to_excel(writer, sheet_name="Pricing Assumptions", index=False)

etl/epc/Record.py

@@ -355,6 +355,7 @@ class EPCRecord:
         self._clean_floor_level()
         self._clean_floor_height()
         self._clean_constituency()
+        self._clean_new_build_descriptions()
 
         # self._clean_potential_energy_efficiency()
         # self._clean_environment_impact_potential()
@@ -397,6 +398,10 @@ class EPCRecord:
         if self.prepared_epc["floor-height"] <= 1.665:
             self.prepared_epc["floor-height"] = average
 
+    def _clean_new_build_descriptions(self):
+        for col in ['roof-description', 'walls-description', 'floor-description']:
+            self.prepared_epc[col] = self.prepared_epc[col].replace("W/m²K", "W/m-¦K")
+
     def _clean_constituency(self):
         """
         We handle the single case of finding a missing constituency by using the local authority

recommendations/RoofRecommendations.py

@@ -163,15 +163,12 @@ class RoofRecommendations:
         if self.property.roof["is_thatched"]:
             return
 
-        # If we have a u-value already, need to implement this
+        # If we have a u-value and we don't have a non-invasive recommendation, we can't recommend anything
-        if u_value:
+        if u_value and not any(
-            if u_value <= self.BUILDING_REGULATIONS_PART_L_MAX_U_VALUE:
+            x in MEASURE_MAP["roof_insulation"] for x in [r["type"] for r in self.property.non_invasive_recommendations]
-                # The Roof is already compliant
+        ):
-                return
+            # We don't have enough information to provide a recommendation
-
+            return
-            if self.property.data["transaction-type"] in ["new dwelling", "not sale or rental"]:
-                return
-            raise NotImplementedError("Implement me")
 
         u_value = get_roof_u_value(
             insulation_thickness=self.property.roof["insulation_thickness"],

recommendations/WallRecommendations.py

@@ -227,24 +227,14 @@ class WallRecommendations(Definitions):
             # external wall insulation
             if (
                 (not is_cavity_wall)
-                and (self.property.year_built >= self.YEAR_WALLS_BUILT_WITH_INSULATION)
                 and (u_value >= self.BUILDING_REGULATIONS_PART_L_MAX_U_VALUE)
             ):
                 # Recommend insulation
-                self.find_insulation(u_value, phase, measures)
+                self.find_insulation(u_value, phase, measures=measures, default_u_values=default_u_values)
                 return
 
-            # We can't detect it's a cavity wall, but it was built after 1990 so likely built with insulation already
+            # We have a sufficiently low U-value
-            # + it already has a U-value better than the building regulations, so we don't need to recommend anything
+            return
-            if (
-                (not is_cavity_wall)
-                and ((self.property.year_built >= self.YEAR_WALLS_BUILT_WITH_INSULATION)
-                     or (u_value <= self.BUILDING_REGULATIONS_PART_L_MAX_U_VALUE))
-            ):
-                # Recommend nothing
-                return
-
-            raise NotImplementedError("Not implemented yet")
 
         u_value = get_wall_u_value(
             clean_description=self.property.walls["clean_description"],
@@ -626,7 +616,7 @@ class WallRecommendations(Definitions):
                         "walls_thermal_transmittance_ending": new_u_value
                     }
 
-                    if default_u_values:
+                    if default_u_values and "Average thermal transmittance" not in new_description:
                         # If we're using default U-values, we overwrite new_u_value
                         new_u_value = get_wall_u_value(
                             clean_description=new_description,