From bdd6171626e85689d430180520e84f507b6010e2 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Fri, 26 Jul 2024 15:07:23 +0100 Subject: [PATCH] Added mapping of age band --- .../functions/energy_assessment_functions.py | 2 +- backend/app/db/models/energy_assessments.py | 37 +++++++- backend/app/plan/router.py | 85 +++++++++++++------ etl/xml_survey_extraction/XmlParser.py | 23 ++++- 4 files changed, 117 insertions(+), 30 deletions(-) diff --git a/backend/app/db/functions/energy_assessment_functions.py b/backend/app/db/functions/energy_assessment_functions.py index 2d6a611e..45fb2b8b 100644 --- a/backend/app/db/functions/energy_assessment_functions.py +++ b/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 diff --git a/backend/app/db/models/energy_assessments.py b/backend/app/db/models/energy_assessments.py index 86230c00..efcbc26c 100644 --- a/backend/app/db/models/energy_assessments.py +++ b/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": {}} diff --git a/backend/app/plan/router.py b/backend/app/plan/router.py index c73aff7e..175561e4 100644 --- a/backend/app/plan/router.py +++ b/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( diff --git a/etl/xml_survey_extraction/XmlParser.py b/etl/xml_survey_extraction/XmlParser.py index 90a51ae6..522cb899 100644 --- a/etl/xml_survey_extraction/XmlParser.py +++ b/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') ],