Model/backend/addresses/Addresses.py

import warnings
import pandas as pd
from typing import Iterator
from backend.addresses.Address import Address
from datatypes.epc.property_type_built_form import PropertyType


class Addresses:
    def __init__(self, addresses: list[Address]):
        self._addresses = addresses
        # self._identity_index = self._build_identity_index()

    def __getitem__(self, index: int) -> Address:
        return self._addresses[index]

    def __len__(self) -> int:
        return len(self._addresses)

    def __iter__(self) -> Iterator[Address]:
        return iter(self._addresses)

    @classmethod
    def from_plan_input(cls, plan_input: list[dict], body) -> "Addresses":
        addresses = []

        for row in plan_input:
            try:
                if body.file_format == "ara_property_list":
                    addr = cls.parse_ara_row(row, body)
                else:
                    addr = cls._parse_row_deprecated(row, body)

            # Fallback if new parser fails
            except Exception:
                warnings.warn(
                    "Falling back to deprecated parser for row",
                    RuntimeWarning,
                    stacklevel=2,
                )
                addr = cls._parse_row_deprecated(row, body)

            addresses.append(addr)

        addresses = cls(addresses)
        addresses.validate_uprns()
        return addresses

    def get_uprns(self):
        return [x.uprn for x in self._addresses if x.uprn is not None]

    def get_landlord_ids(self):
        return [x.landlord_property_id for x in self._addresses if x.landlord_property_id is not None]

    def get_unique_postcodes(self):
        return list({x.postcode for x in self._addresses})

    def get_postcodes_for_flats(self):
        # Method to extract all of the postcodes associated to a flat, which is used for remote assessments
        # on flats
        return [x.postcode for x in self._addresses if x.landlord_property_type in [PropertyType.flat.value]]

    def get_property_requests(self):
        return [x.request_data for x in self._addresses]

    def validate_uprns(self):
        """Raise ValueError if any address has a non-int UPRN."""
        for addr in self._addresses:
            if addr.uprn is not None and not isinstance(addr.uprn, int):
                raise ValueError(f"Address with non-integer UPRN detected: {addr.uprn} in {addr}")

    @staticmethod
    def parse_ara_row(row: dict, body) -> Address:
        """
        Method to parse a row from the ARA property list format, which is a more standardised format that we are
        moving towards.
        :param row: A dictionary representing a row from the ARA property list, which should have keys corresponding
        to the Address dataclass fields. The method will attempt to parse these fields and create an Address object.
        :param body: The PlanTriggerRequest body, which may contain additional information about the file format and
        other details that could be relevant for parsing.
        :return: An Address object created from the parsed row data.
        """
        return Address(
            uprn=int(row["uprn"]),
            landlord_property_id=str(row["landlord_property_id"]) if row.get("landlord_property_id") else None,
            address_1=row["address_1"],
            address_2=row.get("address_2"),
            address_3=row.get("address_3"),
            full_address=row["full_address"],
            postcode=str(row["postcode"]),
            landlord_total_floor_area_m2=float(row["landlord_total_floor_area_m2"]) if row.get(
                "landlord_total_floor_area_m2") else None,
            landlord_property_type=row.get("landlord_property_type"),
            landlord_built_form=row.get("landlord_built_form"),
            landlord_wall_construction=row.get("landlord_wall_construction"),
            landlord_roof_construction=row.get("landlord_roof_construction"),
            landlord_floor_construction=row.get("landlord_floor_construction"),
            landlord_windows_type=row.get("landlord_windows_type"),
            landlord_heating_system=row.get("landlord_heating_system"),
            landlord_fuel_type=row.get("landlord_fuel_type"),
            landlord_heating_controls=row.get("landlord_heating_controls"),
            landlord_hot_water_system=row.get("landlord_hot_water_system"),
            landlord_wall_efficiency=row.get("landlord_wall_efficiency"),
            landlord_roof_efficiency=row.get("landlord_roof_efficiency"),
            landlord_windows_efficiency=row.get("landlord_windows_efficiency"),
            landlord_heating_efficiency=row.get("landlord_heating_efficiency"),
            landlord_heating_controls_efficiency=row.get("landlord_heating_controls_efficiency"),
            landlord_hot_water_efficiency=row.get("landlord_hot_water_efficiency"),
            landlord_has_sloping_ceiling=bool(row.get("landlord_has_sloping_ceiling")) if row.get(
                "landlord_has_sloping_ceiling") is not None else None,
            landlord_multi_glaze_proportion=float(row["landlord_multi_glaze_proportion"]) if row.get(
                "landlord_multi_glaze_proportion") else None,
            landlord_construction_age_band=row.get("landlord_construction_age_band"),
        )

    @staticmethod
    def _parse_row_deprecated(row: dict, body) -> Address:
        def clean_uprn(v):
            if v is None:
                return None
            try:
                return int(float(v))
            except (TypeError, ValueError):
                raise ValueError(f"Invalid UPRN value: {v}")

        uprn_option1 = row.get("uprn")
        uprn_option1 = uprn_option1 if not pd.isnull(uprn_option1) else None
        uprn_option2 = row.get("ordnance_survey_uprn")
        uprn_option2 = uprn_option2 if not pd.isnull(uprn_option2) else None

        uprn = clean_uprn(uprn_option1 or uprn_option2)

        address = row.get("address") or row.get("domna_address_1") or ""
        full_address = row.get("domna_full_address") or address or ""

        postcode = str(row.get("postcode", "")).strip().upper()

        lmk_key = row.get("lmk_key", None)
        # Handle NAN
        if pd.isnull(lmk_key):
            lmk_key = None

        epc_certificate_number = row.get("certificate_number", None)
        if pd.isnull(epc_certificate_number):
            epc_certificate_number = None

        landlord_heating_system = row.get("epc_heating_type", None)
        if pd.isnull(landlord_heating_system):
            landlord_heating_system = None

        return Address(
            uprn=uprn,
            landlord_property_id=str(row["landlord_property_id"]) if row.get("landlord_property_id") else None,
            address_1=str(address).strip(),
            address_2=None,
            address_3=None,
            full_address=str(full_address).strip(),
            postcode=postcode,

            landlord_total_floor_area_m2=None,

            # Map old to new fields
            landlord_property_type=row.get("property_type") or row.get("landlord_property_type"),
            landlord_built_form=row.get("built_form") or row.get("landlord_built_form"),

            landlord_wall_construction=None,
            landlord_roof_construction=None,
            landlord_floor_construction=None,
            landlord_windows_type=None,
            landlord_heating_system=landlord_heating_system,
            landlord_fuel_type=None,
            landlord_heating_controls=None,
            landlord_hot_water_system=None,

            landlord_wall_efficiency=None,
            landlord_roof_efficiency=None,
            landlord_windows_efficiency=None,
            landlord_heating_efficiency=None,
            landlord_heating_controls_efficiency=None,
            landlord_hot_water_efficiency=None,

            landlord_has_sloping_ceiling=None,
            landlord_multi_glaze_proportion=None,
            landlord_construction_age_band=None,

            # EPC identifiers which are helpful if UPRN is problematic
            lmk_key=lmk_key,
            epc_certificate_number=epc_certificate_number
        )