Model/infrastructure/condition/parsing/calico_parser.py

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):
    def __init__(self, uprn_lookup: UprnLookup) -> None:
        self._uprn_lookup = uprn_lookup
        self._blank_placeholder_count = 0
        self._unmatched_reference_count = 0

    @property
    def blank_placeholder_count(self) -> int:
        """Rows dropped on parse for a blank / N/A covering type."""
        return self._blank_placeholder_count

    @property
    def unmatched_reference_count(self) -> int:
        """Distinct asset references dropped for not resolving to a portfolio UPRN."""
        return self._unmatched_reference_count

    def parse(self, file_stream: BinaryIO) -> List[CalicoProperty]:
        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 = self._parse_assets(rows, column_index)
        return self._group_assets_into_properties(assets)

    def _parse_assets(
        self, 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.
                self._blank_placeholder_count += 1
                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:
                self._unmatched_reference_count += 1
                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)
        }

    # Sentinels Calico lodges in the Type column for a placeholder row (no
    # roof-covering record captured), treated as "no covering".
    _BLANK_COVERING_SENTINELS = {"", "n/a", "na", "none"}

    @staticmethod
    def _clean_str(value: object) -> Optional[str]:
        if not isinstance(value, str):
            return None
        stripped = value.strip()
        if stripped.lower() in CalicoParser._BLANK_COVERING_SENTINELS:
            return None
        return stripped

    @staticmethod
    def _to_int(value: object) -> Optional[int]:
        if isinstance(value, (int, float)):
            return int(value)
        return None