Model/backend/addresses/Addresses.py

import warnings
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 = []
        if body.file_format == "ara_property_list":
            row_parser = cls.parse_ara_row
        else:
            warnings.warn(
                "_parse_row_deprecated is deprecated and will be removed in a future version. "
                "Use the parse_ara_row method instead",
                DeprecationWarning,
                stacklevel=2
            )
            row_parser = cls._parse_row_deprecated

        for row in plan_input:
            addresses.append(row_parser(row, body))
        return cls(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]

    @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:
        """
        Is a method to be deprecated in favour of using the new array property list format
        :param row:
        :param body:
        :return:
        """

        def clean_uprn(v):
            try:
                return int(float(v))
            except (TypeError, ValueError):
                return None

        uprn = clean_uprn(row.get("uprn"))

        address = row.get("address")
        if not address and body.file_format == "domna_asset_list":
            address = row.get("domna_address_1")

        full_address = (
            row.get("domna_full_address")
            if body.file_format == "domna_asset_list"
            else None
        )
        if not isinstance(full_address, str):
            full_address = None

        postcode = str(row["postcode"]).strip().upper()

        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() if address else None,
            full_address=str(full_address).strip() if full_address else None,
            postcode=postcode,
            landlord_property_type=row.get("property_type"),
            landlord_built_form=row.get("built_form"),
            # estimated=bool(row.get("estimated", False)),
            address_2=None,
            address_3=None,
            landlord_total_floor_area_m2=None,
            landlord_wall_construction=None,
            landlord_roof_construction=None,
            landlord_floor_construction=None,
            landlord_windows_type=None,
            landlord_heating_system=None,
            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,
        )

    # def _build_identity_index(self) -> dict:
    #     index = {}
    #     for addr in self._addresses:
    #         key = addr.identity_key()
    #         if key in index:
    #             raise ValueError(f"Duplicate address identity detected: {key}")
    #         index[key] = addr
    #     return index