Parse Peabody condition data xlsx 🟥

backend/condition/file_type.py

@@ -2,6 +2,7 @@ from enum import Enum
 
 class FileType(Enum):
     LBWF = "lbwf"
+    Peabody = "peabody"
 
 def detect_file_type(filepath: str) -> FileType:
     path = filepath.lower()
@@ -9,4 +10,7 @@ def detect_file_type(filepath: str) -> FileType:
     if "lbwf" in path:
         return FileType.LBWF
     
+    if "peadbody" in path:
+        return FileType.Peabody
+    
     raise ValueError("Unrecognised file path")

backend/condition/parsing/factory.py

@@ -3,10 +3,14 @@ from backend.condition.domain.mapping.mapper import Mapper
 from backend.condition.file_type import FileType
 from backend.condition.parsing.parser import Parser
 from backend.condition.parsing.lbwf_parser import LbwfParser
+from backend.condition.parsing.peabody_parser import PeabodyParser
 
 def select_parser(file_type: FileType) -> Parser:
     if file_type is FileType.LBWF:
         return LbwfParser()
+    
+    if file_type is FileType.Peabody:
+        return PeabodyParser()
 
     raise ValueError("Unrecognised file type, unable to instantiate Parser")
 

backend/condition/parsing/peabody_parser.py

@@ -0,0 +1,7 @@
+from typing import Any, BinaryIO
+from backend.condition.parsing.parser import Parser
+
+
+class PeabodyParser(Parser):
+    def parse(self, file_stream: BinaryIO) -> Any:
+        raise NotImplementedError

backend/condition/parsing/records/peabody/peabody_asset_condition.py

@@ -0,0 +1,22 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional
+
+@dataclass
+class PeabodyAssetCondition:
+    lo_reference: str
+    full_address: str
+    location_type_code: int
+    parent_lo_reference: str
+    element_code: int
+    element: int
+    sub_element_code: int
+    sub_element: str
+    material_code: int
+    material_or_answer: str
+    renewal_quantity: int
+    renewal: int
+    cloned: str
+    lo_type_code: int
+    renewal_cost: Optional[float] = None
+    condition_survey_date: Optional[datetime] = None

backend/condition/parsing/records/peabody/peabody_property.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+from typing import List
+
+from backend.condition.parsing.records.peabody.peabody_asset_condition import PeabodyAssetCondition
+
+@dataclass
+class PeabodyProperty:
+    uprn: int
+    assets: List[PeabodyAssetCondition]

backend/condition/tests/parsing/test_parsing_factory.py

@@ -11,5 +11,16 @@ def test_selects_lbwf_parser():
     # act
     actual_class_name = select_parser(file_type).__class__.__name__
 
+    # assert
+    assert expected_class_name == actual_class_name
+
+def test_selects_peabody_parser():
+    # arrange
+    file_type = FileType.Peabody
+    expected_class_name = "PeabodyParser"
+
+    # act
+    actual_class_name = select_parser(file_type).__class__.__name__
+
     # assert
     assert expected_class_name == actual_class_name

backend/condition/tests/parsing/test_peabody_parser.py

@@ -0,0 +1,124 @@
+from typing import Any
+import pytest
+from io import BytesIO
+from openpyxl import Workbook
+from datetime import datetime
+
+from backend.condition.parsing.peabody_parser import PeabodyParser
+from backend.condition.parsing.records.peabody.peabody_asset_condition import PeabodyAssetCondition
+from backend.condition.parsing.records.peabody.peabody_property import PeabodyProperty
+
+@pytest.fixture
+def peabody_assets_xlsx_bytes() -> BytesIO:
+    wb = Workbook()
+    survey_records_d_and_lower = wb.active
+    survey_records_d_and_lower.title = "Survey Records - D & Lower"
+    survey_records_d_and_lower.append([
+        "Lo_Reference",
+        "full_address",
+        "location_type_code",
+        "Parent_Lo_Reference",
+        "Element_Code",
+        "Element",
+        "Sub_Element_Code",
+        "Sub_Element",
+        "Material_Code",
+        "material_or_answer",
+        "Renewal_Quantity",
+        "Renewal_Year",
+        "Renewal_Cost",
+        "cloned",
+        "lo_type_code",
+        "condition_survey_date",
+    ])
+    survey_records_d_and_lower.append([
+        "B000RAND",
+        "1-11 RANDOM HOUSE LONDON",
+        3,
+        "RAND2EST",
+        110,
+        "ROOFS",
+        1,
+        "Primary Roof",
+        9,
+        "Other",
+        3,
+        2054,
+        330,
+        "N",
+        3,
+        datetime(2025,12,4,9,17,0)
+    ])
+    survey_records_d_and_lower.append([
+        "B000FAKE",
+        "3-10 FAKE CLOSE LONDON",
+        3,
+        "FAKEEST",
+        100,
+        "GENERAL",
+        15,
+        "External Decoration",
+        2,
+        "Normal",
+        1,
+        2035,
+        1500.7,
+        "N",
+        3,
+        datetime(2025,7,5,0,0,0)
+    ])
+    survey_records_d_and_lower.append([
+        "B000MIS",
+        "99 MISC ROAD LONDON",
+        3,
+        "300828",
+        54,
+        "HHSRS",
+        29,
+        "HHSRS Structural Collapse & Falling Elements",
+        4,
+        "HHSRS Moderate",
+        2,
+        2027,
+        None,
+        "N",
+        3,
+        None
+    ])
+    survey_records_d_and_lower.append([
+        "B000MIS",
+        "99 MISC ROAD LONDON",
+        3,
+        "300828",
+        53,
+        "External",
+        2,
+        "Chimney",
+        2,
+        "Present",
+        33,
+        2053,
+        3531,
+        "N",
+        3,
+        None
+    ])
+
+
+    stream = BytesIO()
+    wb.save(stream)
+    stream.seek(0)
+
+    return stream
+
+def test_peabody_parser_parses_conditions(peabody_assets_xlsx_bytes):
+    # arrange
+    parser = PeabodyParser()
+
+    # act
+    result: Any = parser.parse(peabody_assets_xlsx_bytes)
+
+    # assert
+    assert len(result) == 3
+
+    assert all(isinstance(item, PeabodyProperty) for item in result)