Map peabody data to new structure 🟩

This commit is contained in:
Daniel Roth 2026-01-27 14:30:23 +00:00
parent a0fa676230
commit 0bd5106cb4
2 changed files with 52 additions and 21 deletions

View file

@ -63,8 +63,10 @@ PEABODY_ELEMENT_MAP = {
element=ElementType.EXTERNAL_WALL, aspect_type=AspectType.FINISH
),
(53, 30): ElementMapping(
element=ElementType.SECONDARY_WALL, aspect_type=AspectType.FINISH
), # Should this be combined with primary wall, with different instance value?
element=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.FINISH,
aspect_instance=2,
),
(53, 36): ElementMapping(
element=ElementType.EXTERNAL_WALL, aspect_type=AspectType.INSULATION
),
@ -84,8 +86,8 @@ PEABODY_ELEMENT_MAP = {
element=ElementType.EXTERNAL_WALL, aspect_type=AspectType.FINISH
),
(120, 3): ElementMapping(
element=ElementType.PRIMARY_WALL, aspect_type=AspectType.INSULATION
), # This code element code is actually "WALL" not "external wall" - correct?
element=ElementType.EXTERNAL_WALL, aspect_type=AspectType.INSULATION
),
# ==========================================================
# EXTERNAL ROOFS
# ==========================================================

View file

@ -1,6 +1,9 @@
from typing import Any, List, Optional
from datetime import date
from backend.condition.domain.aspect_condition import AspectCondition
from backend.condition.domain.element import Element
from backend.condition.domain.element_type import ElementType
from backend.condition.domain.mapping.element_mapping import ElementMapping
from backend.condition.domain.mapping.peabody.peabody_element_map import (
PEABODY_ELEMENT_MAP,
@ -17,15 +20,12 @@ class PeabodyMapper(Mapper):
def map_asset_conditions_for_property(
self, client_property_data: Any, survey_year: Optional[int] = None
) -> PropertyConditionSurvey:
raise NotImplementedError
assert isinstance(
client_property_data, PeabodyProperty
) # TODO: think of a better way to do this
mapped_assets: List[Element] = []
mapped_elements: List[Element] = []
uprn: int = client_property_data.uprn
for raw_asset in client_property_data.assets:
try:
element_mapping: ElementMapping = PeabodyMapper._map_element(
@ -38,22 +38,51 @@ class PeabodyMapper(Mapper):
)
continue
mapped_assets.append(
Element(
uprn=uprn,
element_type=element_mapping.element,
aspect_type=element_mapping.aspect_type,
value=raw_asset.material_or_answer,
quantity=raw_asset.renewal_quantity,
install_date=None, # Not available in peabody data
renewal_year=raw_asset.renewal_year,
element_instance=element_mapping.element_instance,
source_system=None, # Once we know the system name we'll set it here
comments=None, # Not available in peabody data
aspect_condition = AspectCondition(
aspect_type=element_mapping.aspect_type,
aspect_instance=element_mapping.aspect_instance or 1,
value=raw_asset.material_or_answer,
quantity=raw_asset.renewal_quantity,
install_date=None, # Not available in peabody data
renewal_year=raw_asset.renewal_year,
comments=None, # Not available in peabody data
)
matching_element_type_instance: Optional[Element] = (
PeabodyMapper._check_for_element_type_and_instance(
mapped_elements,
element_mapping.element,
element_mapping.element_instance or 1,
)
)
return mapped_assets
if not matching_element_type_instance:
mapped_elements.append(
Element(
element_type=element_mapping.element,
element_instance=element_mapping.element_instance or 1,
aspect_conditions=[aspect_condition],
)
)
else:
matching_element_type_instance.aspect_conditions.append(
aspect_condition
)
return PropertyConditionSurvey(
uprn=client_property_data.uprn,
elements=mapped_elements,
date=date(2000, 1, 1), # Temp. Not sure how to get this
source="Peabody", # TODO: Make this the system, not the client
)
@staticmethod
def _check_for_element_type_and_instance(
elements: List[Element], type: ElementType, instance: int
) -> Optional[Element]:
for e in elements:
if e.element_type == type and e.element_instance == instance:
return e
return None
@staticmethod
def _map_element(element_code: int, sub_element_code: int) -> ElementMapping: