Added mapping of age band

backend/app/db/functions/energy_assessment_functions.py

@@ -55,7 +55,7 @@ def get_latest_assessment_by_uprn(session: Session, uprn: int) -> Optional[Energ
         # Query the EnergyAssessment model, filter by uprn, order by inspection_date in descending order
         latest_assessment = session.query(EnergyAssessment).filter_by(uprn=uprn).order_by(
             desc(EnergyAssessment.inspection_date)).first()
-        return latest_assessment.to_dict() if latest_assessment else {}
+        return latest_assessment.to_dict() if latest_assessment else latest_assessment.empty_response()
     except Exception as e:
         print(f"An error occurred: {e}")
         return None

backend/app/db/models/energy_assessments.py

@@ -120,8 +120,43 @@ class EnergyAssessment(Base):
     cylinder_insulation_thickness = Column(Integer)
     cylinder_thermostat = Column(Boolean)
 
+    EPC_KEYS = [
+        'low_energy_fixed_light_count', 'address', 'uprn_source', 'floor_height', 'heating_cost_potential',
+        'unheated_corridor_length', 'hot_water_cost_potential', 'construction_age_band', 'potential_energy_rating',
+        'mainheat_energy_eff', 'windows_env_eff', 'lighting_energy_eff', 'environment_impact_potential', 'glazed_type',
+        'heating_cost_current', 'address3', 'mainheatcont_description', 'sheating_energy_eff', 'property_type',
+        'local_authority_label', 'fixed_lighting_outlets_count', 'energy_tariff', 'mechanical_ventilation',
+        'hot_water_cost_current', 'county', 'postcode', 'solar_water_heating_flag', 'constituency',
+        'co2_emissions_potential', 'number_heated_rooms', 'floor_description', 'energy_consumption_potential',
+        'local_authority', 'built_form', 'number_open_fireplaces', 'windows_description', 'glazed_area',
+        'inspection_date', 'mains_gas_flag', 'co2_emiss_curr_per_floor_area', 'address1', 'heat_loss_corridor',
+        'flat_storey_count', 'constituency_label', 'roof_energy_eff', 'total_floor_area', 'building_reference_number',
+        'environment_impact_current', 'co2_emissions_current', 'roof_description', 'floor_energy_eff',
+        'number_habitable_rooms', 'address2', 'hot_water_env_eff', 'posttown', 'mainheatc_energy_eff', 'main_fuel',
+        'lighting_env_eff', 'windows_energy_eff', 'floor_env_eff', 'sheating_env_eff', 'lighting_description',
+        'roof_env_eff', 'walls_energy_eff', 'photo_supply', 'lighting_cost_potential', 'mainheat_env_eff',
+        'multi_glaze_proportion', 'main_heating_controls', 'lodgement_datetime', 'flat_top_storey',
+        'current_energy_rating', 'secondheat_description', 'walls_env_eff', 'transaction_type', 'uprn',
+        'current_energy_efficiency', 'energy_consumption_current', 'mainheat_description', 'lighting_cost_current',
+        'lodgement_date', 'extension_count', 'mainheatc_env_eff', 'lmk_key', 'wind_turbine_count', 'tenure',
+        'floor_level', 'potential_energy_efficiency', 'hot_water_energy_eff', 'low_energy_lighting',
+        'walls_description', 'hotwater_description'
+    ]
+
     def to_dict(self):
         """
         Convert the SQLAlchemy object to a dictionary.
         """
-        return {column.name: getattr(self, column.name) for column in self.__table__.columns}
+
+        epc = {key.replace("_", "-"): getattr(self, key) for key in self.EPC_KEYS}
+        # Get everything else
+        additional = {
+            column.name: getattr(self, column.name)
+            for column in self.__table__.columns if column.name not in self.EPC_KEYS
+        }
+
+        return {"epc": epc, "additional": additional}
+
+    @staticmethod
+    def empty_response():
+        return {"epc": {}, "additional": {}}

backend/app/plan/router.py

@@ -220,6 +220,60 @@ def extract_portfolio_aggregation_data(
     return aggregation_data
 
 
+def create_epc_records(epc_searcher: SearchEpc, energy_assessment: dict):
+    """
+    This function will set up with epc_records dictionary with the newest EPC, the full SAP EPC and the older EPCs
+    and will factor in an energy assessment that we have performed for a client.
+    :param epc_searcher: An instance of the SearchEpc class
+    :param energy_assessment: The energy assessment we have performed. If we have not performed an energy assessment,
+                              this should be an empty response as defined by the models's
+                              EnergyAssessment.empty_response() method
+    """
+
+    if not energy_assessment["epc"]:
+        return {
+            'original_epc': epc_searcher.newest_epc.copy(),
+            'full_sap_epc': epc_searcher.full_sap_epc.copy(),
+            'old_data': epc_searcher.older_epcs.copy(),
+        }
+
+    epc = energy_assessment["epc"]
+    energy_assessment_date = epc["inspection-date"].strftime("%Y-%m-%d")
+
+    # We check if the energy assessment is newer than the newest EPC
+    if pd.to_datetime(energy_assessment_date) > pd.to_datetime(epc_searcher.newest_epc["inspection-date"]):
+        # In this case, our energy assessment is newer than the EPCs available for this property
+        return {
+            "original_epc": epc,
+            "full_sap_epc": epc_searcher.full_sap_epc.copy(),
+            "old_data": epc_searcher.older_epcs.copy() + [epc_searcher.newest_epc.copy()]
+        }
+
+    # We check if the EPC we have produced is contained in the set of EPCs done for the property
+    # We do this based on inspection-date and SAP
+    epc_in_historicals = [
+        x for x in epc_searcher.older_epcs + [epc_searcher.newest_epc]
+        if x["inspection-date"] == energy_assessment_date and
+                   x["current-energy-efficiency"] == epc["current-energy-efficiency"]
+    ]
+
+    if epc_in_historicals:
+        # Then the EPC we have produced is already in the set of EPCs, and our EPC is older than the newest
+        return {
+            "original_epc": epc_searcher.newest_epc.copy(),
+            "full_sap_epc": epc_searcher.full_sap_epc.copy(),
+            "old_data": epc_searcher.older_epcs.copy()
+        }
+
+    # In this case, our EPC is older than the newest publically avaible one, but is not contained in
+    # the historicals, so it can't have been lodged, so we include it in the old data
+    return {
+        'original_epc': epc_searcher.newest_epc.copy(),
+        'full_sap_epc': epc_searcher.full_sap_epc.copy(),
+        'old_data': epc_searcher.older_epcs.copy() + [epc],
+    }
+
+
 router = APIRouter(
     prefix="/plan",
     tags=["plan"],
@@ -285,7 +339,7 @@ async def trigger_plan(body: PlanTriggerRequest):
             epc_searcher.find_property(skip_os=True)
 
             # We check for an energy assessment we have performed on this property:
-            energy_assessment = get_latest_assessment_by_uprn(session, uprn)
+            energy_assessment = get_latest_assessment_by_uprn(session, uprn if uprn is not None else epc_searcher.uprn)
 
             # Create a record in db
             property_id, is_new = create_property(
@@ -302,32 +356,9 @@ async def trigger_plan(body: PlanTriggerRequest):
                 heat_demand_target=None
             )
 
-            epc_records = {
-                'original_epc': epc_searcher.newest_epc.copy(),
-                'full_sap_epc': epc_searcher.full_sap_epc.copy(),
-                'old_data': epc_searcher.older_epcs.copy(),
-            }
-
-            patch = next((
-                x for x in patches if (x["address"] == config["address"]) and (x["postcode"] == config["postcode"])
-            ), {})
-            epc_records = patch_epc(patch, epc_records)
-
-            prepared_epc = EPCRecord(
-                epc_records=epc_records,
-                run_mode="newdata",
-                cleaning_data=cleaning_data
-            )
-
-            property_already_installed = next((
-                x for x in already_installed if
-                (x["address"] == config["address"]) and (x["postcode"] == config["postcode"])
-            ), {})
-
-            property_non_invasive_recommendations = next((
-                x for x in non_invasive_recommendations if
-                (x["address"] == config["address"]) and (x["postcode"] == config["postcode"])
-            ), {})
+            # If we have an energy assessment in place, that is newer than all of the previous EPCs, we use that.
+            # Otherwise, we use the newest EPC
+            epc_records = create_epc_records(epc_searcher, energy_assessment)
 
             input_properties.append(
                 Property(

etl/xml_survey_extraction/XmlParser.py

@@ -72,6 +72,25 @@ class XmlParser:
 
     floor_dimensions = None
 
+    # The age band lookup is based on the country code
+    AGE_BAND_LOOKUP = {
+        # England & Wales
+        "EAW": {
+            "A": "England and Wales: before 1900",
+            "B": "England and Wales: 1900-1929",
+            "C": "England and Wales: 1930-1949",
+            "D": "England and Wales: 1950-1966",
+            "E": "England and Wales: 1967-1975",
+            "F": "England and Wales: 1976-1982",
+            "G": "England and Wales: 1983-1990",
+            "H": "England and Wales: 1991-1995",
+            "I": "England and Wales: 1996-2002",
+            "J": "England and Wales: 2003-2006",
+            "K": "England and Wales: 2007-2011",
+            "L": "England and Wales: 2012 onwards",
+        }
+    }
+
     RATINGS_MAP = {
         "0": "N/A",
         "1": "Very Poor",
@@ -205,7 +224,9 @@ class XmlParser:
             **self.get_sap(),
             **self.get_property_address(),
             "low-energy-fixed-light-count": self.get_node_value('Low-Energy-Fixed-Lighting-Outlets-Count'),
-            "construction-age-band": self.get_node_value('Construction-Age-Band'),
+            "construction-age-band": self.AGE_BAND_LOOKUP[
+                self.get_node_value('Country-Code')
+            ][self.get_node_value('Construction-Age-Band')],
             "mainheat-energy-eff": self.RATINGS_MAP[
                 self.get_property_summary_value('Main-Heating', 'Energy-Efficiency-Rating')
             ],