adding maps to format_recommendations

etl/customers/ksquared/Wave3 Modelling.py

@@ -12,9 +12,10 @@ load_dotenv(dotenv_path="backend/.env")
 EPC_AUTH_TOKEN = os.getenv("EPC_AUTH_TOKEN")
 USER_ID = 8
 PORTFOLIO_ID = 117
+CAHA_PORTFOLIO_ID = 118
 
 
-def app():
+def hornsey():
     """
     This script prepares the asset lists for the additional housing associations, CAHA and Hornsey Housing Trust,
     that are forming a consortium led by AIHA
@@ -156,3 +157,71 @@ def app():
         "exclusions": ["boiler_upgrade"]
     }
     print(body)
+
+
+def caha():
+    caha_asset_list = pd.read_excel(
+        "/Users/khalimconn-kowlessar/Documents/hestia/Customers/AIHA/Copy of AIHA - WHSHF Wave 3 bid - Consortium "
+        "member properties - CAHA.xlsx",
+        sheet_name="Ksquared-All units information",
+        header=3
+    )
+
+    caha_asset_list = caha_asset_list.iloc[1:]
+    # Fill NA values with empty strings
+    caha_asset_list = caha_asset_list.fillna("")
+    caha_asset_list["Address letter or number"] = caha_asset_list["Address letter or number"].astype(
+        str
+    ).str.strip()
+
+    # We Add POstcode as it wasn't populated - split on space and take the last two entries and re-concatenate on space
+    caha_asset_list["Postcode"] = caha_asset_list["Street address"].str.split(" ").str[-2:].str.join(" ")
+    # Take just the columns we need
+    caha_asset_list = caha_asset_list[["Address letter or number", "Street address", "Postcode"]]
+
+    for col in ["Address letter or number", "Street address", "Postcode"]:
+        caha_asset_list[col] = caha_asset_list[col].str.replace("  ", " ")
+
+    # Pull the data from find my epc
+    remap = {
+        "Flat A, 50 Talbot Road N6 4QP": "50a Talbot Road"
+    }
+    extracted_data = []
+    asset_list = []
+    for _, home in tqdm(caha_asset_list.iterrows(), total=len(caha_asset_list)):
+        unit_number = home["Address letter or number"]
+        street = home["Street address"]
+        postcode = home["Postcode"]
+        address = ", ".join([x for x in [unit_number, street] if x])
+        address = remap.get(address, address)
+        address = address.replace(postcode, "").strip()
+
+        find_epc_searcher = RetrieveFindMyEpc(address=address, postcode=postcode)
+        find_epc_data = find_epc_searcher.retrieve_newest_find_my_epc_data()
+        time.sleep(0.5)
+        # We need uprn
+        searcher = SearchEpc(
+            address1=address,
+            postcode=postcode,
+            auth_token=EPC_AUTH_TOKEN,
+            os_api_key="",
+            full_address=address,
+        )
+        searcher.find_property(skip_os=True)
+        newest_epc = searcher.newest_epc
+
+        extracted_data.append(
+            {
+                "uprn": newest_epc["uprn"],
+                **find_epc_data,
+            }
+        )
+
+        asset_list.append(
+            {
+                "uprn": newest_epc["uprn"],
+                "address": home["Address letter or number"],
+                "postcode": home["Postcode"],
+                "property_type": newest_epc["property-type"],
+            }
+        )

etl/find_my_epc/RetrieveFindMyEpc.py

@@ -225,6 +225,12 @@ class RetrieveFindMyEpc:
                 "roomstat_programmer_trvs", "time_temperature_zone_control"
             ],
             "Low energy lighting": ["low_energy_lighting"],
+            "Increase loft insulation to 270 mm": ["loft_insulation"],
+            "Heating controls (thermostatic radiator valves)": [
+                "roomstat_programmer_trvs", "time_temperature_zone_control"
+            ],
+            "Solar water heating": ["solar_water_heating"],
+            "Solar photovoltaic panels, 2.5 kWp": ["solar_pv"],
         }
 
         formatted_recommendations = []