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"), lmk_key=None, epc_certificate_number=None, ) @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 )