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

Merge pull request #691 from Hestia-Homes/feature/condition-data

Browse source 
Condition Data - map client data to custom data structure
This commit is contained in:
Daniel Roth 2026-01-28 10:42:22 +00:00 • committed by GitHub
commit 3874da6177
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
28 changed files with 3052 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

75
backend/condition/README.md Normal file
View file

@ -0,0 +1,75 @@
# Condition Data Processor
The Condition Data Processor performs the following steps:
- **Extract**
- Ingest client Condition Survey data files (currently from local files; future support planned for S3 and internal survey sources)
- Parse input files into Data Transfer Objects (DTOs)
- **Transform**
- Map source data into the internal domain data model
- **Load**
- Persist transformed data into the ARA database (not yet implemented)
The processor currently supports file formats provided by **Peabody** and **LBWF**.
---
## Running Locally
The `local_runner` script allows the processor to be executed in a local environment.
1. Copy a sample input file into the `sample_data/` directory.
2. Update `local_runner.py` as required, specifically the definitions of:
- `lbwf_path`
- `peabody_path`
- `file_paths`
3. Run `local_runner.py`.
Breakpoints may be added and the script run in debug mode if required.
---
## Known Data Issues
Some inconsistencies exist in the source datasets, primarily involving multiple representations of the same logical element within a single file. In these cases, assumptions have been made in order to normalise the data into the internal domain model.
### Peabody Data – Wall Finish Mapping
In the original Peabody sample dataset, multiple Element/Sub-Element combinations correspond to wall finishes:
| Element_Code | Element | Sub_Element_Code | Sub_Element |
|--------------|----------|------------------|-----------------------|
| 53 | External | 23 | Primary Wall Finish |
| 53 | External | 30 | Secondary Wall Finish |
| 120 | WALLS | 2 | Wall Finish |
A single property may contain records for all three combinations, and each combination may appear multiple times.
For example, the property at **55 Burnaby Street, London** contains entries for all three of the above combinations. However, it contains only a single entry for *“WALLS: Wall structure”*, indicating that the property has only one structure rather than multiple.
This pattern is also observed in other sampled properties. Based on this, the following assumption is applied:
- “Secondary” refers to a secondary **finish**, not a secondary **wall**.
As a result:
- The property is mapped to a single Wall element.
- That Wall element is assigned three Finish aspects:
- Two with `aspect_instance = 1`
- One with `aspect_instance = 2`
This means that the combination of
`UPRN / ElementType / ElementInstance / AspectType / AspectInstance`
is **not guaranteed to be unique**.
### LBWF Data – Wall Finish Mapping
In the LBWF dataset, the following element codes map to wall finishes:
- `EXTWALLFN1`
- `EXTWALLFN2`
These are similarly mapped as multiple instances of the **Finish** aspect for a single Wall element.
---

17
backend/condition/domain/aspect_condition.py Normal file
View file

@ -0,0 +1,17 @@
from dataclasses import dataclass
from typing import Optional
from datetime import date
from backend.condition.domain.aspect_type import AspectType
@dataclass
class AspectCondition:
aspect_type: AspectType
aspect_instance: int
value: Optional[str] = None
quantity: Optional[int] = None
install_date: Optional[date] = None
renewal_year: Optional[int] = None
comments: Optional[str] = None

35
backend/condition/domain/aspect_type.py Normal file
View file

@ -0,0 +1,35 @@
from enum import Enum
class AspectType(str, Enum):
MATERIAL = "material"
CONDITION = "condition"
TYPE = "type"
AREA = "area"
CONFIGURATION = "configuration"
PRESENCE = "presence"
RISK = "risk"
SEVERITY = "severity"
LOCATION = "location"
FINISH = "finish"
INSULATION = "insulation"
POINTING = "pointing"
SPALLING = "spalling"
LINTELS = "lintels"
CLADDING = "cladding"
CATEGORY = "category"
QUANTITY = "quantity"
ADEQUACY = "adequacy"
RATING = "rating"
STRATEGY = "strategy"
EXTENT = "extent"
DISTRIBUTION = "distribution"
STRUCTURE = "structure"
COVERING = "covering"
FIRE_RATING = "fire_rating"
EXTERNAL_DECORATION = "external_decoration"
WORK_REQUIRED = "work_required"
AGE_BAND = "age_band"
CONSTRUCTION_TYPE = "construction_type"
CLASSIFICATION = "classification"
SYSTEM = "system"

12
backend/condition/domain/element.py Normal file
View file

@ -0,0 +1,12 @@
from dataclasses import dataclass
from typing import List
from backend.condition.domain.aspect_condition import AspectCondition
from backend.condition.domain.element_type import ElementType
@dataclass
class Element:
element_type: ElementType
element_instance: int
aspect_conditions: List[AspectCondition]

263
backend/condition/domain/element_type.py Normal file
View file

@ -0,0 +1,263 @@
from enum import Enum
class ElementType(str, Enum):
# ======================
# PROPERTY / GENERAL
# ======================
PROPERTY = "property"
PROPERTY_CONSTRUCTION_TYPE = "property_construction_type"
PROPERTY_CLASSIFICATION = "property_classification"
PROPERTY_AGE_BAND = "property_age_band"
STOREY_COUNT = "storey_count"
FLOOR_LEVEL = "floor_level"
FLOOR_LEVEL_FRONT_DOOR = "floor_level_front_door"
ACCESSIBLE_HOUSING_REGISTER = "accessible_housing_register"
ASBESTOS = "asbestos"
QUALITY_STANDARD = "quality_standard"
CCU = "ccu"
PASSENGER_LIFT = "passenger_lift"
STAIRLIFT = "stairlift"
DISABLED_HOIST_TRACKING = "disabled_hoist_tracking"
DISABLED_FACILITIES = "disabled_facilities"
STEPS_TO_FRONT_DOOR = "steps_to_front_door"
# ======================
# EXTERNAL – ROOF
# ======================
ROOF = "roof"
PITCHED_ROOF_COVERING = "pitched_roof_covering"
FLAT_ROOF_COVERING = "flat_roof_covering"
RAINWATER_GOODS = "rainwater_goods"
LOFT_INSULATION = "loft_insulation"
PORCH_CANOPY = "porch_canopy"
CHIMNEY = "chimney"
FASCIA = "fascia"
SOFFIT = "soffit"
FASCIA_SOFFIT_BARGEBOARDS = "fascia_soffit_bargeboards"
GUTTERS = "gutters"
STORE_ROOF = "store_roof"
GARAGE_ROOF = "garage_roof"
GARAGE_AND_STORE_ROOF = "garage_and_store_roof"
# ======================
# EXTERNAL – WALLS
# ======================
EXTERNAL_WALL = "external_wall"
EXTERNAL_NOISE_INSULATION = "external_noise_insulation"
PRIMARY_WALL = "primary_wall"
SECONDARY_WALL = "secondary_wall"
DOWNPIPES = "downpipes"
EXTERNAL_DECORATION = "external_decoration"
CLADDING = "cladding"
SPANDREL_PANELS = "spandrel_panels"
GARAGE_WALLS = "garage_walls"
PARTY_WALL_FIRE_BREAK = "party_wall_fire_break"
EXTERNAL_BRICKWORK_POINTING = "external_brickwork_pointing"
INTERNAL_DOWNPIPES_EXTERNAL_AREA = "internal_downpipes_external_area"
# ======================
# EXTERNAL – WINDOWS
# ======================
EXTERNAL_WINDOWS = "external_windows"
COMMUNAL_WINDOWS = "communal_windows"
SECONDARY_GLAZING = "secondary_glazing"
STORE_WINDOWS = "store_windows"
GARAGE_WINDOWS = "garage_windows"
GARAGE_AND_STORE_WINDOWS = "garage_and_store_windows"
# ======================
# EXTERNAL – DOORS
# ======================
EXTERNAL_DOOR = "external_door"
FRONT_DOOR = "front_door"
REAR_DOOR = "rear_door"
STORE_DOOR = "store_door"
GARAGE_DOOR = "garage_door"
GARAGE_AND_STORE_DOOR = "garage_and_store_door"
COMMUNAL_ENTRANCE_DOOR = "communal_entrance_door"
MAIN_DOOR = "main_door"
BLOCK_ENTRANCE_DOOR = "block_entrance_door"
LINTEL = "lintel"
PATIO_FRENCH_DOOR = "patio_french_door"
DOOR_ENTRY_HANDSET = "door_entry_handset"
# ======================
# EXTERNAL – AREAS
# ======================
PATHS_AND_HARDSTANDINGS = "paths_and_hardstandings"
PARKING_AREAS = "parking_areas"
BOUNDARY_WALLS = "boundary_walls"
FRONT_FENCING = "front_fencing"
REAR_FENCING = "rear_fencing"
SIDE_FENCING = "side_fencing"
REAR_GATE = "rear_gate"
FRONT_GATE = "front_gate"
GATES = "gates"
RETAINING_WALLS = "retaining_walls"
PRIVATE_BALCONY = "private_balcony"
BALCONY_BALUSTRADE = "balcony_balustrade"
OUTBUILDINGS = "outbuildings"
GARAGE_STRUCTURE = "garage_structure"
PAVING = "paving"
ROADS = "roads"
SOIL_AND_VENT = "soil_and_vent"
SOLAR_THERMALS = "solar_thermals"
DROP_KERB = "drop_kerb"
OUTBUILDING_OVERHAUL = "outbuilding_overhaul"
EXTERNAL_STRUCTURAL_DEFECTS = "external_structural_defects"
ACCESS_RAMP = "access_ramp"
# ======================
# INTERNAL – KITCHEN
# ======================
KITCHEN = "kitchen"
KITCHEN_SPACE_LAYOUT = "kitchen_space_layout"
TENANT_INSTALLED_KITCHEN = "tenant_installed_kitchen"
KITCHEN_EXTRACTOR_FAN = "kitchen_extractor_fan"
# ======================
# INTERNAL – BATHROOM
# ======================
BATHROOM = "bathroom"
SECONDARY_BATHROOM = "secondary_bathroom"
SECONDARY_TOILET = "secondary_toilet"
BATHROOM_EXTRACTOR_FAN = "bathroom_extractor_fan"
ADDITIONAL_WC_OR_WHB = "additional_wc_or_whb"
BATHROOM_REMAINING_LIFE_SOURCE = "bathroom_remaining_life_source"
KITCHEN_REMAINING_LIFE_SOURCE = "kitchen_remaining_life_source"
# ======================
# INTERNAL – HEATING / WATER
# ======================
CENTRAL_HEATING = "central_heating"
HEATING_BOILER = "heating_boiler"
HEATING_DISTRIBUTION = "heating_distribution"
SECONDARY_HEATING = "secondary_heating"
HOT_WATER_SYSTEM = "hot_water_system"
COLD_WATER_STORAGE = "cold_water_storage"
HEATING_SYSTEM = "heating_system"
BOILER_FUEL = "boiler_fuel"
WATER_HEATING = "water_heating"
PROGRAMMABLE_HEATING = "programmable_heating"
COMMUNITY_HEATING = (
"community_heating" # Is this definitely different from COMMUNAL_HEATING?
)
GAS_AVAILABLE = "gas_available"
HEAT_RECOVERY_UNITS = "heat_recovery_units"
HEATING_IMPROVEMENTS = "heating_improvements"
# ======================
# INTERNAL – ELECTRICS / FIRE
# ======================
ELECTRICAL_WIRING = "electrical_wiring"
CONSUMER_UNIT = "consumer_unit"
SMOKE_DETECTION = "smoke_detection"
HEAT_DETECTION = "heat_detection"
CARBON_MONOXIDE_DETECTION = "carbon_monoxide_detection"
FIRE_DOOR_RATING = "fire_door_rating"
FIRE_RISK_ASSESSMENT = "fire_risk_assessment"
INTERNAL_WIRING = (
"internal_wiring" # Is this definitely different from ELECTRICAL_WIRING?
)
ELECTRICS = "electrics"
# ======================
# COMMUNAL
# ======================
COMMUNAL_HEATING = "communal_heating"
COMMUNAL_BOILER = "communal_boiler"
COMMUNAL_ELECTRICS = "communal_electrics"
COMMUNAL_FIRE_ALARM = "communal_fire_alarm"
COMMUNAL_EMERGENCY_LIGHTING = "communal_emergency_lighting"
COMMUNAL_DOOR_ENTRY = "communal_door_entry"
COMMUNAL_CCTV = "communal_cctv"
COMMUNAL_BIN_STORE = "communal_bin_store"
COMMUNAL_BIN_STORE_DOORS = "communal_bin_store_doors"
COMMUNAL_BIN_STORE_WALLS = "communal_bin_store_walls"
COMMUNAL_BIN_STORE_ROOF = "communal_bin_store_roof"
COMMUNAL_REFUSE_CHUTE = "communal_refuse_chute"
COMMUNAL_FLOOR_COVERING = "communal_floor_covering"
COMMUNAL_KITCHEN = "communal_kitchen"
COMMUNAL_BATHROOM = "communal_bathroom"
COMMUNAL_TOILETS = "communal_toilets"
COMMUNAL_GATES = "communal_gates"
COMMUNAL_LIFT = "communal_lift"
COMMUNAL_PASSENGER_LIFT = "communal_passenger_lift"
COMMUNAL_BALCONY_WALKWAY = "communal_balcony_walkway"
COMMUNAL_ENTRANCE = "communal_entrance"
COMMUNAL_INTERNAL_DECORATIONS = "communal_internal_decorations"
COMMUNAL_INTERNAL_FLOOR = "communal_internal_floor"
COMMUNAL_WALKWAYS = "communal_walkways"
COMMUNAL_EXTERNAL_DOORS = "communal_external_doors"
COMMUNAL_STAIRS = "communal_stairs"
COMMUNAL_AERIAL = "communal_aerial"
COMMUNAL_AOV = "communal_aov"
COMMUNAL_INTERNAL_DOORS = "communal_internal_doors"
COMMUNAL_LATERAL_MAINS = "communal_lateral_mains"
COMMUNAL_LIGHTING = "communal_lighting"
COMMUNAL_LIGHTING_CONDUCTOR = "communal_lighting_conductor"
COMMUNAL_STORE_ROOF = "communal_store_roof"
COMMUNAL_STORE_WALLS = "communal_store_walls"
COMMUNAL_STORE_DOORS = "communal_store_doors"
COMMUNAL_WARDEN_CALL_SYSTEM = "communal_warden_call_system"
COMMUNAL_BMS = "communal_bms"
COMMUNAL_BOOSTER_PUMP = "communal_booster_pump"
COMMUNAL_DRY_RISER = "communal_dry_riser"
COMMUNAL_WET_RISER = "communal_wet_riser"
COMMUNAL_COLD_WATER_STORAGE = "communal_cold_water_storage"
COMMUNAL_SPRINKLER = "communal_sprinkler"
COMMUNAL_PLUG_SOCKETS = "communal_plug_sockets"
COMMUNAL_CIRCULATION_SPACE = "communal_circulation_space"
# ======================
# FITNESS FOR HUMAN HABITATION
# ======================
FFHH_DAMP = "ffhh_damp"
FFHH_HOT_AND_COLD_WATER = "ffhh_hold_and_cold_water"
FFHH_DRAINAGE_LAVATORIES = "ffhh_drainage_lavatories"
FFHH_NEGLECTED = "ffhh_neglected"
FFHH_NATURAL_LIGHT = "ffhh_natural_light"
FFHH_VENTILATION = "ffhh_ventilation"
FFHH_FOOD_PREP_AND_WASHUP = "ffhh_food_prep_and_washup"
FFHH_UNSAFE_LAYOUT = "ffhh_unsafe_layout"
FFHH_UNSTABLE_BUILDING = "ffhh_unstable_building"
# ==========================================================
# HHSRS – ALL 29 HAZARDS
# ==========================================================
# TODO: In order to group HHSRS, should there be a single HHSRS element type, and each of the below is an AspectType?
HHSRS_DAMP_AND_MOULD = "hhsrs_damp_and_mould"
HHSRS_EXCESS_COLD = "hhsrs_excess_cold"
HHSRS_EXCESS_HEAT = "hhsrs_excess_heat"
HHSRS_ASBESTOS_AND_MMF = "hhsrs_asbestos_and_mmf"
HHSRS_BIOCIDES = "hhsrs_biocides"
HHSRS_CARBON_MONOXIDE = "hhsrs_carbon_monoxide"
HHSRS_LEAD = "hhsrs_lead"
HHSRS_RADIATION = "hhsrs_radiation"
HHSRS_UNCOMBUSTED_FUEL_GAS = "hhsrs_uncombusted_fuel_gas"
HHSRS_VOLATILE_ORGANIC_COMPOUNDS = "hhsrs_volatile_organic_compounds"
HHSRS_CROWDING_AND_SPACE = "hhsrs_crowding_and_space"
HHSRS_ENTRY_BY_INTRUDERS = "hhsrs_entry_by_intruders"
HHSRS_LIGHTING = "hhsrs_lighting"
HHSRS_NOISE = "hhsrs_noise"
HHSRS_DOMESTIC_HYGIENE_PESTS_REFUSE = "hhsrs_domestic_hygiene_pests_refuse"
HHSRS_FOOD_SAFETY = "hhsrs_food_safety"
HHSRS_PERSONAL_HYGIENE_SANITATION = "hhsrs_personal_hygiene_sanitation"
HHSRS_WATER_SUPPLY = "hhsrs_water_supply"
HHSRS_FALLS_ASSOCIATED_WITH_BATHS = "hhsrs_falls_associated_with_baths"
HHSRS_FALLS_ON_LEVEL_SURFACES = "hhsrs_falls_on_level_surfaces"
HHSRS_FALLS_ON_STAIRS = "hhsrs_falls_on_stairs"
HHSRS_FALLS_BETWEEN_LEVELS = "hhsrs_falls_between_levels"
HHSRS_ELECTRICAL_HAZARDS = "hhsrs_electrical_hazards"
HHSRS_FIRE = "hhsrs_fire"
HHSRS_FLAMES_HOT_SURFACES = "hhsrs_flames_hot_surfaces"
HHSRS_COLLISION_AND_ENTRAPMENT = "hhsrs_collision_and_entrapment"
HHSRS_COLLISION_HAZARDS_LOW_HEADROOM = "hhsrs_collision_hazards_low_headroom"
HHSRS_EXPLOSIONS = "hhsrs_explosions"
HHSRS_ERGONOMICS = "hhsrs_ergonomics"
HHSRS_STRUCTURAL_COLLAPSE = "hhsrs_structural_collapse"
HHSRS_AMENITIES = "hhsrs_amenities"

13
backend/condition/domain/mapping/element_mapping.py Normal file
View file

@ -0,0 +1,13 @@
from dataclasses import dataclass
from typing import Optional
from backend.condition.domain.aspect_type import AspectType
from backend.condition.domain.element_type import ElementType
@dataclass(frozen=True)
class ElementMapping:
elementType: ElementType
aspect_type: AspectType
element_instance: Optional[int] = None
aspect_instance: Optional[int] = None

531
backend/condition/domain/mapping/lbwf/lbwf_element_map.py Normal file
View file

@ -0,0 +1,531 @@
from backend.condition.domain.element_type import ElementType
from backend.condition.domain.aspect_type import AspectType
from backend.condition.domain.mapping.element_mapping import ElementMapping
LBWF_ELEMENT_MAP: dict[str, ElementMapping] = {
# ==========================================================
# PROPERTY / GENERAL
# ==========================================================
"AHR_CAT": ElementMapping(
elementType=ElementType.ACCESSIBLE_HOUSING_REGISTER,
aspect_type=AspectType.CATEGORY,
),
"ASSETSAREA": ElementMapping(
elementType=ElementType.PROPERTY,
aspect_type=AspectType.AREA,
),
# "DECNTHMINC": ElementMapping(
# element=Element.DECENT_HOMES,
# aspect_type=AspectType.INCLUSION,
# ), # Ignore this one
"QUALITYSTD": ElementMapping(
elementType=ElementType.QUALITY_STANDARD,
aspect_type=AspectType.TYPE,
),
"EXTSTOREY": ElementMapping(
elementType=ElementType.PROPERTY,
aspect_type=AspectType.CONFIGURATION,
),
"FLVL": ElementMapping(
elementType=ElementType.FLOOR_LEVEL_FRONT_DOOR,
aspect_type=AspectType.LOCATION,
),
"INTFLRLVL": ElementMapping(
elementType=ElementType.FLOOR_LEVEL,
aspect_type=AspectType.LOCATION,
),
"INTNSEINSL": ElementMapping(
elementType=ElementType.EXTERNAL_NOISE_INSULATION, # Maybe this shouldn't be "EXTERNAL_"
aspect_type=AspectType.ADEQUACY,
),
"INTSTEPSFD": ElementMapping(
elementType=ElementType.STEPS_TO_FRONT_DOOR,
aspect_type=AspectType.QUANTITY,
),
# ==========================================================
# ASBESTOS (NON-HHSRS RECORD)
# ==========================================================
"ASBESTOS": ElementMapping(
elementType=ElementType.ASBESTOS,
aspect_type=AspectType.PRESENCE,
),
# ==========================================================
# INTERNAL – BATHROOMS & KITCHENS
# ==========================================================
"INTBTHRLOC": ElementMapping(
elementType=ElementType.BATHROOM,
aspect_type=AspectType.LOCATION,
),
"INTBTHADEQ": ElementMapping(
elementType=ElementType.BATHROOM,
aspect_type=AspectType.ADEQUACY,
),
"INTKITADEQ": ElementMapping(
elementType=ElementType.KITCHEN,
aspect_type=AspectType.ADEQUACY,
),
"INTCKRLOC": ElementMapping(
elementType=ElementType.KITCHEN,
aspect_type=AspectType.LOCATION,
),
"INTADDWCW": ElementMapping(
elementType=ElementType.ADDITIONAL_WC_OR_WHB,
aspect_type=AspectType.PRESENCE,
),
"INTBTHREML": ElementMapping(
elementType=ElementType.BATHROOM_REMAINING_LIFE_SOURCE,
aspect_type=AspectType.TYPE,
),
"INTKITREML": ElementMapping(
elementType=ElementType.KITCHEN_REMAINING_LIFE_SOURCE,
aspect_type=AspectType.TYPE,
),
"INTTNTINST": ElementMapping(
elementType=ElementType.TENANT_INSTALLED_KITCHEN,
aspect_type=AspectType.TYPE, # Not certain about this aspect type - need more data
),
# ==========================================================
# INTERNAL – FIRE
# ==========================================================
"FRARISKRTG": ElementMapping(
elementType=ElementType.FIRE_RISK_ASSESSMENT,
aspect_type=AspectType.RATING,
),
"FRATYPE": ElementMapping(
elementType=ElementType.FIRE_RISK_ASSESSMENT,
aspect_type=AspectType.TYPE,
),
"FRAEVACSTR": ElementMapping(
elementType=ElementType.FIRE_RISK_ASSESSMENT,
aspect_type=AspectType.STRATEGY,
),
"INTSMKDET": ElementMapping(
elementType=ElementType.SMOKE_DETECTION,
aspect_type=AspectType.PRESENCE,
),
"INTCHEXTNT": ElementMapping(
elementType=ElementType.HEATING_SYSTEM,
aspect_type=AspectType.EXTENT,
),
# ==========================================================
# HEATING & SERVICES
# ==========================================================
"INTCHEXTNT": ElementMapping(
elementType=ElementType.CENTRAL_HEATING,
aspect_type=AspectType.EXTENT,
),
"INTCHDIST": ElementMapping(
elementType=ElementType.HEATING_DISTRIBUTION,
aspect_type=AspectType.TYPE,
),
"INTCHBLR": ElementMapping(
elementType=ElementType.HEATING_BOILER,
aspect_type=AspectType.TYPE,
),
"INTBOILERF": ElementMapping(
elementType=ElementType.BOILER_FUEL,
aspect_type=AspectType.TYPE,
),
"INTHTDISYS": ElementMapping(
elementType=ElementType.HEATING_SYSTEM,
aspect_type=AspectType.DISTRIBUTION,
),
"INTWTRHTNG": ElementMapping(
elementType=ElementType.WATER_HEATING,
aspect_type=AspectType.TYPE,
),
"INTCOMHTG": ElementMapping(
elementType=ElementType.COMMUNITY_HEATING,
aspect_type=AspectType.TYPE,
),
"INTELECTRC": ElementMapping(
elementType=ElementType.ELECTRICS,
aspect_type=AspectType.WORK_REQUIRED, # Not certain about this aspect type - need more data
),
"INTGASAVAI": ElementMapping(
elementType=ElementType.GAS_AVAILABLE,
aspect_type=AspectType.PRESENCE, # Maybe should be AspectType.TYPE ?
),
"INTHEATREC": ElementMapping(
elementType=ElementType.HEAT_RECOVERY_UNITS,
aspect_type=AspectType.PRESENCE,
),
"INTHTIMP": ElementMapping(
elementType=ElementType.GAS_AVAILABLE,
aspect_type=AspectType.WORK_REQUIRED,
),
"INTPROGHTG": ElementMapping(
elementType=ElementType.PROGRAMMABLE_HEATING,
aspect_type=AspectType.TYPE, # Should maybe be PRESENCE, but set to TYPE for consistency with Peabody data
),
# ==========================================================
# EXTERNAL – WALLS (INSTANCED)
# ==========================================================
"EXTWALLSTR": ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.STRUCTURE,
),
"EXTWALLFN1": ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.FINISH,
),
"EXTWALLFN2": ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.FINISH,
aspect_instance=2,
),
"EXTWALLINS": ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.INSULATION,
),
"EXTWALLSPL": ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.CONDITION,
),
"EXTDWNPTYP": ElementMapping(
elementType=ElementType.DOWNPIPES,
aspect_type=AspectType.MATERIAL,
),
"EXTGUTRTYP": ElementMapping(
elementType=ElementType.GUTTERS,
aspect_type=AspectType.MATERIAL,
),
# ==========================================================
# EXTERNAL – ROOFS (INSTANCED)
# ==========================================================
"EXTRFSTR1": ElementMapping(
elementType=ElementType.ROOF,
aspect_type=AspectType.STRUCTURE,
),
"EXTRFSTR2": ElementMapping(
elementType=ElementType.ROOF,
aspect_type=AspectType.STRUCTURE,
aspect_instance=2,
),
"EXTRFSTR3": ElementMapping(
elementType=ElementType.ROOF,
aspect_type=AspectType.STRUCTURE,
aspect_instance=3,
),
"EXTROOF1": ElementMapping(
elementType=ElementType.ROOF,
aspect_type=AspectType.MATERIAL,
),
"EXTROOF2": ElementMapping(
elementType=ElementType.ROOF,
aspect_type=AspectType.MATERIAL,
aspect_instance=2,
),
"EXTROOF3": ElementMapping(
elementType=ElementType.ROOF,
aspect_type=AspectType.MATERIAL,
aspect_instance=3,
),
"EXTCHIMNEY": ElementMapping(
elementType=ElementType.CHIMNEY,
aspect_type=AspectType.WORK_REQUIRED,
),
"EXTFASOFBR": ElementMapping(
elementType=ElementType.FASCIA_SOFFIT_BARGEBOARDS,
aspect_type=AspectType.MATERIAL,
),
"EXTGARROOF": ElementMapping(
elementType=ElementType.GARAGE_ROOF,
aspect_type=AspectType.MATERIAL,
),
"EXTGARSTRF": ElementMapping(
elementType=ElementType.GARAGE_AND_STORE_ROOF,
aspect_type=AspectType.MATERIAL,
),
"EXTSTRROOF": ElementMapping(
elementType=ElementType.STORE_ROOF,
aspect_type=AspectType.MATERIAL,
),
"INTLOFTINS": ElementMapping(
elementType=ElementType.LOFT_INSULATION,
aspect_type=AspectType.TYPE,
),
# ==========================================================
# EXTERNAL – DOORS & WINDOWS
# ==========================================================
"INTFRDOOR": ElementMapping(
elementType=ElementType.EXTERNAL_DOOR,
aspect_type=AspectType.TYPE,
),
"INTFRDRFRR": ElementMapping(
elementType=ElementType.EXTERNAL_DOOR,
aspect_type=AspectType.FIRE_RATING,
),
"EXTBKSDDR1": ElementMapping(
elementType=ElementType.EXTERNAL_DOOR,
aspect_type=AspectType.TYPE,
),
"EXTBKSDDR2": ElementMapping(
elementType=ElementType.EXTERNAL_DOOR,
aspect_type=AspectType.TYPE,
aspect_instance=2,
),
"INTWDWTYPE": ElementMapping(
elementType=ElementType.EXTERNAL_WINDOWS,
aspect_type=AspectType.TYPE,
),
"EXTWNDWS1": ElementMapping(
elementType=ElementType.EXTERNAL_WINDOWS,
aspect_type=AspectType.TYPE,
),
"EXTWNDWS2": ElementMapping(
elementType=ElementType.EXTERNAL_WINDOWS,
aspect_type=AspectType.TYPE,
aspect_instance=2,
),
"EXTGARDOOR": ElementMapping(
elementType=ElementType.GARAGE_DOOR,
aspect_type=AspectType.MATERIAL,
),
"EXTGARSTDR": ElementMapping(
elementType=ElementType.GARAGE_AND_STORE_DOOR,
aspect_type=AspectType.MATERIAL,
),
"EXTSTRDOOR": ElementMapping(
elementType=ElementType.STORE_DOOR,
aspect_type=AspectType.MATERIAL,
),
"EXTGARWDWS": ElementMapping(
elementType=ElementType.GARAGE_WINDOWS,
aspect_type=AspectType.MATERIAL,
),
"EXTSTRWDWS": ElementMapping(
elementType=ElementType.STORE_WINDOWS,
aspect_type=AspectType.MATERIAL,
),
"EXTGARSTWD": ElementMapping(
elementType=ElementType.GARAGE_AND_STORE_WINDOWS,
aspect_type=AspectType.MATERIAL,
),
"EXTLINTELS": ElementMapping(
elementType=ElementType.LINTEL,
aspect_type=AspectType.PRESENCE,
),
"EXTPTFRDR1": ElementMapping(
elementType=ElementType.PATIO_FRENCH_DOOR,
aspect_type=AspectType.MATERIAL,
),
# ==========================================================
# EXTERNAL AREAS
# ==========================================================
"EXTBALCONY": ElementMapping(
elementType=ElementType.PRIVATE_BALCONY,
aspect_type=AspectType.PRESENCE,
),
"EXTBPOINTG": ElementMapping(
elementType=ElementType.EXTERNAL_BRICKWORK_POINTING,
aspect_type=AspectType.PRESENCE,
),
"EXTDRPKERB": ElementMapping(
elementType=ElementType.DROP_KERB,
aspect_type=AspectType.PRESENCE,
),
"EXTEXTDECS": ElementMapping(
elementType=ElementType.EXTERNAL_DECORATION,
aspect_type=AspectType.PRESENCE,
),
"EXTHARDSTD": ElementMapping(
elementType=ElementType.PATHS_AND_HARDSTANDINGS,
aspect_type=AspectType.MATERIAL,
),
"EXTINTDWNP": ElementMapping(
elementType=ElementType.INTERNAL_DOWNPIPES_EXTERNAL_AREA,
aspect_type=AspectType.MATERIAL,
),
"EXTOUTBOH": ElementMapping(
elementType=ElementType.OUTBUILDING_OVERHAUL,
aspect_type=AspectType.TYPE,
),
"EXTPARKING": ElementMapping(
elementType=ElementType.PARKING_AREAS,
aspect_type=AspectType.PRESENCE,
),
"EXTPCHCNPY": ElementMapping(
elementType=ElementType.PORCH_CANOPY,
aspect_type=AspectType.TYPE,
),
"EXTSTRINSP": ElementMapping(
elementType=ElementType.EXTERNAL_STRUCTURAL_DEFECTS,
aspect_type=AspectType.TYPE, # Need more sample data to know whether this is the correct aspect type
),
"INTACCRAMP": ElementMapping(
elementType=ElementType.ACCESS_RAMP,
aspect_type=AspectType.TYPE, # # Need more sample data to know whether this is the correct aspect type
),
# ======================
# FITNESS FOR HUMAN HABITATION
# ======================
"FFHHDAMP": ElementMapping(
elementType=ElementType.FFHH_DAMP,
aspect_type=AspectType.RISK,
),
"FFHHHCWAT": ElementMapping(
elementType=ElementType.FFHH_HOT_AND_COLD_WATER,
aspect_type=AspectType.RISK,
),
"FFHHDRNWC": ElementMapping(
elementType=ElementType.FFHH_DRAINAGE_LAVATORIES,
aspect_type=AspectType.RISK,
),
"FFHHNEGLC": ElementMapping(
elementType=ElementType.FFHH_NEGLECTED,
aspect_type=AspectType.RISK,
),
"FFHHNONAT": ElementMapping(
elementType=ElementType.FFHH_NATURAL_LIGHT,
aspect_type=AspectType.RISK,
),
"FFHHNOVEN": ElementMapping(
elementType=ElementType.FFHH_VENTILATION,
aspect_type=AspectType.RISK,
),
"FFHHPRPCK": ElementMapping(
elementType=ElementType.FFHH_FOOD_PREP_AND_WASHUP,
aspect_type=AspectType.RISK,
),
"FFHHUNLAY": ElementMapping(
elementType=ElementType.FFHH_UNSAFE_LAYOUT,
aspect_type=AspectType.RISK,
),
"FFHHUNSTA": ElementMapping(
elementType=ElementType.FFHH_UNSTABLE_BUILDING,
aspect_type=AspectType.RISK,
),
# ==========================================================
# HHSRS
# ==========================================================
"HHSRSDAMP": ElementMapping(
elementType=ElementType.HHSRS_DAMP_AND_MOULD,
aspect_type=AspectType.RISK,
),
"HHSRSCOLD": ElementMapping(
elementType=ElementType.HHSRS_EXCESS_COLD,
aspect_type=AspectType.RISK,
),
"HHSRSHEAT": ElementMapping(
elementType=ElementType.HHSRS_EXCESS_HEAT,
aspect_type=AspectType.RISK,
),
"HHSRSASB": ElementMapping(
elementType=ElementType.HHSRS_ASBESTOS_AND_MMF,
aspect_type=AspectType.RISK,
),
"HHSRSBIOC": ElementMapping(
elementType=ElementType.HHSRS_BIOCIDES,
aspect_type=AspectType.RISK,
),
"HHSRSCO": ElementMapping(
elementType=ElementType.HHSRS_CARBON_MONOXIDE,
aspect_type=AspectType.RISK,
),
"HHSRSNO2": ElementMapping(
elementType=ElementType.HHSRS_CARBON_MONOXIDE,
aspect_type=AspectType.RISK,
), # Duplicate of HHSRSCO; I think they relate to the same HHSRS hazard
"HHSRSSO2": ElementMapping(
elementType=ElementType.HHSRS_CARBON_MONOXIDE,
aspect_type=AspectType.RISK,
), # Duplicate of HHSRSCO; I think they relate to the same HHSRS hazard
"HHSRSLEAD": ElementMapping(
elementType=ElementType.HHSRS_LEAD,
aspect_type=AspectType.RISK,
),
"HHSRSRADIA": ElementMapping(
elementType=ElementType.HHSRS_RADIATION,
aspect_type=AspectType.RISK,
),
"HHSRSFUEL": ElementMapping(
elementType=ElementType.HHSRS_UNCOMBUSTED_FUEL_GAS,
aspect_type=AspectType.RISK,
),
"HHSRSORGAN": ElementMapping(
elementType=ElementType.HHSRS_VOLATILE_ORGANIC_COMPOUNDS,
aspect_type=AspectType.RISK,
),
"HHSRSCROWD": ElementMapping(
elementType=ElementType.HHSRS_CROWDING_AND_SPACE,
aspect_type=AspectType.RISK,
),
"HHSRSENTRY": ElementMapping(
elementType=ElementType.HHSRS_ENTRY_BY_INTRUDERS,
aspect_type=AspectType.RISK,
),
"HHSRSLIGHT": ElementMapping(
elementType=ElementType.HHSRS_LIGHTING,
aspect_type=AspectType.RISK,
),
"HHSRSNOISE": ElementMapping(
elementType=ElementType.HHSRS_NOISE,
aspect_type=AspectType.RISK,
),
"HHSRSDOMES": ElementMapping(
elementType=ElementType.HHSRS_DOMESTIC_HYGIENE_PESTS_REFUSE,
aspect_type=AspectType.RISK,
),
"HHSRSFOOD": ElementMapping(
elementType=ElementType.HHSRS_FOOD_SAFETY,
aspect_type=AspectType.RISK,
),
"HHSRSPERS": ElementMapping(
elementType=ElementType.HHSRS_PERSONAL_HYGIENE_SANITATION,
aspect_type=AspectType.RISK,
),
"HHSRSWATER": ElementMapping(
elementType=ElementType.HHSRS_WATER_SUPPLY,
aspect_type=AspectType.RISK,
),
"HHSRSFBATH": ElementMapping(
elementType=ElementType.HHSRS_FALLS_ASSOCIATED_WITH_BATHS,
aspect_type=AspectType.RISK,
),
"HHSRSFLEVE": ElementMapping(
elementType=ElementType.HHSRS_FALLS_ON_LEVEL_SURFACES,
aspect_type=AspectType.RISK,
),
"HHSRSFSTAI": ElementMapping(
elementType=ElementType.HHSRS_FALLS_ON_STAIRS,
aspect_type=AspectType.RISK,
),
"HHSRSFBETW": ElementMapping(
elementType=ElementType.HHSRS_FALLS_BETWEEN_LEVELS,
aspect_type=AspectType.RISK,
),
"HHSRSELEC": ElementMapping(
elementType=ElementType.HHSRS_ELECTRICAL_HAZARDS,
aspect_type=AspectType.RISK,
),
"HHSRSFIRE": ElementMapping(
elementType=ElementType.HHSRS_FIRE,
aspect_type=AspectType.RISK,
),
"HHSRSFLAME": ElementMapping(
elementType=ElementType.HHSRS_FLAMES_HOT_SURFACES,
aspect_type=AspectType.RISK,
),
"HHSRSENTRP": ElementMapping(
elementType=ElementType.HHSRS_COLLISION_AND_ENTRAPMENT,
aspect_type=AspectType.RISK,
),
"HHSRSEXPLO": ElementMapping(
elementType=ElementType.HHSRS_EXPLOSIONS,
aspect_type=AspectType.RISK,
),
"HHSRSSTRUC": ElementMapping(
elementType=ElementType.HHSRS_STRUCTURAL_COLLAPSE,
aspect_type=AspectType.RISK,
),
"HHSRSCLOW": ElementMapping(
elementType=ElementType.HHSRS_COLLISION_AND_ENTRAPMENT,
aspect_type=AspectType.RISK,
),
"HHSRSPOSI": ElementMapping(
elementType=ElementType.HHSRS_AMENITIES,
aspect_type=AspectType.RISK,
),
}

128
backend/condition/domain/mapping/lbwf/lbwf_mapper.py Normal file
View file

@ -0,0 +1,128 @@
from typing import Any, Dict, List, Optional, Tuple
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.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,
)
from backend.condition.parsing.records.lbwf.lbwf_house import LbwfHouse
from utils.logger import setup_logger
logger = setup_logger()
class LbwfMapper(Mapper):
def map_asset_conditions_for_property(
self, client_property_data: Any, survey_year: Optional[int] = None
) -> PropertyConditionSurvey:
assert isinstance(
client_property_data, LbwfHouse
) # TODO: think of a better way to do this
elements_by_key: dict[tuple[ElementType, int], Element] = {}
for raw_asset in client_property_data.assets:
if raw_asset.element_code in ["DECNTHMINC", "EICINSFREQ"]:
# skip metadata rows
continue
element_mapping = LbwfMapper._safe_map_element(raw_asset)
if not element_mapping:
continue
aspect_condition = LbwfMapper._build_aspect_condition(
raw_asset, element_mapping, survey_year
)
element_key = (
element_mapping.elementType,
element_mapping.element_instance or 1,
)
LbwfMapper._attach_aspect_condition_to_element(
elements_by_key, element_key, aspect_condition
)
return PropertyConditionSurvey(
uprn=client_property_data.uprn,
elements=list(elements_by_key.values()),
date=date(2000, 1, 1), # Temp - not sure how to get this
source="LBWF", # TODO: Make this the system, not the client
)
@staticmethod
def _safe_map_element(raw_asset: LbwfAssetCondition) -> Optional[ElementMapping]:
try:
return LbwfMapper._map_element(raw_asset.element_code)
except KeyError:
logger.warning(
logger.warning(
f"Unrecognised LBWF Asset Element: "
f"{raw_asset.element_code} ({raw_asset.element_code_description})). "
"Skipping record"
)
)
return None
@staticmethod
def _map_element(lbwf_element_code: str) -> ElementMapping:
return LBWF_ELEMENT_MAP[lbwf_element_code]
@staticmethod
def _build_aspect_condition(
raw_asset, element_mapping: ElementMapping, survey_year: int
) -> AspectCondition:
return AspectCondition(
aspect_type=element_mapping.aspect_type,
aspect_instance=element_mapping.aspect_instance or 1,
value=raw_asset.attribute_code_description,
quantity=raw_asset.quantity,
install_date=raw_asset.install_date,
renewal_year=LbwfMapper._calculate_renewal_year(raw_asset, survey_year),
comments=raw_asset.element_comments,
)
@staticmethod
def _attach_aspect_condition_to_element(
elements_by_key: Dict[Tuple[ElementType, int], Element],
element_key: Tuple[ElementType, int],
aspect_condition: AspectCondition,
) -> None:
element = elements_by_key.get(element_key)
if element is None:
element = Element(
element_type=element_key[0],
element_instance=element_key[1],
aspect_conditions=[],
)
elements_by_key[element_key] = element
element.aspect_conditions.append(aspect_condition)
@staticmethod
def _calculate_renewal_year(
lbwf_asset: LbwfAssetCondition, survey_year: Optional[int]
) -> Optional[int]:
remaining_life_years: Optional[int] = lbwf_asset.remaining_life
if not remaining_life_years:
return None
if not survey_year:
return None
try:
return survey_year + remaining_life_years
except:
logger.debug(
f"Unable to map LBWF Asset remaining life {remaining_life_years} to renewal year, returning None"
)
return None

15
backend/condition/domain/mapping/mapper.py Normal file
View file

@ -0,0 +1,15 @@
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_property_data: Any, survey_year: Optional[int] = None
) -> PropertyConditionSurvey:
# TODO: client_data should be properly typed
pass

693
backend/condition/domain/mapping/peabody/peabody_element_map.py Normal file
View file

@ -0,0 +1,693 @@
from backend.condition.domain.aspect_type import AspectType
from backend.condition.domain.element_type import ElementType
from backend.condition.domain.mapping.element_mapping import ElementMapping
PEABODY_ELEMENT_MAP = {
# ==========================================================
# PROPERTY / GENERAL
# ==========================================================
(100, 1): ElementMapping(
elementType=ElementType.PROPERTY,
aspect_type=AspectType.TYPE,
),
# (100, 3): ElementMapping(element=Element.PROPERTY, aspect_type=AspectType.AGE),
# (100, 14): ElementMapping(element="property", aspect_type="construction_type"),
(50, 2): ElementMapping(
elementType=ElementType.CARBON_MONOXIDE_DETECTION,
aspect_type=AspectType.TYPE,
),
(50, 3): ElementMapping(
elementType=ElementType.CCU,
aspect_type=AspectType.TYPE,
),
(50, 7): ElementMapping(
elementType=ElementType.DISABLED_HOIST_TRACKING,
aspect_type=AspectType.PRESENCE,
),
(50, 11): ElementMapping(
elementType=ElementType.HEAT_DETECTION,
aspect_type=AspectType.TYPE,
),
(50, 21): ElementMapping(
elementType=ElementType.SMOKE_DETECTION,
aspect_type=AspectType.TYPE,
),
(50, 22): ElementMapping(
elementType=ElementType.STAIRLIFT,
aspect_type=AspectType.PRESENCE,
),
(50, 26): ElementMapping(
elementType=ElementType.DISABLED_FACILITIES,
aspect_type=AspectType.TYPE,
),
(100, 3): ElementMapping(
elementType=ElementType.PROPERTY,
aspect_type=AspectType.AGE_BAND,
),
(100, 14): ElementMapping(
elementType=ElementType.PROPERTY,
aspect_type=AspectType.CONSTRUCTION_TYPE,
),
(100, 16): ElementMapping(
elementType=ElementType.PROPERTY,
aspect_type=AspectType.CLASSIFICATION,
),
(210, 2): ElementMapping(
elementType=ElementType.PASSENGER_LIFT,
aspect_type=AspectType.TYPE,
),
# ==========================================================
# EXTERNAL – WALLS
# ==========================================================
(50, 16): ElementMapping(
elementType=ElementType.PARTY_WALL_FIRE_BREAK,
aspect_type=AspectType.PRESENCE,
),
(53, 1): ElementMapping(
elementType=ElementType.BOUNDARY_WALLS,
aspect_type=AspectType.PRESENCE,
),
(53, 4): ElementMapping(
elementType=ElementType.EXTERNAL_DECORATION,
aspect_type=AspectType.PRESENCE,
),
(53, 5): ElementMapping(
elementType=ElementType.EXTERNAL_NOISE_INSULATION,
aspect_type=AspectType.ADEQUACY,
),
(53, 14): ElementMapping(
elementType=ElementType.GARAGE_WALLS,
aspect_type=AspectType.MATERIAL,
),
(53, 23): ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.FINISH,
),
(53, 30): ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.FINISH,
aspect_instance=2,
),
(53, 36): ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.INSULATION,
),
(53, 40): ElementMapping(
elementType=ElementType.SPANDREL_PANELS,
aspect_type=AspectType.MATERIAL,
),
(53, 41): ElementMapping(
elementType=ElementType.CLADDING,
aspect_type=AspectType.MATERIAL,
),
(100, 15): ElementMapping(
elementType=ElementType.EXTERNAL_DECORATION,
aspect_type=AspectType.CONDITION,
),
(120, 1): ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.STRUCTURE,
),
(120, 2): ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.FINISH,
),
(120, 3): ElementMapping(
elementType=ElementType.EXTERNAL_WALL,
aspect_type=AspectType.INSULATION,
),
# ==========================================================
# EXTERNAL – ROOFS
# ==========================================================
(50, 15): ElementMapping(
elementType=ElementType.LOFT_INSULATION,
aspect_type=AspectType.TYPE,
),
(53, 2): ElementMapping(
elementType=ElementType.CHIMNEY,
aspect_type=AspectType.PRESENCE,
),
(53, 6): ElementMapping(
elementType=ElementType.FASCIA_SOFFIT_BARGEBOARDS,
aspect_type=AspectType.MATERIAL,
),
(53, 7): ElementMapping(
elementType=ElementType.FLAT_ROOF_COVERING,
aspect_type=AspectType.MATERIAL,
),
(53, 13): ElementMapping(
elementType=ElementType.GARAGE_ROOF,
aspect_type=AspectType.MATERIAL,
),
(53, 15): ElementMapping(
elementType=ElementType.GUTTERS,
aspect_type=AspectType.MATERIAL,
),
(53, 21): ElementMapping(
elementType=ElementType.PITCHED_ROOF_COVERING,
aspect_type=AspectType.MATERIAL,
),
(53, 22): ElementMapping(
elementType=ElementType.PORCH_CANOPY,
aspect_type=AspectType.TYPE,
),
(53, 47): ElementMapping(
elementType=ElementType.ROOF,
aspect_type=AspectType.STRUCTURE,
),
(110, 1): ElementMapping(
elementType=ElementType.ROOF,
aspect_type=AspectType.MATERIAL,
),
(110, 2): ElementMapping(
elementType=ElementType.ROOF,
aspect_type=AspectType.MATERIAL,
aspect_instance=1,
),
(110, 3): ElementMapping(
elementType=ElementType.CHIMNEY,
aspect_type=AspectType.WORK_REQUIRED,
),
(110, 4): ElementMapping(
elementType=ElementType.FASCIA,
aspect_type=AspectType.MATERIAL,
),
(110, 5): ElementMapping(
elementType=ElementType.SOFFIT,
aspect_type=AspectType.MATERIAL,
),
(110, 6): ElementMapping(
elementType=ElementType.RAINWATER_GOODS,
aspect_type=AspectType.MATERIAL,
),
(110, 7): ElementMapping(
elementType=ElementType.LOFT_INSULATION,
aspect_type=AspectType.WORK_REQUIRED, # possibly not the right aspect type
),
(110, 8): ElementMapping(
elementType=ElementType.PORCH_CANOPY,
aspect_type=AspectType.MATERIAL,
),
# ==========================================================
# EXTERNAL – DOORS & WINDOWS
# ==========================================================
(50, 8): ElementMapping(
elementType=ElementType.DOOR_ENTRY_HANDSET,
aspect_type=AspectType.PRESENCE,
),
(53, 8): ElementMapping(
elementType=ElementType.FRONT_DOOR,
aspect_type=AspectType.MATERIAL,
),
(53, 12): ElementMapping(
elementType=ElementType.GARAGE_DOOR,
aspect_type=AspectType.MATERIAL,
),
(53, 16): ElementMapping(
elementType=ElementType.LINTEL,
aspect_type=AspectType.PRESENCE,
),
(53, 19): ElementMapping(
elementType=ElementType.PATIO_FRENCH_DOOR,
aspect_type=AspectType.MATERIAL,
),
(53, 25): ElementMapping(
elementType=ElementType.REAR_DOOR,
aspect_type=AspectType.MATERIAL,
),
(53, 29): ElementMapping(
elementType=ElementType.SECONDARY_GLAZING,
aspect_type=AspectType.PRESENCE,
),
(53, 35): ElementMapping(
elementType=ElementType.STORE_DOOR,
aspect_type=AspectType.MATERIAL,
),
(53, 38): ElementMapping(
elementType=ElementType.EXTERNAL_WINDOWS,
aspect_type=AspectType.TYPE,
),
(53, 39): ElementMapping(
elementType=ElementType.EXTERNAL_WINDOWS,
aspect_type=AspectType.TYPE,
aspect_instance=2,
),
(53, 43): ElementMapping(
elementType=ElementType.FRONT_DOOR,
aspect_type=AspectType.TYPE,
),
(130, 1): ElementMapping(
elementType=ElementType.EXTERNAL_WINDOWS,
aspect_type=AspectType.MATERIAL,
),
(130, 2): ElementMapping(
elementType=ElementType.COMMUNAL_WINDOWS,
aspect_type=AspectType.MATERIAL,
),
(140, 1): ElementMapping(
elementType=ElementType.MAIN_DOOR,
aspect_type=AspectType.MATERIAL,
),
(140, 2): ElementMapping(
elementType=ElementType.STORE_DOOR,
aspect_type=AspectType.MATERIAL,
), # Duplicate of (53, 35)
(140, 3): ElementMapping(
elementType=ElementType.GARAGE_DOOR,
aspect_type=AspectType.MATERIAL,
), # Duplicate of (53, 12)
(140, 4): ElementMapping(
elementType=ElementType.BLOCK_ENTRANCE_DOOR,
aspect_type=AspectType.MATERIAL,
),
# ==========================================================
# EXTERNAL AREAS
# ==========================================================
(53, 3): ElementMapping(
elementType=ElementType.DOWNPIPES,
aspect_type=AspectType.MATERIAL,
),
(53, 9): ElementMapping(
elementType=ElementType.FRONT_FENCING,
aspect_type=AspectType.MATERIAL,
),
(53, 10): ElementMapping(
elementType=ElementType.FRONT_GATE,
aspect_type=AspectType.TYPE,
),
(53, 17): ElementMapping(
elementType=ElementType.PARKING_AREAS,
aspect_type=AspectType.MATERIAL,
),
(53, 18): ElementMapping(
elementType=ElementType.PATHS_AND_HARDSTANDINGS,
aspect_type=AspectType.MATERIAL,
),
(53, 24): ElementMapping(
elementType=ElementType.PRIVATE_BALCONY,
aspect_type=AspectType.PRESENCE,
),
(53, 26): ElementMapping(
elementType=ElementType.REAR_FENCING,
aspect_type=AspectType.MATERIAL,
),
(53, 27): ElementMapping(
elementType=ElementType.REAR_GATE,
aspect_type=AspectType.TYPE,
),
(53, 28): ElementMapping(
elementType=ElementType.RETAINING_WALLS,
aspect_type=AspectType.PRESENCE,
),
(53, 31): ElementMapping(
elementType=ElementType.SIDE_FENCING,
aspect_type=AspectType.MATERIAL,
),
(53, 32): ElementMapping(
elementType=ElementType.SOIL_AND_VENT,
aspect_type=AspectType.MATERIAL,
),
(53, 34): ElementMapping(
elementType=ElementType.SOLAR_THERMALS,
aspect_type=AspectType.PRESENCE,
),
(53, 44): ElementMapping(
elementType=ElementType.GARAGE_STRUCTURE,
aspect_type=AspectType.TYPE,
),
(53, 45): ElementMapping(
elementType=ElementType.BALCONY_BALUSTRADE,
aspect_type=AspectType.MATERIAL,
),
(150, 1): ElementMapping(
elementType=ElementType.BLOCK_ENTRANCE_DOOR,
aspect_type=AspectType.MATERIAL,
),
(150, 2): ElementMapping(
elementType=ElementType.PATHS_AND_HARDSTANDINGS,
aspect_type=AspectType.MATERIAL,
), # Duplicate of (53, 18) - correct?
(150, 3): ElementMapping(
elementType=ElementType.ROADS,
aspect_type=AspectType.MATERIAL,
),
(150, 4): ElementMapping(
elementType=ElementType.BOUNDARY_WALLS,
aspect_type=AspectType.MATERIAL,
),
(150, 5): ElementMapping(
elementType=ElementType.OUTBUILDINGS,
aspect_type=AspectType.TYPE,
),
(150, 6): ElementMapping(
elementType=ElementType.GARAGE_STRUCTURE,
aspect_type=AspectType.TYPE,
),
# ==========================================================
# INTERNAL – BATHROOMS & KITCHENS
# ==========================================================
(50, 1): ElementMapping(
elementType=ElementType.SECONDARY_TOILET,
aspect_type=AspectType.PRESENCE,
),
(50, 9): ElementMapping(
elementType=ElementType.BATHROOM_EXTRACTOR_FAN,
aspect_type=AspectType.PRESENCE,
),
(50, 9): ElementMapping(
elementType=ElementType.KITCHEN,
aspect_type=AspectType.TYPE,
),
(50, 10): ElementMapping(
elementType=ElementType.KITCHEN_EXTRACTOR_FAN,
aspect_type=AspectType.PRESENCE,
),
(50, 13): ElementMapping(
elementType=ElementType.KITCHEN_SPACE_LAYOUT,
aspect_type=AspectType.ADEQUACY,
),
(50, 14): ElementMapping(
elementType=ElementType.KITCHEN,
aspect_type=AspectType.TYPE,
),
(50, 17): ElementMapping(
elementType=ElementType.BATHROOM,
aspect_type=AspectType.LOCATION,
),
(50, 18): ElementMapping(
elementType=ElementType.BATHROOM,
aspect_type=AspectType.TYPE,
), # Actually "Primary bathroom type" - ok like this?
(50, 20): ElementMapping(
elementType=ElementType.BATHROOM,
aspect_type=AspectType.TYPE,
element_instance=2,
), # Actually "Secondary bathroom type" - ok like this?
(160, 1): ElementMapping(
elementType=ElementType.KITCHEN,
aspect_type=AspectType.CONDITION,
),
(160, 2): ElementMapping(
elementType=ElementType.KITCHEN_SPACE_LAYOUT,
aspect_type=AspectType.ADEQUACY,
),
(190, 1): ElementMapping(
elementType=ElementType.BATHROOM,
aspect_type=AspectType.CONDITION,
),
(190, 2): ElementMapping(
elementType=ElementType.SECONDARY_TOILET,
aspect_type=AspectType.TYPE,
),
# ==========================================================
# COMMUNAL
# ==========================================================
(51, 1): ElementMapping(
elementType=ElementType.COMMUNAL_AERIAL,
aspect_type=AspectType.PRESENCE,
),
(51, 2): ElementMapping(
elementType=ElementType.COMMUNAL_AOV,
aspect_type=AspectType.PRESENCE,
),
(51, 3): ElementMapping(
elementType=ElementType.COMMUNAL_BALCONY_WALKWAY,
aspect_type=AspectType.PRESENCE,
),
(51, 4): ElementMapping(
elementType=ElementType.COMMUNAL_BATHROOM,
aspect_type=AspectType.TYPE,
),
(51, 5): ElementMapping(
elementType=ElementType.COMMUNAL_BIN_STORE_DOORS,
aspect_type=AspectType.PRESENCE,
),
(51, 6): ElementMapping(
elementType=ElementType.COMMUNAL_BIN_STORE_ROOF,
aspect_type=AspectType.PRESENCE,
),
(51, 7): ElementMapping(
elementType=ElementType.COMMUNAL_BIN_STORE_WALLS,
aspect_type=AspectType.MATERIAL,
),
(51, 8): ElementMapping(
elementType=ElementType.COMMUNAL_BMS,
aspect_type=AspectType.PRESENCE,
),
(51, 9): ElementMapping(
elementType=ElementType.COMMUNAL_BOILER,
aspect_type=AspectType.TYPE,
),
(51, 10): ElementMapping(
elementType=ElementType.COMMUNAL_BOOSTER_PUMP,
aspect_type=AspectType.PRESENCE,
),
(51, 11): ElementMapping(
elementType=ElementType.COMMUNAL_CCTV,
aspect_type=AspectType.PRESENCE,
),
(51, 12): ElementMapping(
elementType=ElementType.COMMUNAL_CIRCULATION_SPACE,
aspect_type=AspectType.ADEQUACY,
),
(51, 13): ElementMapping(
elementType=ElementType.COMMUNAL_COLD_WATER_STORAGE,
aspect_type=AspectType.PRESENCE,
),
(51, 14): ElementMapping(
elementType=ElementType.COMMUNAL_DOOR_ENTRY,
aspect_type=AspectType.SYSTEM,
),
(51, 15): ElementMapping(
elementType=ElementType.COMMUNAL_DRY_RISER,
aspect_type=AspectType.PRESENCE,
),
(51, 16): ElementMapping(
elementType=ElementType.COMMUNAL_EMERGENCY_LIGHTING,
aspect_type=AspectType.PRESENCE,
),
(51, 17): ElementMapping(
elementType=ElementType.COMMUNAL_EXTERNAL_DOORS,
aspect_type=AspectType.MATERIAL,
),
(51, 19): ElementMapping(
elementType=ElementType.COMMUNAL_FIRE_ALARM,
aspect_type=AspectType.TYPE,
),
(51, 20): ElementMapping(
elementType=ElementType.COMMUNAL_INTERNAL_DECORATIONS,
aspect_type=AspectType.PRESENCE,
),
(51, 21): ElementMapping(
elementType=ElementType.COMMUNAL_INTERNAL_DOORS,
aspect_type=AspectType.MATERIAL,
),
(51, 22): ElementMapping(
elementType=ElementType.COMMUNAL_INTERNAL_FLOOR,
aspect_type=AspectType.FINISH,
),
(51, 23): ElementMapping(
elementType=ElementType.COMMUNAL_KITCHEN,
aspect_type=AspectType.TYPE,
),
(51, 24): ElementMapping(
elementType=ElementType.COMMUNAL_LATERAL_MAINS,
aspect_type=AspectType.PRESENCE,
),
(51, 25): ElementMapping(
elementType=ElementType.COMMUNAL_LIGHTING,
aspect_type=AspectType.PRESENCE,
),
(51, 26): ElementMapping(
elementType=ElementType.COMMUNAL_LIGHTING_CONDUCTOR,
aspect_type=AspectType.PRESENCE,
),
(51, 27): ElementMapping(
elementType=ElementType.COMMUNAL_PASSENGER_LIFT,
aspect_type=AspectType.TYPE,
),
(51, 28): ElementMapping(
elementType=ElementType.COMMUNAL_ENTRANCE,
aspect_type=AspectType.MATERIAL,
element_instance=1,
),
(51, 30): ElementMapping(
elementType=ElementType.COMMUNAL_ENTRANCE,
aspect_type=AspectType.FINISH,
element_instance=2,
),
(51, 31): ElementMapping(
elementType=ElementType.COMMUNAL_SPRINKLER,
aspect_type=AspectType.PRESENCE,
),
(51, 29): ElementMapping(
elementType=ElementType.COMMUNAL_REFUSE_CHUTE,
aspect_type=AspectType.PRESENCE,
),
(51, 32): ElementMapping(
elementType=ElementType.COMMUNAL_STAIRS,
aspect_type=AspectType.FINISH,
),
(51, 33): ElementMapping(
elementType=ElementType.COMMUNAL_STORE_DOORS,
aspect_type=AspectType.MATERIAL,
),
(51, 34): ElementMapping(
elementType=ElementType.COMMUNAL_STORE_ROOF,
aspect_type=AspectType.MATERIAL,
),
(51, 35): ElementMapping(
elementType=ElementType.COMMUNAL_STORE_WALLS,
aspect_type=AspectType.MATERIAL,
),
(51, 36): ElementMapping(
elementType=ElementType.COMMUNAL_WALKWAYS,
aspect_type=AspectType.FINISH,
),
(51, 37): ElementMapping(
elementType=ElementType.COMMUNAL_WARDEN_CALL_SYSTEM,
aspect_type=AspectType.PRESENCE,
),
(51, 38): ElementMapping(
elementType=ElementType.COMMUNAL_TOILETS,
aspect_type=AspectType.TYPE,
),
(51, 39): ElementMapping(
elementType=ElementType.COMMUNAL_WET_RISER,
aspect_type=AspectType.PRESENCE,
),
(51, 40): ElementMapping(
elementType=ElementType.COMMUNAL_PLUG_SOCKETS,
aspect_type=AspectType.PRESENCE,
),
(200, 1): ElementMapping(
elementType=ElementType.COMMUNAL_BOILER,
aspect_type=AspectType.TYPE,
), # Duplicate of (51, 9) - correct?
(200, 2): ElementMapping(
elementType=ElementType.COMMUNAL_HEATING,
aspect_type=AspectType.TYPE,
),
(200, 3): ElementMapping(
elementType=ElementType.COMMUNAL_ELECTRICS,
aspect_type=AspectType.TYPE,
),
(200, 4): ElementMapping(
elementType=ElementType.COMMUNAL_FIRE_ALARM,
aspect_type=AspectType.TYPE,
),
(200, 5): ElementMapping(
elementType=ElementType.COMMUNAL_LIFT,
aspect_type=AspectType.TYPE,
),
(200, 6): ElementMapping(
elementType=ElementType.COMMUNAL_FLOOR_COVERING,
aspect_type=AspectType.MATERIAL,
),
(200, 7): ElementMapping(
elementType=ElementType.COMMUNAL_KITCHEN,
aspect_type=AspectType.TYPE,
),
(200, 8): ElementMapping(
elementType=ElementType.COMMUNAL_BATHROOM,
aspect_type=AspectType.TYPE,
), # Duplicate of (51, 4) - correct?
(200, 9): ElementMapping(
elementType=ElementType.COMMUNAL_TOILETS,
aspect_type=AspectType.TYPE,
), # Duplicate of (51, 38) - correct?
(200, 10): ElementMapping(
elementType=ElementType.COMMUNAL_GATES,
aspect_type=AspectType.TYPE,
),
# ==========================================================
# INTERNAL – HEATING
# ==========================================================
(50, 4): ElementMapping(
elementType=ElementType.HEATING_BOILER,
aspect_type=AspectType.PRESENCE,
), # This is actually "Central heating boiler" - ok like this?
(50, 5): ElementMapping(
elementType=ElementType.CENTRAL_HEATING,
aspect_type=AspectType.EXTENT,
),
(50, 6): ElementMapping(
elementType=ElementType.COLD_WATER_STORAGE,
aspect_type=AspectType.PRESENCE,
),
(50, 12): ElementMapping(
elementType=ElementType.HEATING_DISTRIBUTION,
aspect_type=AspectType.TYPE,
),
(50, 19): ElementMapping(
elementType=ElementType.PROGRAMMABLE_HEATING,
aspect_type=AspectType.TYPE,
),
(50, 25): ElementMapping(
elementType=ElementType.HEATING_BOILER,
aspect_type=AspectType.TYPE,
),
(170, 1): ElementMapping(
elementType=ElementType.HEATING_BOILER,
aspect_type=AspectType.TYPE,
), # Duplicate of (50,25) - correct?
(170, 2): ElementMapping(
elementType=ElementType.HEATING_DISTRIBUTION,
aspect_type=AspectType.TYPE,
), # Duplicate of (50,12) - correct?
(170, 3): ElementMapping(
elementType=ElementType.SECONDARY_HEATING,
aspect_type=AspectType.TYPE,
),
(170, 4): ElementMapping(
elementType=ElementType.COLD_WATER_STORAGE,
aspect_type=AspectType.TYPE,
),
(170, 5): ElementMapping(
elementType=ElementType.HOT_WATER_SYSTEM,
aspect_type=AspectType.TYPE,
),
# ==========================================================
# ELECTRICS
# ==========================================================
(50, 24): ElementMapping(
elementType=ElementType.INTERNAL_WIRING,
aspect_type=AspectType.MATERIAL,
),
(180, 1): ElementMapping(
elementType=ElementType.ELECTRICAL_WIRING,
aspect_type=AspectType.WORK_REQUIRED,
), # Not certain about the AspectType - only example in the sample data is "Full Rewire"
(180, 2): ElementMapping(
elementType=ElementType.CONSUMER_UNIT,
aspect_type=AspectType.TYPE,
),
(180, 3): ElementMapping(
elementType=ElementType.SMOKE_DETECTION,
aspect_type=AspectType.TYPE,
), # Duplicate of (50, 21) - correct?
(180, 4): ElementMapping(
elementType=ElementType.CARBON_MONOXIDE_DETECTION,
aspect_type=AspectType.TYPE,
), # Duplicate of (50, 2) - correct?
# ==========================================================
# HHSRS
# ==========================================================
(54, 1): ElementMapping(
elementType=ElementType.HHSRS_DAMP_AND_MOULD,
aspect_type=AspectType.RISK,
),
(54, 4): ElementMapping(
elementType=ElementType.HHSRS_ASBESTOS_AND_MMF,
aspect_type=AspectType.RISK,
),
(54, 15): ElementMapping(
elementType=ElementType.HHSRS_DOMESTIC_HYGIENE_PESTS_REFUSE,
aspect_type=AspectType.RISK,
),
(54, 29): ElementMapping(
elementType=ElementType.HHSRS_STRUCTURAL_COLLAPSE,
aspect_type=AspectType.RISK,
),
}

107
backend/condition/domain/mapping/peabody/peabody_mapper.py Normal file
View file

@ -0,0 +1,107 @@
from typing import Any, Dict, Optional, Tuple
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,
)
from backend.condition.domain.mapping.mapper import Mapper
from backend.condition.domain.property_condition_survey import PropertyConditionSurvey
from backend.condition.parsing.records.peabody.peabody_asset_condition import (
PeabodyAssetCondition,
)
from backend.condition.parsing.records.peabody.peabody_property import PeabodyProperty
from utils.logger import setup_logger
logger = setup_logger()
class PeabodyMapper(Mapper):
def map_asset_conditions_for_property(
self, client_property_data: Any, survey_year: Optional[int] = None
) -> PropertyConditionSurvey:
assert isinstance(
client_property_data, PeabodyProperty
) # TODO: think of a better way to do this
elements_by_key: dict[tuple[ElementType, int], Element] = {}
for raw_asset in client_property_data.assets:
element_mapping = PeabodyMapper._safe_map_element(raw_asset)
aspect_condition = PeabodyMapper._build_aspect_condition(
raw_asset, element_mapping
)
element_key = (
element_mapping.elementType,
element_mapping.element_instance or 1,
)
PeabodyMapper._attach_aspect_condition_to_element(
elements_by_key,
element_key,
aspect_condition,
)
return PropertyConditionSurvey(
uprn=client_property_data.uprn,
elements=list(elements_by_key.values()),
date=date(2000, 1, 1), # Temp - not sure how to get this
source="Peabody", # TODO: Make this the system, not the client
)
@staticmethod
def _safe_map_element(raw_asset: PeabodyAssetCondition) -> Optional[ElementMapping]:
try:
return PeabodyMapper._map_element(
raw_asset.element_code,
raw_asset.sub_element_code,
)
except KeyError:
logger.warning(
f"Unrecognised Peabody Asset Element: "
f"{raw_asset.element} ({raw_asset.element_code}), "
f"Sub-Element: {raw_asset.sub_element} ({raw_asset.sub_element_code}). "
"Skipping record"
)
return None
@staticmethod
def _map_element(element_code: int, sub_element_code: int) -> ElementMapping:
return PEABODY_ELEMENT_MAP[(element_code, sub_element_code)]
@staticmethod
def _attach_aspect_condition_to_element(
elements_by_key: Dict[Tuple[ElementType, int], Element],
element_key: Tuple[ElementType, int],
aspect_condition: AspectCondition,
) -> None:
element = elements_by_key.get(element_key)
if element is None:
element = Element(
element_type=element_key[0],
element_instance=element_key[1],
aspect_conditions=[],
)
elements_by_key[element_key] = element
element.aspect_conditions.append(aspect_condition)
@staticmethod
def _build_aspect_condition(
raw_asset, element_mapping: ElementMapping
) -> AspectCondition:
return 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,
)

14
backend/condition/domain/property_condition_survey.py Normal file
View file

@ -0,0 +1,14 @@
from dataclasses import dataclass
from typing import List
from datetime import date
from backend.condition.domain.element import Element
@dataclass
class PropertyConditionSurvey:
uprn: int
elements: List[Element]
date: date
source: str # TODO: make enum

4
backend/condition/file_type.py
View file

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

23
backend/condition/local_runner.py
View file

@ -2,6 +2,7 @@ from pathlib import Path
from backend.condition.processor import process_file from backend.condition.processor import process_file
def main() -> None: def main() -> None:
try: try:
# Works in scripts / debugger / pytest # Works in scripts / debugger / pytest
@ -12,14 +13,22 @@ def main() -> None:
path: Path = ROOT_DIR / "condition" / "sample_data" path: Path = ROOT_DIR / "condition" / "sample_data"
lbwf_path: Path = path / "lbwf" / "LBWF - Example Asset Data September 2025.xlsx" # TODO: get this from s3 as part of devcontainer init # TODO: get these from s3, maybe as part of devcontainer init
lbwf_path: Path = path / "lbwf" / "LBWF - Example Asset Data September 2025.xlsx"
peabody_path: Path = (
path
/ "peabody"
/ "2026_01_06 - Peabody - Stock Condition Data - Survey Records - D Lower.xlsx"
)
filepaths = [lbwf_path, peabody_path]
for fp in filepaths:
with fp.open("rb") as f:
process_file(
file_stream=f,
source_key=fp.as_posix(),
)
with lbwf_path.open("rb") as f:
process_file(
file_stream=f,
source_key=lbwf_path.as_posix(),
)
if __name__ == "__main__": if __name__ == "__main__":
main() main()

18
backend/condition/parsing/factory.py
View file

@ -1,9 +1,27 @@
from backend.condition.domain.mapping.lbwf.lbwf_mapper import LbwfMapper
from backend.condition.domain.mapping.mapper import Mapper
from backend.condition.domain.mapping.peabody.peabody_mapper import PeabodyMapper
from backend.condition.file_type import FileType from backend.condition.file_type import FileType
from backend.condition.parsing.parser import Parser from backend.condition.parsing.parser import Parser
from backend.condition.parsing.lbwf_parser import LbwfParser from backend.condition.parsing.lbwf_parser import LbwfParser
from backend.condition.parsing.peabody_parser import PeabodyParser
def select_parser(file_type: FileType) -> Parser: def select_parser(file_type: FileType) -> Parser:
if file_type is FileType.LBWF: if file_type is FileType.LBWF:
return LbwfParser() return LbwfParser()
if file_type is FileType.Peabody:
return PeabodyParser()
raise ValueError("Unrecognised file type, unable to instantiate Parser") raise ValueError("Unrecognised file type, unable to instantiate Parser")
def select_mapper(file_type: FileType) -> Mapper:
if file_type is FileType.LBWF:
return LbwfMapper()
if file_type is FileType.Peabody:
return PeabodyMapper()
raise ValueError("Unrecognised file type, unable to instantiate Mapper")

37
backend/condition/parsing/lbwf_parser.py
View file

@ -3,18 +3,23 @@ from openpyxl import Workbook, load_workbook
from collections import defaultdict from collections import defaultdict
from backend.condition.parsing.parser import Parser from backend.condition.parsing.parser import Parser
from backend.condition.parsing.records.lbwf.lbwf_asset_condition import LbwfAssetCondition from backend.condition.parsing.records.lbwf.lbwf_asset_condition import (
LbwfAssetCondition,
)
from backend.condition.parsing.records.lbwf.lbwf_house import LbwfHouse from backend.condition.parsing.records.lbwf.lbwf_house import LbwfHouse
from backend.condition.utils.date_utils import normalise_date from backend.condition.utils.date_utils import normalise_date
from utils.logger import setup_logger from utils.logger import setup_logger
logger = setup_logger logger = setup_logger()
class LbwfParser(Parser): class LbwfParser(Parser):
def parse(self, file_stream: BinaryIO) -> Any: def parse(self, file_stream: BinaryIO) -> Any:
wb: Workbook = load_workbook(file_stream) wb: Workbook = load_workbook(file_stream)
address_to_uprn_map: Dict[str, int] = self._generate_address_to_uprn_dict(wb) address_to_uprn_map: Dict[str, int] = LbwfParser._generate_address_to_uprn_dict(
wb
)
assets = self._parse_assets(wb) assets = self._parse_assets(wb)
houses = self._parse_houses(wb, address_to_uprn_map) houses = self._parse_houses(wb, address_to_uprn_map)
@ -82,7 +87,6 @@ class LbwfParser(Parser):
for house in houses: for house in houses:
house.assets = assets_by_ref.get(house.reference, []) house.assets = assets_by_ref.get(house.reference, [])
@staticmethod @staticmethod
def _map_row_to_house_record( def _map_row_to_house_record(
row: Any | Tuple[object | None, ...], row: Any | Tuple[object | None, ...],
@ -100,8 +104,8 @@ class LbwfParser(Parser):
house=row[header_indexes["HOSUE"]], house=row[header_indexes["HOSUE"]],
fail_decency=row[header_indexes["Fail Decency"]], fail_decency=row[header_indexes["Fail Decency"]],
assets=[], assets=[],
) )
@staticmethod @staticmethod
def _map_row_to_asset_record( def _map_row_to_asset_record(
row: Any | Tuple[object | None, ...], row: Any | Tuple[object | None, ...],
@ -119,7 +123,9 @@ class LbwfParser(Parser):
element_code=row[header_indexes["ELEMENT CODE"]], element_code=row[header_indexes["ELEMENT CODE"]],
element_code_description=row[header_indexes["ELEMENT CODE DESCRIPTION"]], element_code_description=row[header_indexes["ELEMENT CODE DESCRIPTION"]],
attribute_code=row[header_indexes["ATTRIBUTE CODE"]], attribute_code=row[header_indexes["ATTRIBUTE CODE"]],
attribute_code_description=row[header_indexes["ATTRIBUTE CODE DESCRIPTION"]], attribute_code_description=row[
header_indexes["ATTRIBUTE CODE DESCRIPTION"]
],
element_date_value=row[header_indexes["ELEMENT DATE VALUE"]], element_date_value=row[header_indexes["ELEMENT DATE VALUE"]],
element_numerical_value=row[header_indexes["ELEMENT NUMERIC VALUE"]], element_numerical_value=row[header_indexes["ELEMENT NUMERIC VALUE"]],
element_text_value=row[header_indexes["ELEMENT TEXT VALUE"]], element_text_value=row[header_indexes["ELEMENT TEXT VALUE"]],
@ -128,11 +134,10 @@ class LbwfParser(Parser):
remaining_life=row[header_indexes["REMAINING LIFE"]], remaining_life=row[header_indexes["REMAINING LIFE"]],
element_comments=row[header_indexes["ELEMENT COMMENTS"]], element_comments=row[header_indexes["ELEMENT COMMENTS"]],
) )
@staticmethod @staticmethod
def _generate_address_to_uprn_dict(wb: Workbook) -> Dict[str, int | None]: def _generate_address_to_uprn_dict(wb: Workbook) -> Dict[str, int | None]:
sheet: Workbook = wb["All Energy Breakdown "] sheet = wb["All Energy Breakdown "]
rows: Iterator[Tuple[object | None, ...]] = sheet.iter_rows(values_only=True) rows: Iterator[Tuple[object | None, ...]] = sheet.iter_rows(values_only=True)
@ -158,9 +163,9 @@ class LbwfParser(Parser):
return mapping return mapping
@staticmethod
def _get_column_indexes_by_name( def _get_column_indexes_by_name(
headers: Tuple[object | None, ...] headers: Tuple[object | None, ...],
) -> Dict[str, int]: ) -> Dict[str, int]:
index: Dict[str, int] = {} index: Dict[str, int] = {}
@ -169,12 +174,14 @@ class LbwfParser(Parser):
index[header] = i index[header] = i
return index return index
def _get_uprn_from_address(address: str, address_to_uprn_map: Dict[str, int]) -> int | None: @staticmethod
def _get_uprn_from_address(
address: str, address_to_uprn_map: Dict[str, int]
) -> int | None:
pseudo_name = address.split(",")[0] pseudo_name = address.split(",")[0]
if pseudo_name.lower() in (k.lower() for k in address_to_uprn_map.keys()): if pseudo_name.lower() in (k.lower() for k in address_to_uprn_map.keys()):
return address_to_uprn_map[pseudo_name.upper()] return address_to_uprn_map[pseudo_name.upper()]
return None return None

145
backend/condition/parsing/peabody_parser.py Normal file
View file

@ -0,0 +1,145 @@
from typing import Any, BinaryIO, Dict, Iterator, List, Tuple, DefaultDict
from openpyxl import Workbook, load_workbook
from collections import defaultdict
from backend.condition.parsing.parser import Parser
from backend.condition.parsing.records.peabody.peabody_asset_condition import PeabodyAssetCondition
from backend.condition.parsing.records.peabody.peabody_property import PeabodyProperty
from utils.logger import setup_logger
logger = setup_logger()
class PeabodyParser(Parser):
def parse(self, file_stream: BinaryIO) -> Any:
wb: Workbook = load_workbook(file_stream)
address_to_uprn_map: Dict[str, int] = PeabodyParser._generate_address_to_uprn_dict(wb)
assets = self._parse_assets(wb)
return self._group_assets_into_properties(
assets=assets,
address_to_uprn_map=address_to_uprn_map,
)
@staticmethod
def _parse_assets(wb: Workbook) -> List[PeabodyAssetCondition]:
assets_sheet = wb["Survey Records - D & Lower"]
asset_rows = assets_sheet.iter_rows(values_only=True)
asset_headers = next(asset_rows)
asset_header_indexes = PeabodyParser._get_column_indexes_by_name(asset_headers)
assets: List[PeabodyAssetCondition] = []
for row in asset_rows:
try:
asset = PeabodyParser._map_row_to_asset_record(row, asset_header_indexes)
if not asset.is_block_level:
# Block-level condition surveys are out of scope for now
# until we have a wider think on how to handle block
assets.append(asset) # TODO: handle block-level assets
except Exception as e:
logger.error(f"Error mapping Peabody row to asset record: {e}")
continue
return assets
@staticmethod
def _group_assets_into_properties(
assets: List[PeabodyAssetCondition],
address_to_uprn_map: Dict[str, int],
) -> List[PeabodyProperty]:
assets_by_address: DefaultDict[str, List[PeabodyAssetCondition]] = defaultdict(list)
for asset in assets:
if asset.full_address is None:
continue
address = asset.full_address.strip()
assets_by_address[address].append(asset)
properties: List[PeabodyProperty] = []
for address, grouped_assets in assets_by_address.items():
uprn = address_to_uprn_map.get(address)
if uprn is None:
logger.warning(f"No UPRN found for address: {address}")
continue
properties.append(
PeabodyProperty(
uprn=uprn,
assets=grouped_assets,
)
)
return properties
@staticmethod
def _map_row_to_asset_record(
row: Any | Tuple[object | None, ...],
header_indexes: Dict[str, int],
) -> PeabodyAssetCondition:
return PeabodyAssetCondition(
lo_reference=row[header_indexes["Lo_Reference"]],
full_address=row[header_indexes["full_address"]],
location_type_code=row[header_indexes["location_type_code"]],
parent_lo_reference=row[header_indexes["Parent_Lo_Reference"]],
element_code=row[header_indexes["Element_Code"]],
element=row[header_indexes["Element"]],
sub_element_code=row[header_indexes["Sub_Element_Code"]],
sub_element=row[header_indexes["Sub_Element"]],
material_code=row[header_indexes["Material_Code"]],
material_or_answer=row[header_indexes["material_or_answer"]],
renewal_quantity=row[header_indexes["Renewal_Quantity"]],
renewal_year=row[header_indexes["Renewal_Year"]],
renewal_cost=row[header_indexes["Renewal_Cost"]],
cloned=row[header_indexes["cloned"]],
lo_type_code=row[header_indexes["lo_type_code"]],
condition_survey_date=row[header_indexes["condition_survey_date"]],
)
@staticmethod
def _generate_address_to_uprn_dict(wb: Workbook) -> Dict[str, int | None]:
sheet = wb["Survey Records - D & Lower"]
rows: Iterator[Tuple[object | None, ...]] = sheet.iter_rows(values_only=True)
headers = next(rows)
header_indexes: Dict[str, int] = PeabodyParser._get_column_indexes_by_name(headers)
address_idx = header_indexes["full_address"]
address_to_uprn: Dict[str, int] = {}
# Generate random UPRNs for now
next_uprn = 1 # TODO: get real UPRNs
for row in rows:
address = row[address_idx]
if address is None:
continue
address = address.strip()
if address not in address_to_uprn:
address_to_uprn[address] = next_uprn
next_uprn += 1
return address_to_uprn
@staticmethod
def _get_column_indexes_by_name(
headers: Tuple[object | None, ...]
) -> Dict[str, int]:
index: Dict[str, int] = {}
for i, header in enumerate(headers):
if isinstance(header, str):
index[header] = i
return index

15
backend/condition/parsing/records/lbwf/lbwf_asset_condition.py
View file

@ -1,5 +1,6 @@
from dataclasses import dataclass from dataclasses import dataclass
from datetime import date from datetime import date
from typing import Optional
@dataclass @dataclass
@ -16,11 +17,11 @@ class LbwfAssetCondition:
element_code_description: str element_code_description: str
attribute_code: str attribute_code: str
attribute_code_description: str attribute_code_description: str
element_date_value: str | None = None element_date_value: Optional[str] = None
element_numerical_value: int | None = None element_numerical_value: Optional[int] = None
element_text_value: str | None = None element_text_value: Optional[str] = None
quantity: int | None = None quantity: Optional[int] = None
install_date: date | None = None install_date: Optional[date] = None
remaining_life: int | None = None remaining_life: Optional[int] = None
element_comments: str | None = None element_comments: Optional[str] = None

4
backend/condition/parsing/records/lbwf/lbwf_house.py
View file

@ -8,8 +8,8 @@ class LbwfHouse:
uprn: int uprn: int
reference: int reference: int
address: str address: str
epc: str # TODO: make enum epc: str # TODO: make enum?
shdf: bool shdf: str
house: str house: str
fail_decency: int fail_decency: int
assets: List[LbwfAssetCondition] assets: List[LbwfAssetCondition]

44
backend/condition/parsing/records/peabody/peabody_asset_condition.py Normal file
View file

@ -0,0 +1,44 @@
import re
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_year: int
cloned: str
lo_type_code: int
renewal_cost: Optional[float] = None
condition_survey_date: Optional[datetime] = None
@property
def is_block_level(self) -> bool:
# TODO: maybe use block codes from other Peabody dataset to do this instead
if not self.full_address:
return False
address = self.full_address.upper()
block_level_patterns = [
r"\bBLOCK\b", # BLOCK MILNE HOUSE
r"\bFLATS\b", # FLATS A-D
r"\b\d+[A-Z]?-\d+[A-Z]?\b", # 1-80, 9A-9H
r"\b\d+[A-Z]-[A-Z]\b", # 81A-B
r"\b\d+\s*&\s*\d+\b", # 73 & 74
]
return any(re.search(pattern, address) for pattern in block_level_patterns)

11
backend/condition/parsing/records/peabody/peabody_property.py Normal file
View file

@ -0,0 +1,11 @@
from dataclasses import dataclass
from typing import List
from backend.condition.parsing.records.peabody.peabody_asset_condition import PeabodyAssetCondition
@dataclass
class PeabodyProperty:
# This could just be a uprn:assets dict, but making it a dataclass for consistency with
# other client models, might change in future
uprn: int
assets: List[PeabodyAssetCondition]

20
backend/condition/processor.py
View file

@ -1,9 +1,13 @@
from typing import Any, BinaryIO, List from typing import Any, BinaryIO, List
from datetime import datetime
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 backend.condition.parsing.parser import Parser
from utils.logger import setup_logger from utils.logger import setup_logger
from backend.condition.file_type import FileType, detect_file_type from backend.condition.file_type import FileType, detect_file_type
from backend.condition.parsing.factory import select_parser from backend.condition.parsing.factory import select_parser, select_mapper
def process_file(file_stream: BinaryIO, source_key: str) -> None: def process_file(file_stream: BinaryIO, source_key: str) -> None:
print(f"[processor] Received file: {source_key}") print(f"[processor] Received file: {source_key}")
@ -11,8 +15,18 @@ def process_file(file_stream: BinaryIO, source_key: str) -> None:
# Instantiation # Instantiation
file_type: FileType = detect_file_type(source_key) file_type: FileType = detect_file_type(source_key)
parser: Parser = select_parser(file_type) parser: Parser = select_parser(file_type)
mapper: Mapper = select_mapper(file_type)
# Orchestration # Orchestration
records: List[Any] = parser.parse(file_stream) raw_properties: List[Any] = parser.parse(file_stream)
print(records) # temp survey_year = datetime.now().year # TODO: get this from filepath or elsewhere
property_condition_surveys: List[PropertyConditionSurvey] = []
for p in raw_properties:
property_condition_surveys.append(
mapper.map_asset_conditions_for_property(p, survey_year)
)
print("done") # temp

74
backend/condition/tests/custom_asserts.py Normal file
View file

@ -0,0 +1,74 @@
from backend.condition.domain.property_condition_survey import PropertyConditionSurvey
class CustomAsserts:
def assert_property_condition_surveys_equal(
actual: PropertyConditionSurvey,
expected: PropertyConditionSurvey,
) -> bool:
assert actual.uprn == expected.uprn, "UPRN differs"
assert actual.source == expected.source, "Source differs"
assert actual.date == expected.date, "Date differs"
assert len(actual.elements) == len(expected.elements), (
f"Expected {len(expected.elements)} elements, "
f"got {len(actual.elements)}"
)
for i, (actual_element, expected_element) in enumerate(
zip(actual.elements, expected.elements)
):
assert actual_element.element_type == expected_element.element_type, (
f"Element[{i}] type differs: "
f"{actual_element.element_type} != {expected_element.element_type}"
)
assert (
actual_element.element_instance == expected_element.element_instance
), (
f"Element[{i}] instance differs: "
f"{actual_element.element_instance} != {expected_element.element_instance}"
)
assert len(actual_element.aspect_conditions) == len(
expected_element.aspect_conditions
), f"Element[{i}] aspect count differs"
for j, (actual_aspect, expected_aspect) in enumerate(
zip(
actual_element.aspect_conditions,
expected_element.aspect_conditions,
)
):
prefix = f"Element[{i}].Aspect[{j}]"
assert actual_aspect.aspect_type == expected_aspect.aspect_type, (
f"{prefix}.aspect_type differs: "
f"{actual_aspect.aspect_type} != {expected_aspect.aspect_type}"
)
assert (
actual_aspect.aspect_instance == expected_aspect.aspect_instance
), (
f"{prefix}.aspect_instance differs: "
f"{actual_aspect.aspect_instance} != {expected_aspect.aspect_instance}"
)
assert actual_aspect.value == expected_aspect.value, (
f"{prefix}.value differs: "
f"{actual_aspect.value} != {expected_aspect.value}"
)
assert actual_aspect.quantity == expected_aspect.quantity, (
f"{prefix}.quantity differs: "
f"{actual_aspect.quantity} != {expected_aspect.quantity}"
)
assert actual_aspect.install_date == expected_aspect.install_date, (
f"{prefix}.install_date differs: "
f"{actual_aspect.install_date} != {expected_aspect.install_date}"
)
assert actual_aspect.renewal_year == expected_aspect.renewal_year, (
f"{prefix}.renewal_year differs: "
f"{actual_aspect.renewal_year} != {expected_aspect.renewal_year}"
)
assert actual_aspect.comments == expected_aspect.comments, (
f"{prefix}.comments differs: "
f"{actual_aspect.comments} != {expected_aspect.comments}"
)
return True

366
backend/condition/tests/mapping/test_lbwf_mapper.py Normal file
View file

@ -0,0 +1,366 @@
from datetime import 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.lbwf.lbwf_mapper import LbwfMapper
from backend.condition.domain.property_condition_survey import PropertyConditionSurvey
from backend.condition.parsing.records.lbwf.lbwf_house import LbwfHouse
from backend.condition.parsing.records.lbwf.lbwf_asset_condition import (
LbwfAssetCondition,
)
from backend.condition.domain.element import Element
from backend.condition.tests.custom_asserts import CustomAsserts
def test_lbwf_mapper_maps_house():
# arrange
lbwf_house = LbwfHouse(
uprn=1,
reference=100,
address="123 Fake Street, London, A10 1AB",
epc="F",
shdf="NO",
house="HOUSE",
fail_decency=2025,
assets=[
LbwfAssetCondition(
prop_ref=100,
domna=100,
address="123 Fake Street, London, A10 1AB",
ownership="LBWF_OWNED",
prop_status="OCCP",
prop_type="HOU",
prop_sub_type="TERRACED",
element_group="ASSETS",
element_code="AHR_CAT",
element_code_description="Accessible Housing Register Category",
attribute_code="F",
attribute_code_description="General Needs",
element_date_value=None,
element_numerical_value=None,
element_text_value=None,
quantity=1,
install_date=None,
remaining_life=None,
element_comments=None,
),
LbwfAssetCondition(
prop_ref=100,
domna=100,
address="123 Fake Street, London, A10 1AB",
ownership="LBWF_OWNED",
prop_status="OCCP",
prop_type="HOU",
prop_sub_type="TERRACED",
element_group="ASSETS",
element_code="FLVL",
element_code_description="Floor Level of Front Door",
attribute_code="0G",
attribute_code_description="Ground Floor",
element_date_value=None,
element_numerical_value=None,
element_text_value=None,
quantity=1,
install_date=None,
remaining_life=None,
element_comments=None,
),
LbwfAssetCondition(
prop_ref=100,
domna=100,
address="123 Fake Street, London, A10 1AB",
ownership="LBWF_OWNED",
prop_status="OCCP",
prop_type="HOU",
prop_sub_type="TERRACED",
element_group="ASSETS",
element_code="ASBESTOS",
element_code_description="Asbestos Present",
attribute_code="YES",
attribute_code_description="Yes",
element_date_value=None,
element_numerical_value=None,
element_text_value=None,
quantity=None,
install_date=None,
remaining_life=None,
element_comments="Source of Data = ACT",
),
LbwfAssetCondition(
prop_ref=100,
domna=100,
address="123 Fake Street, London, A10 1AB",
ownership="LBWF_OWNED",
prop_status="OCCP",
prop_type="HOU",
prop_sub_type="TERRACED",
element_group="ASSETS",
element_code="HHSRSASB",
element_code_description="Asbestos (and MMF)",
attribute_code="TYPRISK",
attribute_code_description="Category 4 - Typical Risk",
element_date_value=None,
element_numerical_value=None,
element_text_value=None,
quantity=None,
install_date=None,
remaining_life=None,
element_comments="Source of Data = ACT",
),
LbwfAssetCondition(
prop_ref=100,
domna=100,
address="123 Fake Street, London, A10 1AB",
ownership="LBWF_OWNED",
prop_status="OCCP",
prop_type="HOU",
prop_sub_type="TERRACED",
element_group="ASSETS",
element_code="INTBTHRLOC",
element_code_description="Location of Bathroom in Property",
attribute_code="ENTRANCE",
attribute_code_description="Bathroom on Entrance Level in Property",
element_date_value=None,
element_numerical_value=None,
element_text_value=None,
quantity=1,
install_date=None,
remaining_life=None,
element_comments="Source of Data = Codeman",
),
LbwfAssetCondition(
prop_ref=100,
domna=100,
address="123 Fake Street, London, A10 1AB",
ownership="LBWF_OWNED",
prop_status="OCCP",
prop_type="HOU",
prop_sub_type="TERRACED",
element_group="ASSETS",
element_code="INTCHEXTNT",
element_code_description="Extent of Central Heating in Property",
attribute_code="NONE",
attribute_code_description="No Central Heating in Property",
element_date_value=None,
element_numerical_value=None,
element_text_value=None,
quantity=1,
install_date=None,
remaining_life=None,
element_comments="Source of Data = Codeman",
),
LbwfAssetCondition(
prop_ref=100,
domna=100,
address="123 Fake Street, London, A10 1AB",
ownership="LBWF_OWNED",
prop_status="OCCP",
prop_type="HOU",
prop_sub_type="TERRACED",
element_group="ASSETS",
element_code="HHSRSFIRE",
element_code_description="Fire",
attribute_code="TYPRISK",
attribute_code_description="Category 4 - Typical Risk",
element_date_value=None,
element_numerical_value=None,
element_text_value=None,
quantity=1,
install_date=None,
remaining_life=None,
element_comments="Source of Data = Morgan Sindall",
),
LbwfAssetCondition(
prop_ref=100,
domna=100,
address="123 Fake Street, London, A10 1AB",
ownership="LBWF_OWNED",
prop_status="OCCP",
prop_type="HOU",
prop_sub_type="TERRACED",
element_group="ASSETS",
element_code="EXTWALLFN1",
element_code_description="Wall Finish 1 in External Area",
attribute_code="SMTHRENDER",
attribute_code_description="Render or Pebbledash in External Area",
element_date_value=None,
element_numerical_value=None,
element_text_value=None,
quantity=1,
install_date=date(2009, 4, 1),
remaining_life=26,
element_comments="Source of Data = Codeman",
),
LbwfAssetCondition(
prop_ref=100,
domna=100,
address="123 Fake Street, London, A10 1AB",
ownership="LBWF_OWNED",
prop_status="OCCP",
prop_type="HOU",
prop_sub_type="TERRACED",
element_group="ASSETS",
element_code="EXTWALLFN2",
element_code_description="Wall Finish 2 in External Area",
attribute_code="SMTHRENDER",
attribute_code_description="Smooth Render Wall Finish 2 in External Area",
element_date_value=None,
element_numerical_value=None,
element_text_value=None,
quantity=1,
install_date=date(2009, 4, 1),
remaining_life=26,
element_comments="Source of Data = Codeman",
),
],
)
mapper = LbwfMapper()
survey_year = 2026
expected_condition_survey = PropertyConditionSurvey(
uprn=1,
elements=[
Element(
element_type=ElementType.ACCESSIBLE_HOUSING_REGISTER,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.CATEGORY,
aspect_instance=1,
value="General Needs",
quantity=1,
install_date=None,
renewal_year=None,
comments=None,
)
],
),
Element(
element_type=ElementType.FLOOR_LEVEL_FRONT_DOOR,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.LOCATION,
aspect_instance=1,
value="Ground Floor",
quantity=1,
install_date=None,
renewal_year=None,
comments=None,
)
],
),
Element(
element_type=ElementType.ASBESTOS,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.PRESENCE,
aspect_instance=1,
value="Yes",
quantity=None,
install_date=None,
renewal_year=None,
comments="Source of Data = ACT",
)
],
),
Element(
element_type=ElementType.HHSRS_ASBESTOS_AND_MMF,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.RISK,
aspect_instance=1,
value="Category 4 - Typical Risk",
quantity=None,
renewal_year=None,
comments="Source of Data = ACT",
)
],
),
Element(
element_type=ElementType.BATHROOM,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.LOCATION,
aspect_instance=1,
value="Bathroom on Entrance Level in Property",
quantity=1,
install_date=None,
renewal_year=None,
comments="Source of Data = Codeman",
)
],
),
Element(
element_type=ElementType.CENTRAL_HEATING,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.EXTENT,
aspect_instance=1,
value="No Central Heating in Property",
quantity=1,
install_date=None,
renewal_year=None,
comments="Source of Data = Codeman",
)
],
),
Element(
element_type=ElementType.HHSRS_FIRE,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.RISK,
aspect_instance=1,
value="Category 4 - Typical Risk",
quantity=1,
install_date=None,
renewal_year=None,
comments="Source of Data = Morgan Sindall",
)
],
),
Element(
element_type=ElementType.EXTERNAL_WALL,
element_instance=1,
aspect_conditions=[
AspectCondition(
aspect_type=AspectType.FINISH,
aspect_instance=1,
value="Render or Pebbledash in External Area",
quantity=1,
install_date=date(2009, 4, 1),
renewal_year=2052,
comments="Source of Data = Codeman",
),
AspectCondition(
aspect_type=AspectType.FINISH,
aspect_instance=2,
value="Smooth Render Wall Finish 2 in External Area",
quantity=1,
install_date=date(2009, 4, 1),
renewal_year=2052,
comments="Source of Data = Codeman",
),
],
),
],
date=date(2000, 1, 1), # what should this be?
source="LBWF",
)
# act
actual_condition_survey: PropertyConditionSurvey = (
mapper.map_asset_conditions_for_property(lbwf_house, survey_year)
)
# assert
assert CustomAsserts.assert_property_condition_surveys_equal(
actual_condition_survey, expected_condition_survey
)

220
backend/condition/tests/mapping/test_peabody_mapper.py Normal file
View file

@ -0,0 +1,220 @@
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,
)
from backend.condition.parsing.records.peabody.peabody_property import PeabodyProperty
from backend.condition.domain.element import Element
from backend.condition.tests.custom_asserts import CustomAsserts
def test_peabody_mapper_maps_property():
# arrange
peabody_property = PeabodyProperty(
uprn=1,
assets=[
PeabodyAssetCondition(
lo_reference="1000RAND0000",
full_address="FLAT 1 RANDOM SQUARE FAKE STREET LONDON E1 1EE",
location_type_code=1,
parent_lo_reference="RAND1000",
element_code=130,
element="WINDOWS",
sub_element_code=1,
sub_element="Windows",
material_code=1,
material_or_answer="UPVC Double Glazed",
renewal_quantity=8,
renewal_year=2036,
renewal_cost=4800,
cloned="N",
lo_type_code=1,
condition_survey_date=datetime(2024, 2, 15, 12, 47, 0),
),
PeabodyAssetCondition(
lo_reference="1000RAND0000",
full_address="FLAT 1 RANDOM SQUARE FAKE STREET LONDON E1 1EE",
location_type_code=1,
parent_lo_reference="RAND1000",
element_code=100,
element="GENERAL",
sub_element_code=15,
sub_element="External Decoration",
material_code=2,
material_or_answer="Normal",
renewal_quantity=1,
renewal_year=2029,
renewal_cost=1500,
cloned="N",
lo_type_code=1,
condition_survey_date=datetime(2024, 2, 15, 12, 47, 0),
),
],
)
mapper = PeabodyMapper()
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_condition_survey: PropertyConditionSurvey = (
mapper.map_asset_conditions_for_property(peabody_property)
)
# assert
assert actual_condition_survey == expected_condition_survey
def test_wall_primary_and_secondary_wall_finish_map_correctly():
# arrange
peabody_property = PeabodyProperty(
uprn=1,
assets=[
PeabodyAssetCondition(
lo_reference="1000RAND0000",
full_address="FLAT 1 RANDOM SQUARE FAKE STREET LONDON E1 1EE",
location_type_code=1,
parent_lo_reference="RAND1000",
element_code=53,
element="External",
sub_element_code=23,
sub_element="Primary Wall Finish",
material_code=4,
material_or_answer="Pointed",
renewal_quantity=65,
renewal_year=2045,
renewal_cost=3835,
cloned="N",
lo_type_code=1,
condition_survey_date=datetime(2024, 2, 15, 12, 47, 0),
),
PeabodyAssetCondition(
lo_reference="1000RAND0000",
full_address="FLAT 1 RANDOM SQUARE FAKE STREET LONDON E1 1EE",
location_type_code=1,
parent_lo_reference="RAND1000",
element_code=120,
element="WALLS",
sub_element_code=2,
sub_element="Wall Finish",
material_code=1,
material_or_answer="Pointing",
renewal_quantity=1,
renewal_year=2069,
renewal_cost=2450,
cloned="N",
lo_type_code=1,
condition_survey_date=datetime(2014, 2, 15, 12, 47, 0),
),
PeabodyAssetCondition(
lo_reference="1000RAND0000",
full_address="FLAT 1 RANDOM SQUARE FAKE STREET LONDON E1 1EE",
location_type_code=1,
parent_lo_reference="RAND1000",
element_code=53,
element="External",
sub_element_code=30,
sub_element="Secondary Wall Finish",
material_code=8,
material_or_answer="Tile Hung",
renewal_quantity=8,
renewal_year=2049,
renewal_cost=472,
cloned="N",
lo_type_code=1,
condition_survey_date=datetime(2014, 2, 15, 12, 47, 0),
),
],
)
mapper = PeabodyMapper()
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_condition_survey: PropertyConditionSurvey = (
mapper.map_asset_conditions_for_property(peabody_property)
)
# assert
assert CustomAsserts.assert_property_condition_surveys_equal(
actual_condition_survey, expected_condition_survey
)

2
backend/condition/tests/parsing/test_lbwf_parser.py
View file

@ -112,7 +112,7 @@ def lbwf_homes_xlsx_bytes() -> BytesIO:
return stream return stream
def test_lbwf_parser_passes_houses(lbwf_homes_xlsx_bytes): def test_lbwf_parser_parses_houses(lbwf_homes_xlsx_bytes):
# arrange # arrange
parser = LbwfParser() parser = LbwfParser()

11
backend/condition/tests/parsing/test_parsing_factory.py
View file

@ -11,5 +11,16 @@ def test_selects_lbwf_parser():
# act # act
actual_class_name = select_parser(file_type).__class__.__name__ 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
assert expected_class_name == actual_class_name assert expected_class_name == actual_class_name

190
backend/condition/tests/parsing/test_peabody_parser.py Normal file
View file

@ -0,0 +1,190 @@
import pytest
from typing import Any
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 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([
"B000BLOCK",
"1100 BLOCK",
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 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)
@pytest.fixture
def asset_condition_factory():
def _factory(full_address: str) -> PeabodyAssetCondition:
return PeabodyAssetCondition(
lo_reference="",
full_address=full_address,
location_type_code=0,
parent_lo_reference="",
element_code=0,
element="",
sub_element_code=0,
sub_element="",
material_code=0,
material_or_answer="",
renewal_quantity=0,
renewal_year=2026,
cloned="",
lo_type_code=0,
renewal_cost=None,
condition_survey_date=None,
)
return _factory
@pytest.mark.parametrize(
"full_address, expected_block_level",
[
("1-80 PRINCESS ALICE HOUSE LONDON", True),
("FLATS A-D 7 ST CHARLES SQUARE LONDON", True),
("9A-9H HEDGEGATE COURT LONDON", True),
("BLOCK MILNE HOUSE LONDON", True),
("81A-B GORE ROAD LONDON", True),
("73 & 74 HARVEST COURT ST. ALBANS", True),
("25 HAVERSHAM COURT GREENFORD", False),
("FLAT 10 SPARROW COURT SOUTHMERE DRIVE LONDON SE2 9ES", False)
],
)
def test_peabody_asset_is_block_level(
asset_condition_factory,
full_address,
expected_block_level,
):
# arrange
asset_condition = asset_condition_factory(full_address)
# act + assert
assert asset_condition.is_block_level == expected_block_level