diff --git a/infrastructure/condition/parsing/calico_parser.py b/infrastructure/condition/parsing/calico_parser.py index 5d4edad0b..4532250a4 100644 --- a/infrastructure/condition/parsing/calico_parser.py +++ b/infrastructure/condition/parsing/calico_parser.py @@ -1,8 +1,16 @@ -from typing import BinaryIO, List +from collections import defaultdict +from typing import Any, BinaryIO, Dict, List, Optional, Tuple +from openpyxl import load_workbook + +from domain.condition.records.calico.calico_asset_condition import CalicoAssetCondition from domain.condition.records.calico.calico_property import CalicoProperty +from infrastructure.condition.date_utils import normalise_date from infrastructure.condition.parsing.parser import Parser from repositories.condition.uprn_lookup import UprnLookup +from utils.logger import setup_logger + +logger = setup_logger() class CalicoParser(Parser): @@ -10,4 +18,78 @@ class CalicoParser(Parser): self._uprn_lookup = uprn_lookup def parse(self, file_stream: BinaryIO) -> List[CalicoProperty]: - raise NotImplementedError + workbook = load_workbook(file_stream, data_only=True) + sheet = workbook[workbook.sheetnames[0]] + + rows = sheet.iter_rows(values_only=True) + column_index = CalicoParser._column_indexes(next(rows)) + + assets = CalicoParser._parse_assets(rows, column_index) + return self._group_assets_into_properties(assets) + + @staticmethod + def _parse_assets( + rows: Any, column_index: Dict[str, int] + ) -> List[CalicoAssetCondition]: + assets: List[CalicoAssetCondition] = [] + for row in rows: + covering_type = CalicoParser._clean_str(row[column_index["Type"]]) + if covering_type is None: + # A blank covering type is a placeholder row (no roof-covering + # record captured), not an observation -- drop on parse. + continue + assets.append( + CalicoAssetCondition( + asset_reference=int(row[column_index["Asset Reference"]]), + path=CalicoParser._clean_str(row[column_index["Path"]]) or "", + covering_type=covering_type, + fitted_date=normalise_date(row[column_index["Fitted / Renewed Date"]]), + quantity=CalicoParser._to_int(row[column_index["Quantity"]]), + remaining_life=CalicoParser._to_int( + row[column_index["Remaining Life"]] + ), + ) + ) + return assets + + def _group_assets_into_properties( + self, assets: List[CalicoAssetCondition] + ) -> List[CalicoProperty]: + ref_to_uprn: Dict[str, int] = self._uprn_lookup.get_property_ref_to_uprn_lookup() + + assets_by_ref: Dict[int, List[CalicoAssetCondition]] = defaultdict(list) + for asset in assets: + assets_by_ref[asset.asset_reference].append(asset) + + properties: List[CalicoProperty] = [] + for reference, grouped_assets in assets_by_ref.items(): + uprn = ref_to_uprn.get(str(reference)) + if uprn is None: + logger.debug( + f"[CalicoParser] No UPRN for asset reference {reference}; dropping" + ) + continue + properties.append(CalicoProperty(uprn=uprn, assets=grouped_assets)) + + return properties + + @staticmethod + def _column_indexes(headers: Tuple[object | None, ...]) -> Dict[str, int]: + return { + header: index + for index, header in enumerate(headers) + if isinstance(header, str) + } + + @staticmethod + def _clean_str(value: object) -> Optional[str]: + if not isinstance(value, str): + return None + stripped = value.strip() + return stripped or None + + @staticmethod + def _to_int(value: object) -> Optional[int]: + if isinstance(value, (int, float)): + return int(value) + return None