junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Map peabody data to new structure 🟥

Browse source
This commit is contained in:
Daniel Roth 2026-01-27 13:53:09 +00:00
parent 6a2bb26bae
commit a0fa676230
7 changed files with 127 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
backend/condition/domain/element.py
View file

@ -7,6 +7,6 @@ from backend.condition.domain.element_type import ElementType
@dataclass
class Element:
element: ElementType
element_type: ElementType
element_instance: int
aspect_conditions: List[AspectCondition]

15
backend/condition/domain/mapping/lbwf/lbwf_mapper.py
View file

@ -4,6 +4,7 @@ from backend.condition.domain.element import Element
from backend.condition.domain.mapping.element_mapping import ElementMapping
from backend.condition.domain.mapping.lbwf.lbwf_element_map import LBWF_ELEMENT_MAP
from backend.condition.domain.mapping.mapper import Mapper
from backend.condition.domain.property_condition_survey import PropertyConditionSurvey
from backend.condition.parsing.records.lbwf.lbwf_asset_condition import (
LbwfAssetCondition,
)
@ -16,16 +17,18 @@ logger = setup_logger()
class LbwfMapper(Mapper):
def map_asset_conditions_for_property(
self, client_data: Any, survey_year: Optional[int] = None
) -> List[Element]:
self, client_property_data: Any, survey_year: Optional[int] = None
) -> PropertyConditionSurvey:
raise NotImplementedError
assert isinstance(
client_data, LbwfHouse
client_property_data, LbwfHouse
) # TODO: think of a better way to do this
mapped_assets: List[Element] = []
uprn: int = client_data.uprn
for raw_asset in client_data.assets:
uprn: int = client_property_data.uprn
for raw_asset in client_property_data.assets:
# Ignore metadata rows
if raw_asset.element_code not in ["EICINSFREQ", "DECNTHMINC"]:
try:
@ -41,7 +44,7 @@ class LbwfMapper(Mapper):
mapped_assets.append(
Element(
uprn=uprn,
element=element_mapping.element,
element_type=element_mapping.element,
aspect_type=element_mapping.aspect_type,
value=raw_asset.attribute_code_description,
quantity=raw_asset.quantity,

5
backend/condition/domain/mapping/mapper.py
View file

@ -2,13 +2,14 @@ from abc import ABC, abstractmethod
from typing import Any, List, Optional
from backend.condition.domain.element import Element
from backend.condition.domain.property_condition_survey import PropertyConditionSurvey
class Mapper(ABC):
@abstractmethod
def map_asset_conditions_for_property(
self, client_data: Any, survey_year: Optional[int] = None
) -> List[Element]:
self, client_property_data: Any, survey_year: Optional[int] = None
) -> PropertyConditionSurvey:
# TODO: client_data should be properly typed
pass

15
backend/condition/domain/mapping/peabody/peabody_mapper.py
View file

@ -6,6 +6,7 @@ from backend.condition.domain.mapping.peabody.peabody_element_map import (
PEABODY_ELEMENT_MAP,
)
from backend.condition.domain.mapping.mapper import Mapper
from backend.condition.domain.property_condition_survey import PropertyConditionSurvey
from backend.condition.parsing.records.peabody.peabody_property import PeabodyProperty
from utils.logger import setup_logger
@ -14,16 +15,18 @@ logger = setup_logger()
class PeabodyMapper(Mapper):
def map_asset_conditions_for_property(
self, client_data: Any, survey_year: Optional[int] = None
) -> List[Element]:
self, client_property_data: Any, survey_year: Optional[int] = None
) -> PropertyConditionSurvey:
raise NotImplementedError
assert isinstance(
client_data, PeabodyProperty
client_property_data, PeabodyProperty
) # TODO: think of a better way to do this
mapped_assets: List[Element] = []
uprn: int = client_data.uprn
for raw_asset in client_data.assets:
uprn: int = client_property_data.uprn
for raw_asset in client_property_data.assets:
try:
element_mapping: ElementMapping = PeabodyMapper._map_element(
raw_asset.element_code, raw_asset.sub_element_code
@ -38,7 +41,7 @@ class PeabodyMapper(Mapper):
mapped_assets.append(
Element(
uprn=uprn,
element=element_mapping.element,
element_type=element_mapping.element,
aspect_type=element_mapping.aspect_type,
value=raw_asset.material_or_answer,
quantity=raw_asset.renewal_quantity,

9
backend/condition/processor.py
View file

@ -1,8 +1,8 @@
from typing import Any, BinaryIO, List
from datetime import datetime
from backend.condition.domain.element import Element
from backend.condition.domain.mapping.mapper import Mapper
from backend.condition.domain.property_condition_survey import PropertyConditionSurvey
from backend.condition.parsing.parser import Parser
from utils.logger import setup_logger
from backend.condition.file_type import FileType, detect_file_type
@ -22,8 +22,11 @@ def process_file(file_stream: BinaryIO, source_key: str) -> None:
survey_year = datetime.now().year # TODO: get this from filepath or elsewhere
assets: List[Element] = []
property_condition_surveys: List[PropertyConditionSurvey] = []
for p in raw_properties:
assets.extend(mapper.map_asset_conditions_for_property(p, survey_year))
property_condition_surveys.push(
mapper.map_asset_conditions_for_property(p, survey_year)
)
print("done") # temp

18
backend/condition/tests/mapping/test_lbwf_mapper.py
View file

@ -222,7 +222,7 @@ def test_lbwf_mapper_maps_house():
expected_assets: List[Element] = [
Element(
uprn=1,
element=ElementType.ACCESSIBLE_HOUSING_REGISTER,
element_type=ElementType.ACCESSIBLE_HOUSING_REGISTER,
aspect_type=AspectType.CATEGORY,
element_instance=None,
value="General Needs",
@ -233,7 +233,7 @@ def test_lbwf_mapper_maps_house():
),
Element(
uprn=1,
element=ElementType.FLOOR_LEVEL_FRONT_DOOR,
element_type=ElementType.FLOOR_LEVEL_FRONT_DOOR,
aspect_type=AspectType.LOCATION,
element_instance=None,
value="Ground Floor",
@ -244,7 +244,7 @@ def test_lbwf_mapper_maps_house():
),
Element(
uprn=1,
element=ElementType.ASBESTOS,
element_type=ElementType.ASBESTOS,
aspect_type=AspectType.PRESENCE,
element_instance=None,
value="Yes",
@ -255,7 +255,7 @@ def test_lbwf_mapper_maps_house():
),
Element(
uprn=1,
element=ElementType.HHSRS_ASBESTOS_AND_MMF,
element_type=ElementType.HHSRS_ASBESTOS_AND_MMF,
aspect_type=AspectType.RISK,
element_instance=None,
value="Category 4 - Typical Risk",
@ -266,7 +266,7 @@ def test_lbwf_mapper_maps_house():
),
Element(
uprn=1,
element=ElementType.BATHROOM,
element_type=ElementType.BATHROOM,
aspect_type=AspectType.LOCATION,
element_instance=None,
value="Bathroom on Entrance Level in Property",
@ -277,7 +277,7 @@ def test_lbwf_mapper_maps_house():
),
Element(
uprn=1,
element=ElementType.CENTRAL_HEATING,
element_type=ElementType.CENTRAL_HEATING,
aspect_type=AspectType.EXTENT,
element_instance=None,
value="No Central Heating in Property",
@ -288,7 +288,7 @@ def test_lbwf_mapper_maps_house():
),
Element(
uprn=1,
element=ElementType.HHSRS_FIRE,
element_type=ElementType.HHSRS_FIRE,
aspect_type=AspectType.RISK,
element_instance=None,
value="Category 4 - Typical Risk",
@ -299,7 +299,7 @@ def test_lbwf_mapper_maps_house():
),
Element(
uprn=1,
element=ElementType.EXTERNAL_WALL,
element_type=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.FINISH,
element_instance=1,
value="Render or Pebbledash in External Area",
@ -310,7 +310,7 @@ def test_lbwf_mapper_maps_house():
),
Element(
uprn=1,
element=ElementType.EXTERNAL_WALL,
element_type=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.FINISH,
element_instance=2,
value="Smooth Render Wall Finish 2 in External Area",

169
backend/condition/tests/mapping/test_peabody_mapper.py
View file

@ -1,9 +1,10 @@
from datetime import datetime
from typing import List
from datetime import datetime, date
from backend.condition.domain.aspect_condition import AspectCondition
from backend.condition.domain.aspect_type import AspectType
from backend.condition.domain.element_type import ElementType
from backend.condition.domain.mapping.peabody.peabody_mapper import PeabodyMapper
from backend.condition.domain.property_condition_survey import PropertyConditionSurvey
from backend.condition.parsing.records.peabody.peabody_asset_condition import (
PeabodyAssetCondition,
)
@ -56,40 +57,51 @@ def test_peabody_mapper_maps_property():
)
mapper = PeabodyMapper()
expected_assets: List[Element] = [
Element(
uprn=1,
element=ElementType.EXTERNAL_WINDOWS,
aspect_type=AspectType.MATERIAL,
value="UPVC Double Glazed",
quantity=8,
install_date=None,
renewal_year=2036,
element_instance=None,
source_system=None,
comments=None,
),
Element(
uprn=1,
element=ElementType.EXTERNAL_DECORATION,
aspect_type=AspectType.CONDITION,
value="Normal",
quantity=1,
install_date=None,
renewal_year=2029,
element_instance=None,
source_system=None,
comments=None,
),
]
expected_condition_survey = PropertyConditionSurvey(
uprn=1,
elements=[
Element(
element_type=ElementType.EXTERNAL_WINDOWS,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.MATERIAL,
aspect_instance=1,
value="UPVC Double Glazed",
quantity=8,
install_date=None,
renewal_year=2036,
comments=None,
),
],
),
Element(
element_type=ElementType.EXTERNAL_DECORATION,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.CONDITION,
aspect_instance=1,
value="Normal",
quantity=1,
install_date=None,
renewal_year=2029,
comments=None,
)
],
),
],
date=date(2000, 1, 1), # what should this be?
source="Peabody",
)
# act
actual_assets = mapper.map_asset_conditions_for_property(peabody_property)
actual_condition_survey: PropertyConditionSurvey = (
mapper.map_asset_conditions_for_property(peabody_property)
)
# assert
assert len(actual_assets) == len(expected_assets)
for i, (actual, expected) in enumerate(zip(actual_assets, expected_assets)):
assert actual == expected, f"Mismatch at index {i}"
assert actual_condition_survey == expected_condition_survey
def test_wall_primary_and_secondary_wall_finish_map_correctly():
@ -155,52 +167,51 @@ def test_wall_primary_and_secondary_wall_finish_map_correctly():
)
mapper = PeabodyMapper()
expected_assets: List[Element] = [
Element(
uprn=1,
element=ElementType.EXTERNAL_WALLS,
aspect_type=AspectType.FINISH,
value="Pointed",
element_instance=1,
aspect_instance=1,
quantity=65,
install_date=None,
renewal_year=2045,
source_system=None,
comments=None,
),
Element(
uprn=1,
element=ElementType.EXTERNAL_WALLS,
aspect_type=AspectType.FINISH,
value="Pointing",
element_instance=1,
aspect_instance=1,
quantity=1,
install_date=None,
renewal_year=2069,
source_system=None,
comments=None,
),
Element(
uprn=1,
element=ElementType.EXTERNAL_WALLS,
aspect_type=AspectType.FINISH,
value="Tile Hung",
element_instance=1,
aspect_instance=2,
quantity=8,
install_date=None,
renewal_year=2049,
source_system=None,
comments=None,
),
]
expected_condition_survey = PropertyConditionSurvey(
uprn=1,
elements=[
Element(
element_type=ElementType.EXTERNAL_WALL,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.FINISH,
aspect_instance=1,
value="Pointed",
quantity=65,
install_date=None,
renewal_year=2045,
comments=None,
),
AspectCondition(
aspect_type=AspectType.FINISH,
aspect_instance=1,
value="Pointing",
quantity=1,
install_date=None,
renewal_year=2069,
comments=None,
),
AspectCondition(
aspect_type=AspectType.FINISH,
aspect_instance=2,
value="Tile Hung",
quantity=8,
install_date=None,
renewal_year=2049,
comments=None,
),
],
),
],
date=date(2000, 1, 1), # what should this be?
source="Peabody",
)
# act
actual_assets = mapper.map_asset_conditions_for_property(peabody_property)
actual_condition_survey: PropertyConditionSurvey = (
mapper.map_asset_conditions_for_property(peabody_property)
)
# assert
assert len(actual_assets) == len(expected_assets)
for i, (actual, expected) in enumerate(zip(actual_assets, expected_assets)):
assert actual == expected, f"Mismatch at index {i}"
assert actual_condition_survey == expected_condition_survey