Parse a Calico workbook and drop rows with no covering type 🟩

CalicoParser reads the workbook into CalicoAssetCondition rows, drops rows
with no covering type, and groups them by asset reference into CalicoProperty,
resolving each reference to a UPRN via the injected lookup (unmatched dropped).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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