From 5fa6289b4414bb394a6119892a11345587369488 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Thu, 5 Feb 2026 11:19:03 +0000 Subject: [PATCH] setting up handler with example event --- backend/onboarders/base.py | 14 +- backend/onboarders/epc_descriptions.py | 513 ------------------ backend/onboarders/factory.py | 10 + backend/onboarders/handler.py | 33 ++ .../onboarders/mappings/parity/age_band.py | 2 +- backend/onboarders/mappings/parity/roof.py | 358 ++++++++++++ backend/onboarders/mappings/parity/walls.py | 155 ++++++ backend/onboarders/parity.py | 23 +- 8 files changed, 577 insertions(+), 531 deletions(-) delete mode 100644 backend/onboarders/epc_descriptions.py create mode 100644 backend/onboarders/factory.py diff --git a/backend/onboarders/base.py b/backend/onboarders/base.py index b90f5fc4..0e2351bd 100644 --- a/backend/onboarders/base.py +++ b/backend/onboarders/base.py @@ -1,5 +1,5 @@ import pandas as pd -from utils.s3 import read_from_s3 +from utils.s3 import read_from_s3, read_excel_from_s3 class OnboarderBase: @@ -37,8 +37,16 @@ class OnboarderBase: landlord_property_type: str = "landlord_property_type" landlord_built_form: str = "landlord_built_form" - def read_s3(self, bucket_name: str, file_name: str): - self.data = read_from_s3(bucket_name=bucket_name, s3_file_name=file_name) + def read_s3(self, bucket_name: str, file_name: str, **kwargs): + if kwargs.get("format") == "xlsx": + self.data = read_excel_from_s3( + bucket_name=bucket_name, + file_key=file_name, + sheet_name=kwargs.get("sheet_name"), + header_row=kwargs.get("header_row", 0) + ) + else: + self.data = read_from_s3(bucket_name=bucket_name, s3_file_name=file_name) def write(self): pass diff --git a/backend/onboarders/epc_descriptions.py b/backend/onboarders/epc_descriptions.py deleted file mode 100644 index 78cc57c1..00000000 --- a/backend/onboarders/epc_descriptions.py +++ /dev/null @@ -1,513 +0,0 @@ -import pandas as pd -from enum import Enum -from collections.abc import Mapping -from typing import Callable, Union -from datatypes.epc.construction_age_band import EpcConstructionAgeBand -from datatypes.epc.efficiency import EpcEfficiency -from datatypes.epc.walls import EpcWallDescriptions -from datatypes.epc.roof import EpcRoofDescriptions - - -def cavity_filled_efficiency(age_band: EpcConstructionAgeBand) -> EpcEfficiency: - """" - Maps cavity filled to efficiency based on construction age band. - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - if age_band in { - EpcConstructionAgeBand.from_2023_onwards - }: - return EpcEfficiency.VERY_GOOD - - return EpcEfficiency.GOOD - - -def internal_external_insulation_efficiency( - age_band: EpcConstructionAgeBand, -) -> EpcEfficiency: - """ - Maps: - - cavity unfilled with internal/external insulation to efficiency based on construction age band. We assumed - based on 100mm insulation - - solid brick with internal/external insulation to efficiency based on construction age band. We assumed - based on 100mm insulation - - system built with internal/external insulation to efficiency based on construction age band. We assumed - based on 100mm insulation - - All of these wall types have the same behaviour in elmhurst - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - if age_band in { - EpcConstructionAgeBand.from_1983_to_1990, - EpcConstructionAgeBand.from_1991_to_1995, - EpcConstructionAgeBand.from_1996_to_2002, - EpcConstructionAgeBand.from_2003_to_2006, - EpcConstructionAgeBand.from_2007_to_2011, - EpcConstructionAgeBand.from_2012_to_2022, - EpcConstructionAgeBand.from_2023_onwards, - }: - return EpcEfficiency.VERY_GOOD - - return EpcEfficiency.GOOD - - -def timber_granite_sandstone_internal_external_efficiency(age_band: EpcConstructionAgeBand) -> EpcEfficiency: - """" - Maps: - - timber frame with internal/external wall insulation to efficiency based on construction age band. - - sandstone/limestone with internal/external wall insulation to efficiency based on construction age band. - - granite/whinstone with internal/external wall insulation to efficiency based on construction age band. - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - if age_band in { - EpcConstructionAgeBand.from_2023_onwards - }: - return EpcEfficiency.VERY_GOOD - - return EpcEfficiency.GOOD - - -WallEfficiencyRule = Union[ - EpcEfficiency, - Callable[[EpcConstructionAgeBand, int | None], EpcEfficiency], -] - -WALL_DESCRIPTION_EFFICIENCIES: Mapping[EpcWallDescriptions, WallEfficiencyRule] = { - # Note: all function mappings have been defined based on Elmhurst - # Cavity - # value mappings - EpcWallDescriptions.cavity_no_insulation_assumed: EpcEfficiency.POOR, - EpcWallDescriptions.cavity_partial_insulated_assumed: EpcEfficiency.AVERAGE, - EpcWallDescriptions.cavity_insulated_assumed: EpcEfficiency.GOOD, - EpcWallDescriptions.cavity_filled_plus_internal: EpcEfficiency.VERY_GOOD, - EpcWallDescriptions.cavity_filled_plus_external: EpcEfficiency.VERY_GOOD, - # function mappings - EpcWallDescriptions.cavity_filled_cavity: cavity_filled_efficiency, - EpcWallDescriptions.cavity_internal_insulation: internal_external_insulation_efficiency, - EpcWallDescriptions.cavity_external_insulation: internal_external_insulation_efficiency, - - # Solid brick - # value mappings - EpcWallDescriptions.solid_brick_no_insulation_assumed: EpcEfficiency.POOR, - EpcWallDescriptions.solid_brick_partial_insulated_assumed: EpcEfficiency.AVERAGE, - EpcWallDescriptions.solid_brick_insulated_assumed: EpcEfficiency.GOOD, - # function mappings - EpcWallDescriptions.solid_brick_internal_insulation: internal_external_insulation_efficiency, - EpcWallDescriptions.solid_brick_external_insulation: internal_external_insulation_efficiency, - - # System - # value mappings - EpcWallDescriptions.system_no_insulation_assumed: EpcEfficiency.POOR, - EpcWallDescriptions.system_partial_insulated_assumed: EpcEfficiency.AVERAGE, - EpcWallDescriptions.system_insulated_assumed: EpcEfficiency.GOOD, - # function mappings - EpcWallDescriptions.system_internal_insulation: internal_external_insulation_efficiency, - EpcWallDescriptions.system_external_insulation: internal_external_insulation_efficiency, - - # Timber frame - # value mappings - EpcWallDescriptions.timber_frame_no_insulation_assumed: EpcEfficiency.POOR, - EpcWallDescriptions.timber_frame_partial_insulated_assumed: EpcEfficiency.AVERAGE, - EpcWallDescriptions.timber_frame_insulated_assumed: EpcEfficiency.GOOD, - # function mappings - EpcWallDescriptions.timber_frame_internal_insulation: timber_granite_sandstone_internal_external_efficiency, - EpcWallDescriptions.timber_frame_external_insulation: timber_granite_sandstone_internal_external_efficiency, - - # Granite / whinstone - EpcWallDescriptions.granite_whinstone_no_insulation_assumed: EpcEfficiency.VERY_POOR, - EpcWallDescriptions.granite_whinstone_partial_insulated_assumed: EpcEfficiency.AVERAGE, - EpcWallDescriptions.granite_whinestone_insulated_assumed: EpcEfficiency.GOOD, - # function mappings - EpcWallDescriptions.granite_whinstone_internal_insulation: timber_granite_sandstone_internal_external_efficiency, - EpcWallDescriptions.granite_whinstone_external_insulation: timber_granite_sandstone_internal_external_efficiency, - - # Sandstone / limestone - EpcWallDescriptions.sandstone_limestone_no_insulation_assumed: EpcEfficiency.VERY_POOR, - EpcWallDescriptions.sandstone_limestone_partial_insulated_assumed: EpcEfficiency.AVERAGE, - EpcWallDescriptions.sandstone_limestone_insulated_assumed: EpcEfficiency.GOOD, - # function mappings - EpcWallDescriptions.sandstone_limestone_internal_insulation: timber_granite_sandstone_internal_external_efficiency, - EpcWallDescriptions.sandstone_limestone_external_insulation: timber_granite_sandstone_internal_external_efficiency, - - # Cob (special case) - EpcWallDescriptions.cob_as_built_average: EpcEfficiency.AVERAGE, - EpcWallDescriptions.cob_as_built_good: EpcEfficiency.GOOD, - - # Unknown mappings which are unhandled - EpcWallDescriptions.cavity_as_built_unknown: EpcEfficiency.NA, - EpcWallDescriptions.solid_brick_as_built_unknown: EpcEfficiency.NA, - EpcWallDescriptions.system_as_built_unknown: EpcEfficiency.NA, - EpcWallDescriptions.timber_frame_as_built_unknown: EpcEfficiency.NA, - EpcWallDescriptions.granite_as_built_unknown: EpcEfficiency.NA, - EpcWallDescriptions.sandstone_as_built_unknown: EpcEfficiency.NA, - EpcWallDescriptions.cob_as_built_unknown: EpcEfficiency.NA, - -} - - -def resolve_wall_efficiency( - description: EpcWallDescriptions, - age_band: EpcConstructionAgeBand, -) -> EpcEfficiency: - rule = WALL_DESCRIPTION_EFFICIENCIES[description] - - if isinstance(rule, EpcEfficiency): - return rule - - return rule(age_band) - - -RoofEfficiencyRule = Union[ - EpcEfficiency, - Callable[[EpcConstructionAgeBand, int | None], EpcEfficiency], -] - - -def flat_insulated_efficiency_age_band(age_band: EpcConstructionAgeBand) -> EpcEfficiency: - """ - before 1900, 1900-1929, 1930-1949, 1950-1966, 1967-1975 -> Pitched, no insulation, Very Poor - 1976-1982 -> Pitched, limited insulation, Poor - 1983-1990, to 1996-2002 Pitched, insulated, Average - 2003 - 2006, 2012-2022 -> Pitched, insulated, Good - 2023 onwards -> Pitched, insulated, Very Good - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - - start_year = age_band.start_year() - if start_year >= 2023: - return EpcEfficiency.VERY_GOOD - - if start_year >= 2003: - return EpcEfficiency.GOOD - - if start_year >= 1983: - return EpcEfficiency.AVERAGE - - if start_year >= 1976: - return EpcEfficiency.POOR - - return EpcEfficiency.VERY_POOR - - -def flat_insulated_efficiency_thickness(insulation_thickness: int | None) -> EpcEfficiency: - """ - 12mm -> Very Poor - 25mm - 50mm -> Poor - 75mm - 125mm -> Pitched, insulated, average - 150mm - 250mm -> good - 270mm+ -> very good - :param insulation_thickness: Insulation thickness in mm - :return: EpcEfficiency - """ - - if insulation_thickness is None: - raise ValueError("Insulation thickness is required for flat insulated efficiency calculation") - - if insulation_thickness >= 270: - return EpcEfficiency.VERY_GOOD - - if 150 <= insulation_thickness <= 250: - return EpcEfficiency.GOOD - - if 75 <= insulation_thickness <= 125: - return EpcEfficiency.AVERAGE - - if 25 <= insulation_thickness <= 50: - return EpcEfficiency.POOR - - return EpcEfficiency.VERY_POOR - - -def flat_efficiency(insulation_thickness: int | None, age_band: EpcConstructionAgeBand) -> EpcEfficiency: - """ - Combines both age band and insulation thickness to determine flat roof efficiency. - :param insulation_thickness: Insulation thickness in mm - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - if insulation_thickness is not None: - return flat_insulated_efficiency_thickness(insulation_thickness) - - return flat_insulated_efficiency_age_band(age_band) - - -def loft_insulated_efficiency(age_band: EpcConstructionAgeBand) -> EpcEfficiency: - """ - 2023 onwards -> Very Good - 2012-2022 -> Very Good - 2007-2011 -> Very Good - 2003-2006 -> Very Good - 1996-2002 -> Good - 1991-1995 -> Good - 1983-1990 -> Average - 1976-1982 -> Average - 1967-1975 -> Average - 1950-1966 -> Average - 1930-1949 -> Average - 1900-1929 -> Average - before 1900 -> Average - :param age_band: Input age band, EpcConstructionAgeBand - :return: EpcEfficiency - """ - year = age_band.start_year() - if year >= 2003: - return EpcEfficiency.VERY_GOOD - if year >= 1991: - return EpcEfficiency.GOOD - - return EpcEfficiency.AVERAGE - - -def thatched_efficiency_age_band(age_band: EpcConstructionAgeBand) -> EpcEfficiency: - """ - Maps thatched roof efficiency based on construction age band. - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - year = age_band.start_year() - if year >= 2023: - return EpcEfficiency.VERY_GOOD - if year >= 2003: - return EpcEfficiency.GOOD - - return EpcEfficiency.AVERAGE - - -def thatched_efficiency_thickness(insulation_thickness: int | None) -> EpcEfficiency: - """ - Maps thatched roof efficiency based on insulation thickness. - :param insulation_thickness: Insulation thickness in mm - :return: EpcEfficiency - """ - if insulation_thickness is None: - raise ValueError("Insulation thickness is required for thatched efficiency calculation") - - if insulation_thickness >= 175: - return EpcEfficiency.VERY_GOOD - - if insulation_thickness >= 25: - return EpcEfficiency.GOOD - - return EpcEfficiency.AVERAGE - - -def thatched_efficiency( - insulation_thickness: int | None, - age_band: EpcConstructionAgeBand, -) -> EpcEfficiency: - """ - Combines both age band and insulation thickness to determine thatched roof efficiency. - :param insulation_thickness: Insulation thickness in mm - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - if insulation_thickness is not None: - return thatched_efficiency_thickness(insulation_thickness) - - return thatched_efficiency_age_band(age_band) - - -def sloping_ceiling_efficiency_age_band(age_band: EpcConstructionAgeBand) -> EpcEfficiency: - """ - Maps sloping ceiling roof efficiency based on construction age band. - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - year = age_band.start_year() - if year >= 2023: - return EpcEfficiency.VERY_GOOD - if year >= 2003: - return EpcEfficiency.GOOD - if year >= 1983: - return EpcEfficiency.AVERAGE - if year >= 1976: - return EpcEfficiency.POOR - - return EpcEfficiency.VERY_POOR - - -def sloping_ceiling_efficiency_thickness(insulation_thickness: int | None) -> EpcEfficiency: - """ - Maps sloping ceiling roof efficiency based on insulation thickness. - :param insulation_thickness: Insulation thickness in mm - :return: EpcEfficiency - """ - if insulation_thickness is None: - raise ValueError("Insulation thickness is required for sloping ceiling efficiency calculation") - - if insulation_thickness >= 270: - return EpcEfficiency.VERY_GOOD - - if insulation_thickness >= 150: - return EpcEfficiency.GOOD - - if insulation_thickness >= 75: - return EpcEfficiency.AVERAGE - - if insulation_thickness >= 25: - return EpcEfficiency.POOR - - return EpcEfficiency.VERY_POOR - - -def sloping_ceiling_efficiency( - insulation_thickness: int | None, - age_band: EpcConstructionAgeBand, -) -> EpcEfficiency: - """ - Combines both age band and insulation thickness to determine sloping ceiling roof efficiency. - :param insulation_thickness: Insulation thickness in mm - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - if insulation_thickness is not None: - return sloping_ceiling_efficiency_thickness(insulation_thickness) - - return sloping_ceiling_efficiency_age_band(age_band) - - -def loft_insulated_at_rafters_efficiency_thickness(insulation_thickness: int | None) -> EpcEfficiency: - """ - 400mm, 350mm = very good - 200-300mm = good - 125-175 = average - 50-100 = poor - 25 and below= very poor - :return: - """ - if insulation_thickness is None: - raise ValueError("Insulation thickness is required for loft insulated at rafters efficiency calculation") - - if insulation_thickness >= 350: - return EpcEfficiency.VERY_GOOD - - if insulation_thickness >= 200: - return EpcEfficiency.GOOD - - if insulation_thickness >= 125: - return EpcEfficiency.AVERAGE - - if insulation_thickness >= 50: - return EpcEfficiency.POOR - - return EpcEfficiency.VERY_POOR - - -def loft_insulated_at_rafters_efficiency_age_band(age_band: EpcConstructionAgeBand) -> EpcEfficiency: - """ - # 2023 onwards -> Very Good - # 2003-2006, 2012-2022 -> Good - # 1983 - 1990, 1996-2002 -> Average - # 1976-1982 -> Poor - # 1967-1975 and earlier bands -> Very Poor - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - year = age_band.start_year() - if year >= 2023: - return EpcEfficiency.VERY_GOOD - if year >= 2003: - return EpcEfficiency.GOOD - if year >= 1983: - return EpcEfficiency.AVERAGE - if year >= 1976: - return EpcEfficiency.POOR - - return EpcEfficiency.VERY_POOR - - -def loft_insulated_at_rafters_efficiency( - insulation_thickness: int | None, - age_band: EpcConstructionAgeBand, -) -> EpcEfficiency: - """ - Combines both age band and insulation thickness to determine loft insulated at rafters roof efficiency. - :param insulation_thickness: Insulation thickness in mm - :param age_band: EpcConstructionAgeBand - :return: EpcEfficiency - """ - if insulation_thickness is not None: - return loft_insulated_at_rafters_efficiency_thickness(insulation_thickness) - - return loft_insulated_at_rafters_efficiency_age_band(age_band) - - -ROOF_DESCRIPTION_EFFICIENCIES: Mapping[EpcRoofDescriptions, RoofEfficiencyRule] = { - # Flat roof - EpcRoofDescriptions.flat_no_insulation: EpcEfficiency.VERY_POOR, - EpcRoofDescriptions.flat_limited_insulation: flat_efficiency, - EpcRoofDescriptions.flat_insulated: flat_efficiency, - - # Loft: - # value mappings - EpcRoofDescriptions.loft_12mm_insulation: EpcEfficiency.VERY_POOR, - EpcRoofDescriptions.loft_25mm_insulation: EpcEfficiency.POOR, - EpcRoofDescriptions.loft_50mm_insulation: EpcEfficiency.POOR, - EpcRoofDescriptions.loft_75mm_insulation: EpcEfficiency.AVERAGE, - EpcRoofDescriptions.loft_100mm_insulation: EpcEfficiency.AVERAGE, - EpcRoofDescriptions.loft_125mm_insulation: EpcEfficiency.AVERAGE, - EpcRoofDescriptions.loft_150mm_insulation: EpcEfficiency.GOOD, - EpcRoofDescriptions.loft_175mm_insulation: EpcEfficiency.GOOD, - EpcRoofDescriptions.loft_200mm_insulation: EpcEfficiency.GOOD, - EpcRoofDescriptions.loft_250mm_insulation: EpcEfficiency.GOOD, - EpcRoofDescriptions.loft_270mm_insulation: EpcEfficiency.VERY_GOOD, - EpcRoofDescriptions.loft_300mm_insulation: EpcEfficiency.VERY_GOOD, - EpcRoofDescriptions.loft_350mm_insulation: EpcEfficiency.VERY_GOOD, - EpcRoofDescriptions.loft_400mm_plus_insulation: EpcEfficiency.VERY_GOOD, - EpcRoofDescriptions.pitched_no_insulation: EpcEfficiency.VERY_POOR, - # function mappings - EpcRoofDescriptions.pitched_insulated_assumed: loft_insulated_efficiency, - - # Loft af rafters - EpcRoofDescriptions.loft_insulated_at_rafters: loft_insulated_at_rafters_efficiency, - - # Another dwelling above - EpcRoofDescriptions.another_dwelling_above: EpcEfficiency.NA, - - # Thatched - EpcRoofDescriptions.thatched: thatched_efficiency, - EpcRoofDescriptions.thatched_with_additional_insulation: thatched_efficiency, - - # Sloping ceiling - EpcRoofDescriptions.sloping_pitched_insulated: sloping_ceiling_efficiency, - EpcRoofDescriptions.sloping_pitched_limited_insulation: sloping_ceiling_efficiency, - EpcRoofDescriptions.sloping_pitched_no_insulation: EpcEfficiency.VERY_POOR, - -} - - -def resolve_roof_efficiency( - description: EpcRoofDescriptions, - age_band: EpcConstructionAgeBand | None, - insulation_thickness: int | None, -) -> EpcEfficiency: - """ - Resolve roof efficiency from description + age band + insulation thickness. - """ - - # Unknown / holding descriptions → efficiency unknown - if description in description.unknown_descriptions: - return EpcEfficiency.NA - - rule = ROOF_DESCRIPTION_EFFICIENCIES.get(description) - - if rule is None: - return EpcEfficiency.NA - - # Fixed efficiency - if isinstance(rule, EpcEfficiency): - return rule - - # Callable rule - if age_band is None or pd.isnull(age_band): - return EpcEfficiency.NA - - try: - # Try (thickness, age_band) - return rule(insulation_thickness, age_band) - except TypeError: - # Fallback to (age_band) - return rule(age_band) diff --git a/backend/onboarders/factory.py b/backend/onboarders/factory.py new file mode 100644 index 00000000..13dd5505 --- /dev/null +++ b/backend/onboarders/factory.py @@ -0,0 +1,10 @@ +from onboarders.parity import ParityOnboarder + + +class OnboarderFactory: + @staticmethod + def create_onboarder(onboarder_type): + if onboarder_type == "parity": + return ParityOnboarder + + raise ValueError(f"Unknown onboarder type: {onboarder_type}") diff --git a/backend/onboarders/handler.py b/backend/onboarders/handler.py index e69de29b..0c38e4d9 100644 --- a/backend/onboarders/handler.py +++ b/backend/onboarders/handler.py @@ -0,0 +1,33 @@ +import json +from onboarders.factory import OnboarderFactory +from utils.logger import setup_logger + +logger = setup_logger() + + +def handler(event, context): + """ + Lambda handler that triggers the model engine for each SQS message. + """ + for record in event.get("Records", []): + try: + event_body = json.loads(record["body"]) + # TODO: Implement logic to check which file type we have + # Sample input data + event_body = { + "s3_uri": "s3://retrofit-data-dev/ara_raw_inputs/peabody/2025_11_11 - Peabody - Data Extracts for " + "Domna.xlsx", + "system": "parity", + "format": "xlsx", + "sheet_name": "Sustainability" + } + logger.info("Processing record with body: %s", event_body) + Onboarder = OnboarderFactory.create_onboarder(event_body["system"]) + onboarder = Onboarder(fileuri=event_body["s3_uri"]) + + logger.info("Transforming data for record with body: %s", event_body) + onboarder.transform() + logger.info("Writing data for record with body: %s", event_body) + onboarder.write() + except Exception as e: + logger.error(f"Failed to process record: {e}") diff --git a/backend/onboarders/mappings/parity/age_band.py b/backend/onboarders/mappings/parity/age_band.py index e49fede8..406d39c1 100644 --- a/backend/onboarders/mappings/parity/age_band.py +++ b/backend/onboarders/mappings/parity/age_band.py @@ -1,4 +1,4 @@ -from backend.onboarders.epc_descriptions import EpcConstructionAgeBand +from datatypes.epc.construction_age_band import EpcConstructionAgeBand parity_map = { "Before 1900": EpcConstructionAgeBand.before_1900, diff --git a/backend/onboarders/mappings/parity/roof.py b/backend/onboarders/mappings/parity/roof.py index 14f0c34e..02518c3e 100644 --- a/backend/onboarders/mappings/parity/roof.py +++ b/backend/onboarders/mappings/parity/roof.py @@ -1,5 +1,10 @@ +import pandas as pd from numpy import nan +from typing import Union, Callable +from collections.abc import Mapping from datatypes.epc.roof import EpcRoofDescriptions +from datatypes.epc.efficiency import EpcEfficiency +from datatypes.epc.construction_age_band import EpcConstructionAgeBand roof_map = { # Dwelling above @@ -101,3 +106,356 @@ roof_unknown_age_fallback = { "PitchedNormalLoftAccess": EpcRoofDescriptions.loft_as_built_unknown, "PitchedNormalNoLoftAccess": EpcRoofDescriptions.loft_as_built_unknown, } + +RoofEfficiencyRule = Union[ + EpcEfficiency, + Callable[[EpcConstructionAgeBand, int | None], EpcEfficiency], +] + + +def flat_insulated_efficiency_age_band(age_band: EpcConstructionAgeBand) -> EpcEfficiency: + """ + before 1900, 1900-1929, 1930-1949, 1950-1966, 1967-1975 -> Pitched, no insulation, Very Poor + 1976-1982 -> Pitched, limited insulation, Poor + 1983-1990, to 1996-2002 Pitched, insulated, Average + 2003 - 2006, 2012-2022 -> Pitched, insulated, Good + 2023 onwards -> Pitched, insulated, Very Good + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + + start_year = age_band.start_year() + if start_year >= 2023: + return EpcEfficiency.VERY_GOOD + + if start_year >= 2003: + return EpcEfficiency.GOOD + + if start_year >= 1983: + return EpcEfficiency.AVERAGE + + if start_year >= 1976: + return EpcEfficiency.POOR + + return EpcEfficiency.VERY_POOR + + +def flat_insulated_efficiency_thickness(insulation_thickness: int | None) -> EpcEfficiency: + """ + 12mm -> Very Poor + 25mm - 50mm -> Poor + 75mm - 125mm -> Pitched, insulated, average + 150mm - 250mm -> good + 270mm+ -> very good + :param insulation_thickness: Insulation thickness in mm + :return: EpcEfficiency + """ + + if insulation_thickness is None: + raise ValueError("Insulation thickness is required for flat insulated efficiency calculation") + + if insulation_thickness >= 270: + return EpcEfficiency.VERY_GOOD + + if 150 <= insulation_thickness <= 250: + return EpcEfficiency.GOOD + + if 75 <= insulation_thickness <= 125: + return EpcEfficiency.AVERAGE + + if 25 <= insulation_thickness <= 50: + return EpcEfficiency.POOR + + return EpcEfficiency.VERY_POOR + + +def flat_efficiency(insulation_thickness: int | None, age_band: EpcConstructionAgeBand) -> EpcEfficiency: + """ + Combines both age band and insulation thickness to determine flat roof efficiency. + :param insulation_thickness: Insulation thickness in mm + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + if insulation_thickness is not None: + return flat_insulated_efficiency_thickness(insulation_thickness) + + return flat_insulated_efficiency_age_band(age_band) + + +def loft_insulated_efficiency(age_band: EpcConstructionAgeBand) -> EpcEfficiency: + """ + 2023 onwards -> Very Good + 2012-2022 -> Very Good + 2007-2011 -> Very Good + 2003-2006 -> Very Good + 1996-2002 -> Good + 1991-1995 -> Good + 1983-1990 -> Average + 1976-1982 -> Average + 1967-1975 -> Average + 1950-1966 -> Average + 1930-1949 -> Average + 1900-1929 -> Average + before 1900 -> Average + :param age_band: Input age band, EpcConstructionAgeBand + :return: EpcEfficiency + """ + year = age_band.start_year() + if year >= 2003: + return EpcEfficiency.VERY_GOOD + if year >= 1991: + return EpcEfficiency.GOOD + + return EpcEfficiency.AVERAGE + + +def thatched_efficiency_age_band(age_band: EpcConstructionAgeBand) -> EpcEfficiency: + """ + Maps thatched roof efficiency based on construction age band. + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + year = age_band.start_year() + if year >= 2023: + return EpcEfficiency.VERY_GOOD + if year >= 2003: + return EpcEfficiency.GOOD + + return EpcEfficiency.AVERAGE + + +def thatched_efficiency_thickness(insulation_thickness: int | None) -> EpcEfficiency: + """ + Maps thatched roof efficiency based on insulation thickness. + :param insulation_thickness: Insulation thickness in mm + :return: EpcEfficiency + """ + if insulation_thickness is None: + raise ValueError("Insulation thickness is required for thatched efficiency calculation") + + if insulation_thickness >= 175: + return EpcEfficiency.VERY_GOOD + + if insulation_thickness >= 25: + return EpcEfficiency.GOOD + + return EpcEfficiency.AVERAGE + + +def thatched_efficiency( + insulation_thickness: int | None, + age_band: EpcConstructionAgeBand, +) -> EpcEfficiency: + """ + Combines both age band and insulation thickness to determine thatched roof efficiency. + :param insulation_thickness: Insulation thickness in mm + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + if insulation_thickness is not None: + return thatched_efficiency_thickness(insulation_thickness) + + return thatched_efficiency_age_band(age_band) + + +def sloping_ceiling_efficiency_age_band(age_band: EpcConstructionAgeBand) -> EpcEfficiency: + """ + Maps sloping ceiling roof efficiency based on construction age band. + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + year = age_band.start_year() + if year >= 2023: + return EpcEfficiency.VERY_GOOD + if year >= 2003: + return EpcEfficiency.GOOD + if year >= 1983: + return EpcEfficiency.AVERAGE + if year >= 1976: + return EpcEfficiency.POOR + + return EpcEfficiency.VERY_POOR + + +def sloping_ceiling_efficiency_thickness(insulation_thickness: int | None) -> EpcEfficiency: + """ + Maps sloping ceiling roof efficiency based on insulation thickness. + :param insulation_thickness: Insulation thickness in mm + :return: EpcEfficiency + """ + if insulation_thickness is None: + raise ValueError("Insulation thickness is required for sloping ceiling efficiency calculation") + + if insulation_thickness >= 270: + return EpcEfficiency.VERY_GOOD + + if insulation_thickness >= 150: + return EpcEfficiency.GOOD + + if insulation_thickness >= 75: + return EpcEfficiency.AVERAGE + + if insulation_thickness >= 25: + return EpcEfficiency.POOR + + return EpcEfficiency.VERY_POOR + + +def sloping_ceiling_efficiency( + insulation_thickness: int | None, + age_band: EpcConstructionAgeBand, +) -> EpcEfficiency: + """ + Combines both age band and insulation thickness to determine sloping ceiling roof efficiency. + :param insulation_thickness: Insulation thickness in mm + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + if insulation_thickness is not None: + return sloping_ceiling_efficiency_thickness(insulation_thickness) + + return sloping_ceiling_efficiency_age_band(age_band) + + +def loft_insulated_at_rafters_efficiency_thickness(insulation_thickness: int | None) -> EpcEfficiency: + """ + 400mm, 350mm = very good + 200-300mm = good + 125-175 = average + 50-100 = poor + 25 and below= very poor + :return: + """ + if insulation_thickness is None: + raise ValueError("Insulation thickness is required for loft insulated at rafters efficiency calculation") + + if insulation_thickness >= 350: + return EpcEfficiency.VERY_GOOD + + if insulation_thickness >= 200: + return EpcEfficiency.GOOD + + if insulation_thickness >= 125: + return EpcEfficiency.AVERAGE + + if insulation_thickness >= 50: + return EpcEfficiency.POOR + + return EpcEfficiency.VERY_POOR + + +def loft_insulated_at_rafters_efficiency_age_band(age_band: EpcConstructionAgeBand) -> EpcEfficiency: + """ + # 2023 onwards -> Very Good + # 2003-2006, 2012-2022 -> Good + # 1983 - 1990, 1996-2002 -> Average + # 1976-1982 -> Poor + # 1967-1975 and earlier bands -> Very Poor + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + year = age_band.start_year() + if year >= 2023: + return EpcEfficiency.VERY_GOOD + if year >= 2003: + return EpcEfficiency.GOOD + if year >= 1983: + return EpcEfficiency.AVERAGE + if year >= 1976: + return EpcEfficiency.POOR + + return EpcEfficiency.VERY_POOR + + +def loft_insulated_at_rafters_efficiency( + insulation_thickness: int | None, + age_band: EpcConstructionAgeBand, +) -> EpcEfficiency: + """ + Combines both age band and insulation thickness to determine loft insulated at rafters roof efficiency. + :param insulation_thickness: Insulation thickness in mm + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + if insulation_thickness is not None: + return loft_insulated_at_rafters_efficiency_thickness(insulation_thickness) + + return loft_insulated_at_rafters_efficiency_age_band(age_band) + + +ROOF_DESCRIPTION_EFFICIENCIES: Mapping[EpcRoofDescriptions, RoofEfficiencyRule] = { + # Flat roof + EpcRoofDescriptions.flat_no_insulation: EpcEfficiency.VERY_POOR, + EpcRoofDescriptions.flat_limited_insulation: flat_efficiency, + EpcRoofDescriptions.flat_insulated: flat_efficiency, + + # Loft: + # value mappings + EpcRoofDescriptions.loft_12mm_insulation: EpcEfficiency.VERY_POOR, + EpcRoofDescriptions.loft_25mm_insulation: EpcEfficiency.POOR, + EpcRoofDescriptions.loft_50mm_insulation: EpcEfficiency.POOR, + EpcRoofDescriptions.loft_75mm_insulation: EpcEfficiency.AVERAGE, + EpcRoofDescriptions.loft_100mm_insulation: EpcEfficiency.AVERAGE, + EpcRoofDescriptions.loft_125mm_insulation: EpcEfficiency.AVERAGE, + EpcRoofDescriptions.loft_150mm_insulation: EpcEfficiency.GOOD, + EpcRoofDescriptions.loft_175mm_insulation: EpcEfficiency.GOOD, + EpcRoofDescriptions.loft_200mm_insulation: EpcEfficiency.GOOD, + EpcRoofDescriptions.loft_250mm_insulation: EpcEfficiency.GOOD, + EpcRoofDescriptions.loft_270mm_insulation: EpcEfficiency.VERY_GOOD, + EpcRoofDescriptions.loft_300mm_insulation: EpcEfficiency.VERY_GOOD, + EpcRoofDescriptions.loft_350mm_insulation: EpcEfficiency.VERY_GOOD, + EpcRoofDescriptions.loft_400mm_plus_insulation: EpcEfficiency.VERY_GOOD, + EpcRoofDescriptions.pitched_no_insulation: EpcEfficiency.VERY_POOR, + # function mappings + EpcRoofDescriptions.pitched_insulated_assumed: loft_insulated_efficiency, + + # Loft af rafters + EpcRoofDescriptions.loft_insulated_at_rafters: loft_insulated_at_rafters_efficiency, + + # Another dwelling above + EpcRoofDescriptions.another_dwelling_above: EpcEfficiency.NA, + + # Thatched + EpcRoofDescriptions.thatched: thatched_efficiency, + EpcRoofDescriptions.thatched_with_additional_insulation: thatched_efficiency, + + # Sloping ceiling + EpcRoofDescriptions.sloping_pitched_insulated: sloping_ceiling_efficiency, + EpcRoofDescriptions.sloping_pitched_limited_insulation: sloping_ceiling_efficiency, + EpcRoofDescriptions.sloping_pitched_no_insulation: EpcEfficiency.VERY_POOR, + +} + + +def resolve_roof_efficiency( + description: EpcRoofDescriptions, + age_band: EpcConstructionAgeBand | None, + insulation_thickness: int | None, +) -> EpcEfficiency: + """ + Resolve roof efficiency from description + age band + insulation thickness. + """ + + # Unknown / holding descriptions → efficiency unknown + if description in description.unknown_descriptions: + return EpcEfficiency.NA + + rule = ROOF_DESCRIPTION_EFFICIENCIES.get(description) + + if rule is None: + return EpcEfficiency.NA + + # Fixed efficiency + if isinstance(rule, EpcEfficiency): + return rule + + # Callable rule + if age_band is None or pd.isnull(age_band): + return EpcEfficiency.NA + + try: + # Try (thickness, age_band) + return rule(insulation_thickness, age_band) + except TypeError: + # Fallback to (age_band) + return rule(age_band) diff --git a/backend/onboarders/mappings/parity/walls.py b/backend/onboarders/mappings/parity/walls.py index b46559b9..0ad6d6e1 100644 --- a/backend/onboarders/mappings/parity/walls.py +++ b/backend/onboarders/mappings/parity/walls.py @@ -1,4 +1,8 @@ +from typing import Callable, Union +from collections.abc import Mapping from datatypes.epc.walls import EpcWallDescriptions +from datatypes.epc.construction_age_band import EpcConstructionAgeBand +from datatypes.epc.efficiency import EpcEfficiency # Unique combinations wall_map = { @@ -54,3 +58,154 @@ wall_unknown_age_fallback = { "Sandstone": EpcWallDescriptions.sandstone_as_built_unknown, "Cob": EpcWallDescriptions.cob_as_built_unknown, } + + +def cavity_filled_efficiency(age_band: EpcConstructionAgeBand) -> EpcEfficiency: + """" + Maps cavity filled to efficiency based on construction age band. + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + if age_band in { + EpcConstructionAgeBand.from_2023_onwards + }: + return EpcEfficiency.VERY_GOOD + + return EpcEfficiency.GOOD + + +def internal_external_insulation_efficiency( + age_band: EpcConstructionAgeBand, +) -> EpcEfficiency: + """ + Maps: + - cavity unfilled with internal/external insulation to efficiency based on construction age band. We assumed + based on 100mm insulation + - solid brick with internal/external insulation to efficiency based on construction age band. We assumed + based on 100mm insulation + - system built with internal/external insulation to efficiency based on construction age band. We assumed + based on 100mm insulation + + All of these wall types have the same behaviour in elmhurst + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + if age_band in { + EpcConstructionAgeBand.from_1983_to_1990, + EpcConstructionAgeBand.from_1991_to_1995, + EpcConstructionAgeBand.from_1996_to_2002, + EpcConstructionAgeBand.from_2003_to_2006, + EpcConstructionAgeBand.from_2007_to_2011, + EpcConstructionAgeBand.from_2012_to_2022, + EpcConstructionAgeBand.from_2023_onwards, + }: + return EpcEfficiency.VERY_GOOD + + return EpcEfficiency.GOOD + + +def timber_granite_sandstone_internal_external_efficiency(age_band: EpcConstructionAgeBand) -> EpcEfficiency: + """" + Maps: + - timber frame with internal/external wall insulation to efficiency based on construction age band. + - sandstone/limestone with internal/external wall insulation to efficiency based on construction age band. + - granite/whinstone with internal/external wall insulation to efficiency based on construction age band. + :param age_band: EpcConstructionAgeBand + :return: EpcEfficiency + """ + if age_band in { + EpcConstructionAgeBand.from_2023_onwards + }: + return EpcEfficiency.VERY_GOOD + + return EpcEfficiency.GOOD + + +WallEfficiencyRule = Union[ + EpcEfficiency, + Callable[[EpcConstructionAgeBand, int | None], EpcEfficiency], +] + +WALL_DESCRIPTION_EFFICIENCIES: Mapping[EpcWallDescriptions, WallEfficiencyRule] = { + # Note: all function mappings have been defined based on Elmhurst + # Cavity + # value mappings + EpcWallDescriptions.cavity_no_insulation_assumed: EpcEfficiency.POOR, + EpcWallDescriptions.cavity_partial_insulated_assumed: EpcEfficiency.AVERAGE, + EpcWallDescriptions.cavity_insulated_assumed: EpcEfficiency.GOOD, + EpcWallDescriptions.cavity_filled_plus_internal: EpcEfficiency.VERY_GOOD, + EpcWallDescriptions.cavity_filled_plus_external: EpcEfficiency.VERY_GOOD, + # function mappings + EpcWallDescriptions.cavity_filled_cavity: cavity_filled_efficiency, + EpcWallDescriptions.cavity_internal_insulation: internal_external_insulation_efficiency, + EpcWallDescriptions.cavity_external_insulation: internal_external_insulation_efficiency, + + # Solid brick + # value mappings + EpcWallDescriptions.solid_brick_no_insulation_assumed: EpcEfficiency.POOR, + EpcWallDescriptions.solid_brick_partial_insulated_assumed: EpcEfficiency.AVERAGE, + EpcWallDescriptions.solid_brick_insulated_assumed: EpcEfficiency.GOOD, + # function mappings + EpcWallDescriptions.solid_brick_internal_insulation: internal_external_insulation_efficiency, + EpcWallDescriptions.solid_brick_external_insulation: internal_external_insulation_efficiency, + + # System + # value mappings + EpcWallDescriptions.system_no_insulation_assumed: EpcEfficiency.POOR, + EpcWallDescriptions.system_partial_insulated_assumed: EpcEfficiency.AVERAGE, + EpcWallDescriptions.system_insulated_assumed: EpcEfficiency.GOOD, + # function mappings + EpcWallDescriptions.system_internal_insulation: internal_external_insulation_efficiency, + EpcWallDescriptions.system_external_insulation: internal_external_insulation_efficiency, + + # Timber frame + # value mappings + EpcWallDescriptions.timber_frame_no_insulation_assumed: EpcEfficiency.POOR, + EpcWallDescriptions.timber_frame_partial_insulated_assumed: EpcEfficiency.AVERAGE, + EpcWallDescriptions.timber_frame_insulated_assumed: EpcEfficiency.GOOD, + # function mappings + EpcWallDescriptions.timber_frame_internal_insulation: timber_granite_sandstone_internal_external_efficiency, + EpcWallDescriptions.timber_frame_external_insulation: timber_granite_sandstone_internal_external_efficiency, + + # Granite / whinstone + EpcWallDescriptions.granite_whinstone_no_insulation_assumed: EpcEfficiency.VERY_POOR, + EpcWallDescriptions.granite_whinstone_partial_insulated_assumed: EpcEfficiency.AVERAGE, + EpcWallDescriptions.granite_whinestone_insulated_assumed: EpcEfficiency.GOOD, + # function mappings + EpcWallDescriptions.granite_whinstone_internal_insulation: timber_granite_sandstone_internal_external_efficiency, + EpcWallDescriptions.granite_whinstone_external_insulation: timber_granite_sandstone_internal_external_efficiency, + + # Sandstone / limestone + EpcWallDescriptions.sandstone_limestone_no_insulation_assumed: EpcEfficiency.VERY_POOR, + EpcWallDescriptions.sandstone_limestone_partial_insulated_assumed: EpcEfficiency.AVERAGE, + EpcWallDescriptions.sandstone_limestone_insulated_assumed: EpcEfficiency.GOOD, + # function mappings + EpcWallDescriptions.sandstone_limestone_internal_insulation: timber_granite_sandstone_internal_external_efficiency, + EpcWallDescriptions.sandstone_limestone_external_insulation: timber_granite_sandstone_internal_external_efficiency, + + # Cob (special case) + EpcWallDescriptions.cob_as_built_average: EpcEfficiency.AVERAGE, + EpcWallDescriptions.cob_as_built_good: EpcEfficiency.GOOD, + + # Unknown mappings which are unhandled + EpcWallDescriptions.cavity_as_built_unknown: EpcEfficiency.NA, + EpcWallDescriptions.solid_brick_as_built_unknown: EpcEfficiency.NA, + EpcWallDescriptions.system_as_built_unknown: EpcEfficiency.NA, + EpcWallDescriptions.timber_frame_as_built_unknown: EpcEfficiency.NA, + EpcWallDescriptions.granite_as_built_unknown: EpcEfficiency.NA, + EpcWallDescriptions.sandstone_as_built_unknown: EpcEfficiency.NA, + EpcWallDescriptions.cob_as_built_unknown: EpcEfficiency.NA, + +} + + +def resolve_wall_efficiency( + description: EpcWallDescriptions, + age_band: EpcConstructionAgeBand, +) -> EpcEfficiency: + rule = WALL_DESCRIPTION_EFFICIENCIES[description] + + if isinstance(rule, EpcEfficiency): + return rule + + return rule(age_band) diff --git a/backend/onboarders/parity.py b/backend/onboarders/parity.py index c7f982df..8fc5496e 100644 --- a/backend/onboarders/parity.py +++ b/backend/onboarders/parity.py @@ -2,12 +2,15 @@ import re from tqdm import tqdm import pandas as pd from backend.onboarders.base import OnboarderBase +# Parity mappings from backend.onboarders.mappings.parity.property_type import parity_map as property_map from backend.onboarders.mappings.parity.age_band import parity_map as age_band_map from backend.onboarders.mappings.parity.built_form import parity_map as built_form_map -from backend.onboarders.mappings.parity.walls import wall_map, wall_unknown_age_fallback -from backend.onboarders.epc_descriptions import EpcWallDescriptions, EpcConstructionAgeBand, EpcEfficiency, \ - WALL_DESCRIPTION_EFFICIENCIES, resolve_roof_efficiency +from backend.onboarders.mappings.parity.walls import wall_map, wall_unknown_age_fallback, WALL_DESCRIPTION_EFFICIENCIES +from onboarders.mappings.parity.roof import roof_map, roof_unknown_age_fallback, resolve_roof_efficiency +from onboarders.mappings.parity.floor import floor_map +from onboarders.mappings.parity.heating import heating_map +from onboarders.mappings.parity.glazing import glazing_map from backend.onboarders.mappings.parity.as_built_wall_classifiers import as_built_wall_classifiers from backend.onboarders.mappings.parity.as_built_roof_classifiers import as_built_roof_classifiers from backend.onboarders.mappings.parity.as_built_floor_classifiers import ( @@ -15,20 +18,12 @@ from backend.onboarders.mappings.parity.as_built_floor_classifiers import ( ) from datatypes.epc.roof import EpcRoofDescriptions from datatypes.epc.floor import EpcFloorDescriptions -from onboarders.mappings.parity.roof import roof_map, roof_unknown_age_fallback -from onboarders.mappings.parity.floor import floor_map -from onboarders.mappings.parity.heating import heating_map -from onboarders.mappings.parity.glazing import glazing_map +from datatypes.epc.construction_age_band import EpcConstructionAgeBand +from datatypes.epc.walls import EpcWallDescriptions +from datatypes.epc.efficiency import EpcEfficiency tqdm.pandas() -# Sample input data -data = pd.read_excel( - "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Peabody/Nov 2025 Consulting Project/2025_11_11 - Peabody " - "- Data Extracts for Domna.xlsx", - sheet_name="Sustainability" -) - class ParityOnboarder(OnboarderBase):