Merge pull request #453 from Hestia-Homes/main

adding extra inputs into backend body

asset_list/AssetList.py

@@ -2622,7 +2622,8 @@ class AssetList:
 
         # Add in deal and pipeline information
         programme_data["dealname"] = (
-            programme_data[self.STANDARD_FULL_ADDRESS] + " : " + programme_data["domna_product"]
+            programme_data[self.STANDARD_FULL_ADDRESS] + ", " +
+            programme_data[self.STANDARD_POSTCODE] + " : " + programme_data["domna_product"]
         )
         programme_data['Pipeline <DEAL pipeline>'] = hubspot_config.CRM_PIPELINE_NAME
         programme_data['Associations: Listing'] = "Property Owner"
@@ -2656,7 +2657,11 @@ class AssetList:
 
         # Ammend the property type and built form columns
         programme_data["hubspot_property_type"] = programme_data[self.STANDARD_PROPERTY_TYPE].copy()
-        programme_data["hubspot_built_form"] = programme_data[self.STANDARD_BUILT_FORM].copy()
+        # We don't already have this
+        if self.STANDARD_BUILT_FORM in programme_data.columns:
+            programme_data["hubspot_built_form"] = programme_data[self.STANDARD_BUILT_FORM].copy()
+        else:
+            programme_data["hubspot_built_form"] = None
 
         def _replace_property_description_data(programme_data, column_name):
             """

asset_list/abs_estimates.py

@@ -13,10 +13,22 @@ from backend.app.utils import sap_to_epc
 load_dotenv(dotenv_path="backend/.env")
 EPC_AUTH_TOKEN = os.getenv("EPC_AUTH_TOKEN")
 
+# project = pd.read_excel(
+
+# )
+#
+# cavity = project[project["cavity_reason"].isin(
+#     ["EPC Shows Empty Cavity: SAP Rating 54 or less", "EPC Shows Empty Cavity: SAP Rating 55-68"]
+# )]
+
 asset_list = pd.read_excel(
-    "/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/Livewest South-West - Standardised V2.xlsx",
+    "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Abri/Abs Rates/Desktop ABRI data - Standardised After "
-    sheet_name="Cavity Route (Insta Review)"
+    "Programmes.xlsx",
+    sheet_name="Reviewed List"
 )
+asset_list = asset_list[asset_list["cavity_reason"].isin(
+    ["EPC Shows Empty Cavity: SAP Rating 54 or less", "EPC Shows Empty Cavity: SAP Rating 55-68"]
+)]
 
 abs_matrix = pd.read_csv(
     "/Users/khalimconn-kowlessar/Downloads/ECO4 Full Project Scores Matrix.csv"
@@ -30,7 +42,7 @@ pps_matrix.columns = [c.strip() for c in pps_matrix.columns]
 # We need to estimate the number of points the work will produce and the finishing band. For this, we assume 7 for
 # cavity and 15 for solar. We'll be more specific in the future, but for now, this is a good enough estimate.
 route = asset_list[["domna_address_1", "domna_postcode", "epc_os_uprn"]].rename(
-    columns={"domna_address_1": "address", "domna_postcode": "postcode", "epc_os_uprn": "upr"}
+    columns={"domna_address_1": "address", "domna_postcode": "postcode", "epc_os_uprn": "uprn"}
 )
 route["address"] = route["address"].astype(str)
 
@@ -42,120 +54,121 @@ asset_list_epc_client = AssetListEpcData(
 asset_list_epc_client.get_data()
 asset_list_epc_client.get_non_invasive_recommendations()
 
-solar_sap_points = []
+# solar_sap_points = []
-for r in asset_list_epc_client.non_invasive_recommendations:
-    if not r.get("recommendations"):
-        continue
-    solar_recommendations = [
-        x for x in r["recommendations"] if "solar_pv" in x["type"]
-    ]
-    if solar_recommendations:
-        solar_recommendations = solar_recommendations[0]
-    else:
-        continue
-
-    address = r["address"]
-    postcode = r["postcode"]
-
-    solar_sap_points.append(
-        {
-            "address": address,
-            "postcode": postcode,
-            "sap_points": solar_recommendations["sap_points"]
-        }
-    )
-
-solar_sap_points = pd.DataFrame(solar_sap_points)
-solar_sap_points.drop_duplicates(subset=["address", "postcode"], inplace=True)
-# Store the sap points in the cavity route to csv
-# cwi_sap_points.to_csv(
-#     "/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/cwi_sap_points_livewest_sw.csv",
-#     index=False
-# )
-
-avg_solar_points_by_postcode = solar_sap_points.groupby(["postcode"]).agg({"sap_points": "mean"}).reset_index()
-avg_solar_points = solar_sap_points["sap_points"].median()
-asset_list["domna_address_1"] = asset_list["domna_address_1"].astype(str)
-asset_list = asset_list.merge(
-    solar_sap_points, how="left", left_on=["domna_address_1", "domna_postcode"], right_on=["address", "postcode"]
-).drop(
-    columns=["address", "postcode"]
-)
-
-# Fill the sap points with the average cwi points
-asset_list = asset_list.merge(
-    avg_solar_points_by_postcode.rename(columns={"postcode": "domna_postcode"}),
-    how="left", on=["domna_postcode"], suffixes=("", "_avg")
-)
-asset_list["sap_points"] = asset_list["sap_points"].fillna(asset_list["sap_points_avg"])
-asset_list.drop(columns=["sap_points_avg"], inplace=True)
-
-asset_list["sap_points"] = asset_list["sap_points"].fillna(avg_solar_points)
-asset_list["post_works_sap"] = asset_list["epc_sap_score_on_register"] + asset_list["sap_points"]
-asset_list["post_works_epc"] = asset_list["post_works_sap"].apply(lambda x: sap_to_epc(x))
-asset_list["starting_half_band"] = asset_list["epc_sap_score_on_register"].apply(lambda x: Funding.get_sap_band(x))
-asset_list["ending_half_band"] = asset_list["post_works_sap"].apply(lambda x: Funding.get_sap_band(x))
-asset_list["floor_area_band"] = asset_list["epc_total_floor_area"].apply(lambda x: Funding.get_floor_area_band(x))
-
-asset_list["ending_half_band"] = np.where(
-    (asset_list["post_works_epc"] == asset_list["epc_rating_on_register"]),
-    "Low_C",
-    asset_list["ending_half_band"]
-)
-# Realistically, we'll take the properties to a low C at worst
-asset_list["ending_half_band"] = np.where(
-    (asset_list["post_works_sap"] < 69),
-    "Low_C",
-    asset_list["ending_half_band"]
-)
-
-asset_list = asset_list.merge(
-    abs_matrix, how="left", left_on=["starting_half_band", "ending_half_band", "floor_area_band"],
-    right_on=['Starting Band', 'Finishing Band', 'Floor Area Segment', ]
-)
-asset_list = asset_list.drop(columns=['Starting Band', 'Finishing Band', 'Floor Area Segment'])
-
-asset_list = asset_list.rename(
-    columns={"Cost Savings": "funding_abs"}
-)
-
-print(asset_list["domna_property_id"].duplicated().sum())
-
-# Store this data
-asset_list.to_csv(
-    "/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/livewest_sw_solar_abs_estimates-solar.csv",
-    index=False
-)
-
-# Cavity process!
-# cwi_sap_points = []
 # for r in asset_list_epc_client.non_invasive_recommendations:
 #     if not r.get("recommendations"):
 #         continue
-#     cwi_recommendations = [
+#     solar_recommendations = [
-#         x for x in r["recommendations"] if "cavity_wall_insulation" in x["type"]
+#         x for x in r["recommendations"] if "solar_pv" in x["type"]
 #     ]
-#     if cwi_recommendations:
+#     if solar_recommendations:
-#         cwi_recommendations = cwi_recommendations[0]
+#         solar_recommendations = solar_recommendations[0]
 #     else:
 #         continue
 #
 #     address = r["address"]
 #     postcode = r["postcode"]
 #
-#     cwi_sap_points.append(
+#     solar_sap_points.append(
 #         {
 #             "address": address,
 #             "postcode": postcode,
-#             "sap_points": cwi_recommendations["sap_points"]
+#             "sap_points": solar_recommendations["sap_points"]
 #         }
 #     )
 #
-# cwi_sap_points = pd.DataFrame(cwi_sap_points)
+# solar_sap_points = pd.DataFrame(solar_sap_points)
-# cwi_sap_points = pd.read_csv(
+# solar_sap_points.drop_duplicates(subset=["address", "postcode"], inplace=True)
-#     "/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/cwi_sap_points_livewest_sw.csv"
+# # Store the sap points in the cavity route to csv
+# solar_sap_points.to_csv(
+#     "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Abri/Abs Rates/cwi_sap_points.csv",
+#     index=False
 # )
-# cwi_sap_points.drop_duplicates(subset=["address", "postcode"], inplace=True)
+#
+# avg_solar_points_by_postcode = solar_sap_points.groupby(["postcode"]).agg({"sap_points": "mean"}).reset_index()
+# avg_solar_points = solar_sap_points["sap_points"].median()
+# asset_list["domna_address_1"] = asset_list["domna_address_1"].astype(str)
+# asset_list = asset_list.merge(
+#     solar_sap_points, how="left", left_on=["domna_address_1", "domna_postcode"], right_on=["address", "postcode"]
+# ).drop(
+#     columns=["address", "postcode"]
+# )
+#
+# # Fill the sap points with the average cwi points
+# asset_list = asset_list.merge(
+#     avg_solar_points_by_postcode.rename(columns={"postcode": "domna_postcode"}),
+#     how="left", on=["domna_postcode"], suffixes=("", "_avg")
+# )
+# asset_list["sap_points"] = asset_list["sap_points"].fillna(asset_list["sap_points_avg"])
+# asset_list.drop(columns=["sap_points_avg"], inplace=True)
+#
+# asset_list["sap_points"] = asset_list["sap_points"].fillna(avg_solar_points)
+# asset_list["post_works_sap"] = asset_list["epc_sap_score_on_register"] + asset_list["sap_points"]
+# asset_list["post_works_epc"] = asset_list["post_works_sap"].apply(lambda x: sap_to_epc(x))
+# asset_list["starting_half_band"] = asset_list["epc_sap_score_on_register"].apply(lambda x: Funding.get_sap_band(x))
+# asset_list["ending_half_band"] = asset_list["post_works_sap"].apply(lambda x: Funding.get_sap_band(x))
+# asset_list["floor_area_band"] = asset_list["epc_total_floor_area"].apply(lambda x: Funding.get_floor_area_band(x))
+#
+# asset_list["ending_half_band"] = np.where(
+#     (asset_list["post_works_epc"] == asset_list["epc_rating_on_register"]),
+#     "Low_C",
+#     asset_list["ending_half_band"]
+# )
+# # Realistically, we'll take the properties to a low C at worst
+# asset_list["ending_half_band"] = np.where(
+#     (asset_list["post_works_sap"] < 69),
+#     "Low_C",
+#     asset_list["ending_half_band"]
+# )
+#
+# asset_list = asset_list.merge(
+#     abs_matrix, how="left", left_on=["starting_half_band", "ending_half_band", "floor_area_band"],
+#     right_on=['Starting Band', 'Finishing Band', 'Floor Area Segment', ]
+# )
+# asset_list = asset_list.drop(columns=['Starting Band', 'Finishing Band', 'Floor Area Segment'])
+#
+# asset_list = asset_list.rename(
+#     columns={"Cost Savings": "funding_abs"}
+# )
+#
+# print(asset_list["domna_property_id"].duplicated().sum())
+#
+# # Store this data
+# asset_list.to_csv(
+#     "/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/livewest_sw_solar_abs_estimates-solar.csv",
+#     index=False
+# )
+
+# Cavity process!
+cwi_sap_points = []
+for r in asset_list_epc_client.non_invasive_recommendations:
+    if not r.get("recommendations"):
+        continue
+    cwi_recommendations = [
+        x for x in r["recommendations"] if "cavity_wall_insulation" in x["type"]
+    ]
+    if cwi_recommendations:
+        cwi_recommendations = cwi_recommendations[0]
+    else:
+        continue
+
+    address = r["address"]
+    postcode = r["postcode"]
+
+    cwi_sap_points.append(
+        {
+            "address": address,
+            "postcode": postcode,
+            "type": cwi_recommendations["type"],
+            "sap_points": cwi_recommendations["sap_points"]
+        }
+    )
+
+cwi_sap_points = pd.DataFrame(cwi_sap_points)
+cwi_sap_points = pd.read_csv(
+    "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Abri/Abs Rates/cwi_sap_points.csv",
+)
+cwi_sap_points.drop_duplicates(subset=["address", "postcode"], inplace=True)
 avg_cwi_points_by_postcode = cwi_sap_points.groupby(["postcode"]).agg({"sap_points": "mean"}).reset_index()
 avg_cwi_points = cwi_sap_points["sap_points"].median()
 asset_list = asset_list.merge(
@@ -186,13 +199,22 @@ asset_list["funding_scheme"] = np.where(
     "GBIS",
     "ECO4"
 )
+# Note - anything that is EPC E or below that doesn't go up to a C will be GBIS
+# To detect this, if the starting sap score is 54 or below and the endding SAP sore is 68 or below
+# we will assume it is GBIS
+asset_list["funding_scheme"] = np.where(
+    (asset_list["post_works_sap"] < 69) & (asset_list["epc_sap_score_on_register"] < 55),
+    "GBIS",
+    asset_list["funding_scheme"]
+)
+
 asset_list = asset_list.merge(
     abs_matrix, how="left", left_on=["starting_half_band", "ending_half_band", "floor_area_band"],
     right_on=['Starting Band', 'Finishing Band', 'Floor Area Segment', ]
 )
 asset_list = asset_list.drop(columns=['Starting Band', 'Finishing Band', 'Floor Area Segment'])
 
-# Using CWI solid 1.7 -> 0.3 rates
+# Using CWI 0.033 as the partial project score
 cwi_pps_matrix = pps_matrix[
     pps_matrix["Measure_Type"].isin(["CWI_0.033"])
 ]
@@ -220,10 +242,26 @@ asset_list["funding_abs"] = np.where(
     asset_list["Cost Savings"]
 )
 
-asset_list["domna_property_id"].duplicated().sum()
+from recommendations.recommendation_utils import (
+    estimate_external_wall_area,
+)
+
+# For some reason, estimated insulation wall area is missing
+asset_list["estimated_insulation_wall_area"] = asset_list.apply(
+    lambda x: estimate_external_wall_area(
+        num_floors=x["attribute_est_number_floors"],
+        floor_height=(
+            float(x["epc_floor_height"]) if
+            not pd.isnull(x["epc_floor_height"]) else 2.5
+        ),
+        perimeter=x["attribute_est_perimter"],
+        built_form=x["epc_archetype"]
+    ),
+    axis=1
+)
 
 # Store this data
 asset_list.to_csv(
-    "/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/livewest_sw_abs_estimates.csv",
+    "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Abri/Abs Rates/Abri CWI ABS Estimates.csv",
     index=False
 )

asset_list/app.py

@@ -60,11 +60,11 @@ def app():
     """
 
     # TODO: Delete me
-    data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Apr 2025 Programme Rebuild"
+    data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/NRLA/"
-    data_filename = "Bromford Asset List.xlsx"
+    data_filename = "20250716 Asset List.xlsx"
-    sheet_name = "Asset List"
+    sheet_name = "Sheet 1"
-    postcode_column = 'PostCode'
+    postcode_column = 'Postcode'
-    fulladdress_column = "FullAddress"
+    fulladdress_column = "Full Address"
     address1_column = None
     address1_method = "house_number_extraction"
     address_cols_to_concat = []
@@ -76,24 +76,93 @@ def app():
     landlord_wall_construction = None
     landlord_heating_system = None
     landlord_existing_pv = None
-    landlord_property_id = "Asset"
+    landlord_property_id = "Row ID"
     outcomes_filename = []
     outcomes_sheetname = []
     outcomes_postcode = []
     outcomes_houseno = []
     outcomes_address = []
-    outcomes_id = [None]
+    outcomes_id = []
-    master_filepaths = [os.path.join("/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/",
+    master_filepaths = []
-                                     "Needs ID/SOLAR PV ONLY-Table 1.csv")]
     master_to_asset_list_filepath = None
     asset_list_header = 0
     landlord_block_reference = None
-    master_id_colnames = [None]
+    master_id_colnames = []
     landlord_roof_construction = None
     phase = False
     landlord_sap = None
     ecosurv_landlords = None
 
+    # Southend
+    data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Southend/July 2025 Programme"
+    data_filename = "SOUTHEND - RYAN.xlsx"
+    sheet_name = "July 2025 Surveys"
+    postcode_column = 'Postcode'
+    fulladdress_column = "Full postal address"
+    address1_column = None
+    address1_method = "house_number_extraction"
+    address_cols_to_concat = []
+    missing_postcodes_method = None
+    landlord_year_built = "Property age"
+    landlord_os_uprn = None
+    landlord_property_type = "Property type"
+    landlord_built_form = "Property type"
+    landlord_wall_construction = None
+    landlord_heating_system = None
+    landlord_existing_pv = None
+    landlord_property_id = "ID"
+    outcomes_filename = []
+    outcomes_sheetname = []
+    outcomes_postcode = []
+    outcomes_houseno = []
+    outcomes_address = []
+    outcomes_id = []
+    master_filepaths = []
+    master_to_asset_list_filepath = None
+    asset_list_header = 0
+    landlord_block_reference = None
+    master_id_colnames = []
+    landlord_roof_construction = None
+    phase = False
+    landlord_sap = None
+    ecosurv_landlords = None
+
+    # For Rooftop
+    data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Rooftop"
+    data_filename = "Rooftop Asset List - July 2025.xlsx"
+    sheet_name = "Sheet1"
+    postcode_column = 'post_code'
+    fulladdress_column = None
+    address1_column = "add_1"
+    address1_method = None
+    address_cols_to_concat = [
+        "add_1", "add_2", "add_3", "add_4"
+    ]
+    missing_postcodes_method = None
+    landlord_year_built = "date_built"
+    landlord_os_uprn = None
+    landlord_property_type = "ConstructionStyle"
+    landlord_built_form = "ConstructionStyle"
+    landlord_wall_construction = None
+    landlord_heating_system = "Description"
+    landlord_existing_pv = None
+    landlord_property_id = "PropertyCode"
+    outcomes_filename = [os.path.join(data_folder, "Rooftop_Outcomes.xlsx")]
+    outcomes_sheetname = ["OUTCOMESs"]
+    outcomes_postcode = ["POSTCODE"]
+    outcomes_houseno = ["NO"]
+    outcomes_address = ["ADDRESS"]
+    outcomes_id = [None]
+    master_filepaths = [os.path.join(data_folder, "Master.csv")]
+    master_to_asset_list_filepath = None
+    asset_list_header = 1
+    landlord_block_reference = "bl_rec_ref"
+    master_id_colnames = [None]
+    landlord_roof_construction = None
+    phase = False
+    landlord_sap = None
+    ecosurv_landlords = "rooftop"
+
     # For Housing
     data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/For Housing/New Programme July 2025"
     data_filename = "FOR HOUSING Asset List (Combined).xlsx"

asset_list/hubspot/prepare_for_hubspot.py

@@ -45,13 +45,13 @@ def app():
 
     # inputs:
     reconcile_programme = True  # If True, the hubspot upload will include all properties with a project code
-    customer_domain = "https://calico.org.uk"
+    customer_domain = "https://southend.gov.uk"
-    installer_name = "WARM FRONT"
+    installer_name = "J & J CRUMP"
     asset_list_filepath = (
-        "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Calico/Hubspot/07.04 CALICO - Final List - "
+        "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Southend/July 2025 Programme/SOUTHEND - RYAN - "
-        "Standardised.xlsx"
+        "Standardised 2.xlsx"
     )
-    asset_list_sheet_name = "Final Route March"
+    asset_list_sheet_name = "Standardised Asset List"
     asset_list_header = 0
 
     contact_details_filepath = None
@@ -107,7 +107,7 @@ def app():
         raise ValueError("FIX MEEE")
 
     if pd.isnull(asset_list.hubspot_data['Deal Stage <DEAL dealstage>']).any():
-        raise ValueError("Warning: Some rows have missing project codes. These will not be uploaded to HubSpot.")
+        raise ValueError("Warning: Some rows have missing deal stage. These will not be uploaded to HubSpot.")
 
     # Just store locally
     asset_list.hubspot_data.to_csv(output_filepath, index=False, encoding="utf-8-sig")

asset_list/mappings/built_form.py

@@ -384,6 +384,7 @@ BUILT_FORM_MAPPINGS = {
     'Cottage Flat': 'ground floor',
     'Maisonette Over Shop': 'mid-floor',
     'Medium Rise Flat': 'mid-floor',
-    'Maisonette Medium Rise': 'unknown'
+    'Maisonette Medium Rise': 'unknown',
+    'End-terraced house': 'end-terrace'
 
 }

backend/app/plan/schemas.py

@@ -102,3 +102,9 @@ class PlanTriggerRequest(BaseModel):
     # If true, before optimising the engine will select a slightly larger package, to account for the SAP 10 causing
     # scores to drop by a few points
     simulate_sap_10: Optional[bool] = False
+
+    # Add in optional fields which describe the format of the asset list being used
+    
+    file_type: Optional[Literal["csv", "xlsx"]] = None,
+    file_format: Optional[Literal["domna_asset_list"]] = None,
+    sheet_name: Optional[str] = None

etl/customers/abri/abs_rates.py

@@ -0,0 +1,12 @@
+import pandas as pd
+
+project = pd.read_excel(
+    "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Abri/Abs Rates/Desktop ABRI data - Standardised After "
+    "Programmes.xlsx"
+)
+
+cavity = project[project["cavity_reason"].isin(
+    ["EPC Shows Empty Cavity: SAP Rating 54 or less", "EPC Shows Empty Cavity: SAP Rating 55-68"]
+)]
+
+# Pull the data

etl/find_my_epc/RetrieveFindMyEpc.py

@@ -681,7 +681,9 @@ class RetrieveFindMyEpc:
             ],
             "High heat retention storage heaters and dual rate meter": [
                 "high_heat_retention_storage_heater"
-            ]
+            ],
+            "Increase loft insulation to 250mm": ["loft_insulation"],
+            "Solar photovoltaics panels, 25% of roof area": ["solar_pv"],
         }
 
         survey = True