mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
190 lines
8 KiB
Python
190 lines
8 KiB
Python
import warnings
|
|
import pandas as pd
|
|
from typing import Iterator
|
|
from backend.addresses.Address import Address
|
|
from domain.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
|
|
)
|