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

docs(ara): composable stage-orchestrator design (ADR-0011 + ADR-0003 amend + CONTEXT)

Browse source 
Records the grill-with-docs outcomes for the ara_first_run rebuild: three
composable stage orchestrators (Ingestion/Baseline/Modelling), one lambda per
use case chaining them through repos (not in-memory), and the Fetcher-vs-Repo
data-source taxonomy. Amends ADR-0003's chaining rule to generalise beyond
RefreshOrchestrator. Adds the pipeline-composition + First Run vocabulary to
CONTEXT.md.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-05-30 19:26:17 +00:00 • committed by Jun-te Kim
parent ca6a0efd70
commit 8291f29721
4 changed files with 780 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
CONTEXT.md
View file

@ -129,6 +129,26 @@ _Avoid_: UCL adjustment, energy correction, metered correction
A per-field indicator that a Property's value for an EPC field differs significantly from Comparable Properties; advisory only — surfaces in the UI to prompt user review, does not block modelling.
_Avoid_: outlier, mismatch, divergence flag
### Pipeline composition
The modelling backend is composed from three independently-invocable **stage orchestrators**, chained differently per use case. This composability — not a single end-to-end function — is the point: it is what lets the interactive single-property flow pause between stages where the batch flows do not. (Supersedes the monolithic `model_engine`.)
**Ingestion**:
The first stage. Acquires a Property's external source data — the EPC certificate (New EPC API) and Google Solar insights — and resolves its coordinates, then writes everything to repos. Writes only; runs no modelling business logic. Per ADR-0003 nothing downstream reads across this seam by calling back to a source — downstream stages read the persisted data from repos.
_Avoid_: fetching (a fetch is one source call; Ingestion is the whole write stage), data load
**Baseline** (stage):
The second stage. Reads the persisted source data from repos, hydrates the **Property** aggregate, resolves its **Effective EPC**, and establishes its **Baseline Performance**. Re-scoring after a user override lives here. Distinct from **Baseline Performance** (the aggregate it produces).
_Avoid_: rebaseline (that is a specific ML trigger — see Rebaselining), enrichment
**Modelling** (stage):
The third stage. Takes the baselined Property plus a set of **Scenarios** and produces **Recommendations** → an **Optimised Package** per **Scenario Phase** → **Plans**, persisted to repos. A separate orchestrator from Baseline so the single-property flow can stop after Baseline and only run Modelling when the user hits "play".
_Avoid_: scoring (overloaded), recommendation engine
**First Run**:
The use case where a Property has only a row in the property table (post address→UPRN matching) and no existing **Plan**: the pipeline runs Ingestion → Baseline → Modelling end-to-end over a batch. The first sibling lambda being built (`ara_first_run`).
_Avoid_: initial run, cold run
### ML training
**EPC ML Transform**:

3
docs/adr/0003-strict-ingestion-modelling-separation.md
View file

@ -1,5 +1,8 @@
# Strict separation between Ingestion and Modelling
**Status: Accepted, refined by [ADR-0011](0011-composable-stage-orchestrators.md).** The one-way flow below stands. ADR-0011 generalises the chaining rule: it is no longer "only a `RefreshOrchestrator` may chain" — it is *"only a top-level use-case pipeline orchestrator (e.g. `FirstRunPipeline`) may chain across the Ingestion→Modelling seam; the stage orchestrators communicate through repos and never call across it."*
Data flows one way only: **Ingestion → Repos → Modelling**. Modelling services never make external HTTP calls; Ingestion services never run business logic. If Modelling needs fresh data, it sees a stale record in a repo and returns; the caller (a refresh orchestrator or the FE) decides whether to ingest first. We considered allowing modelling services to call fetchers directly on cache miss — convenient — and rejected it.
The trade-off is that modelling cannot "self-heal" by going to the gov EPC API when it finds stale data. The benefit is that modelling becomes a deterministic function of repository state: same Property in the repos, same modelling output. That is the property that makes modelling unit-testable against fakes (no DB, no network, no ML lambda), reproducible, and debuggable. It also enables a per-property UI flow where fetched data is shown to the user for review and possible override **before** modelling runs.

41
docs/adr/0011-composable-stage-orchestrators.md Normal file
View file

@ -0,0 +1,41 @@
# Composable stage orchestrators; one lambda per use case; stages communicate through repos
**Status: Accepted.** Refines [ADR-0003](0003-strict-ingestion-modelling-separation.md) (Ingestion→Repos→Modelling one-way flow) for the concrete shape of the rebuilt backend. Decided in a `/grill-with-docs` session (2026-05-30) before the first `ara_first_run` slice. Replaces the stale §4 / §9 / §11 architecture of `ara_backend_design.md`, which predates this thinking.
## Context
The pipeline must serve three use cases from the *same building blocks*:
- **First Run** (batch) — a property has only a row in the property table; run everything end-to-end.
- **Refresh** (batch) — re-check for new data and re-model if it changed.
- **Single-property interactive** (a new front end) — fetch, **pause** for the user to validate/override, re-score, **pause** again, then model on demand.
The single-property flow is the forcing function: it must be able to stop *between* establishing baseline data and producing recommendations. The legacy `model_engine` (one 1331-line function) cannot be re-entered partway, which is why it cannot serve this flow.
## Decision
**Three independently-invocable stage orchestrators**, in `orchestration/`:
| Stage | Reads | Writes | Role |
|---|---|---|---|
| `IngestionOrchestrator` | Fetchers (EPC, Solar) + reference Repos (Geospatial) | source Repos | acquire + persist external source data |
| `BaselineOrchestrator` | source Repos | `Property` + Baseline Performance | hydrate the aggregate; resolve Effective EPC; re-score on override |
| `ModellingOrchestrator` | baselined Repos + Scenario/Materials Repos | Plans / Recommendations Repos | scenarios → recommendations → optimise → plans |
**One lambda per use case** composes these via a thin pipeline object. `applications/ara_first_run/` is the first: a `handler.py` that only wires dependencies and delegates to a `FirstRunPipeline` (`Ingestion → Baseline → Modelling`). `refresh` and the single-property app are later siblings composing the *same three* stages differently.
**Stages communicate through the repos, not in-memory.** The pipeline threads only identifiers (`property_ids`) between stages; each stage reads what it needs from repos and writes its outputs back. Baseline is therefore byte-identical whether ingestion ran 50 ms ago (First Run) or last week (single-property review) — there is no second entry mode.
**Data-source taxonomy: "external" does not mean "Fetcher."** A **Fetcher** hits a *live, per-entity* API and returns raw data (infra client, no DB): the New EPC API, Google Solar. A **Repo** reads *stored data by key* — ours *or* a hosted reference dataset — and returns domain objects (no HTTP): Ordnance Survey Open-UPRN coordinates (`GeospatialRepo`), cost data (`MaterialsRepo`). When a fetch needs reference data (Solar needs lat/long), the **orchestrator** reads the repo and threads the value into the fetcher; fetchers never call each other.
## Considered options
- **One lambda per stage, coordinated by AWS Step Functions** — rejected. Step Functions buys cross-lambda completion signalling we don't need when the three stages are cheap to keep warm in one process and a batch is bite-size (≤~100 properties). Promoting a stage to its own lambda later is cheap *because* it is already a separate class.
- **In-memory hand-off between stages in First Run** — rejected as the default. It gives `BaselineOrchestrator` two entry modes (fresh object vs repo read) and hides EPC persistence loss until a later Refresh reads the data back. Going through repos surfaces that loss inside First Run on day one. May be added later as an opt-in fast path where a profiler justifies it.
## Consequences
- A few redundant reads of rows just written, within one process — negligible at batch scale, and the price of each stage being a pure function of repo state.
- Each stage is unit-testable against fake repos with no upstream stage present.
- No HTTP library may appear in the `BaselineOrchestrator` / `ModellingOrchestrator` import graph (ADR-0003 holds per-stage).
- Because stages round-trip `EpcPropertyData` through persistence in First Run, a **persistence round-trip fidelity test** (fetch EPCs across schema versions → map → save → load → map back → assert deep-equality) is a prerequisite deliverable: it is what proves `epc_property` + child tables actually cover the domain object, and surfaces any required FE-owned migration early.

716
infrastructure/postgres/epc_property_table.py Normal file
View file

@ -0,0 +1,716 @@
from __future__ import annotations
from typing import ClassVar, Optional, Union
from sqlalchemy import Column
from sqlalchemy.dialects.postgresql import JSONB
from sqlmodel import SQLModel, Field
from datatypes.epc.domain.epc_property_data import (
EpcPropertyData,
EnergyElement,
MainHeatingDetail,
SapBuildingPart,
SapFloorDimension,
SapFlatDetails,
SapWindow,
)
class EpcPropertyModel(SQLModel, table=True):
__tablename__: ClassVar[str] = "epc_property" # pyright: ignore[reportIncompatibleVariableOverride]
id: Optional[int] = Field(default=None, primary_key=True)
property_id: Optional[int] = Field(default=None)
portfolio_id: Optional[int] = Field(default=None)
uploaded_file_id: Optional[int] = Field(default=None)
# Identity / admin
uprn: Optional[int] = Field(default=None)
uprn_source: Optional[str] = Field(default=None)
report_reference: Optional[str] = Field(default=None)
report_type: Optional[str] = Field(default=None)
assessment_type: Optional[str] = Field(default=None)
sap_version: Optional[float] = Field(default=None)
schema_type: Optional[str] = Field(default=None)
schema_versions_original: Optional[str] = Field(default=None)
status: Optional[str] = Field(default=None)
calculation_software_version: Optional[str] = Field(default=None)
# Address
address_line_1: Optional[str] = Field(default=None)
address_line_2: Optional[str] = Field(default=None)
post_town: Optional[str] = Field(default=None)
postcode: Optional[str] = Field(default=None)
region_code: Optional[str] = Field(default=None)
country_code: Optional[str] = Field(default=None)
language_code: Optional[str] = Field(default=None)
# Property description
dwelling_type: str
property_type: Optional[str] = Field(default=None)
built_form: Optional[str] = Field(default=None)
tenure: str
transaction_type: str
inspection_date: str # store as ISO string; cast on read if needed
completion_date: Optional[str] = Field(default=None)
registration_date: Optional[str] = Field(default=None)
total_floor_area_m2: float
measurement_type: Optional[int] = Field(default=None)
# Flags
solar_water_heating: bool
has_hot_water_cylinder: bool
has_fixed_air_conditioning: bool
has_conservatory: Optional[bool] = Field(default=None)
has_heated_separate_conservatory: Optional[bool] = Field(default=None)
conservatory_type: Optional[int] = Field(default=None)
# Counts
door_count: int
wet_rooms_count: int
extensions_count: int
heated_rooms_count: int
open_chimneys_count: int
habitable_rooms_count: int
insulated_door_count: int
cfl_fixed_lighting_bulbs_count: int
led_fixed_lighting_bulbs_count: int
incandescent_fixed_lighting_bulbs_count: int
blocked_chimneys_count: Optional[int] = Field(default=None)
draughtproofed_door_count: Optional[int] = Field(default=None)
energy_rating_average: Optional[int] = Field(default=None)
low_energy_fixed_lighting_bulbs_count: Optional[int] = Field(default=None)
fixed_lighting_outlets_count: Optional[int] = Field(default=None)
low_energy_fixed_lighting_outlets_count: Optional[int] = Field(default=None)
number_of_storeys: Optional[int] = Field(default=None)
any_unheated_rooms: Optional[bool] = Field(default=None)
mechanical_vent_duct_insulation_level: Optional[int] = Field(default=None)
# Addendum (cert-level construction flags)
addendum_stone_walls: Optional[bool] = Field(default=None)
addendum_system_build: Optional[bool] = Field(default=None)
addendum_numbers: Optional[list[int]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
# Misc
hydro: Optional[bool] = Field(default=None)
photovoltaic_array: Optional[bool] = Field(default=None)
waste_water_heat_recovery: Optional[str] = Field(default=None)
pressure_test: Optional[int] = Field(default=None)
pressure_test_certificate_number: Optional[int] = Field(default=None)
percent_draughtproofed: Optional[int] = Field(default=None)
insulated_door_u_value: Optional[float] = Field(default=None)
multiple_glazed_proportion: Optional[int] = Field(default=None)
windows_transmission_u_value: Optional[float] = Field(default=None)
windows_transmission_data_source: Optional[int] = Field(default=None)
windows_transmission_solar_transmittance: Optional[float] = Field(default=None)
# Energy source
energy_mains_gas: bool
energy_meter_type: str
energy_pv_battery_count: int
energy_wind_turbines_count: int
energy_gas_smart_meter_present: bool
energy_is_dwelling_export_capable: bool
energy_wind_turbines_terrain_type: str
energy_electricity_smart_meter_present: bool
energy_pv_connection: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
energy_pv_percent_roof_area: Optional[int] = Field(default=None)
energy_pv_battery_capacity: Optional[float] = Field(default=None)
energy_wind_turbine_hub_height: Optional[float] = Field(default=None)
energy_wind_turbine_rotor_diameter: Optional[float] = Field(default=None)
# Heating config
# Union[int, str] code fields stored as JSONB to preserve the int (API) vs
# str (Site Notes) distinction on round-trip (see docs/migrations/epc-property-round-trip-fidelity.md §1).
heating_cylinder_size: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
heating_water_heating_code: Optional[int] = Field(default=None)
heating_water_heating_fuel: Optional[int] = Field(default=None)
heating_immersion_heating_type: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
heating_cylinder_insulation_type: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
heating_cylinder_thermostat: Optional[str] = Field(default=None)
heating_secondary_fuel_type: Optional[int] = Field(default=None)
heating_secondary_heating_type: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
heating_cylinder_insulation_thickness_mm: Optional[int] = Field(default=None)
heating_wwhrs_index_number_1: Optional[int] = Field(default=None)
heating_wwhrs_index_number_2: Optional[int] = Field(default=None)
heating_shower_outlet_type: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
heating_shower_wwhrs: Optional[int] = Field(default=None)
heating_number_baths: Optional[int] = Field(default=None)
heating_number_baths_wwhrs: Optional[int] = Field(default=None)
heating_electric_shower_count: Optional[int] = Field(default=None)
heating_mixer_shower_count: Optional[int] = Field(default=None)
# Ventilation
ventilation_type: Optional[str] = Field(default=None)
ventilation_draught_lobby: Optional[bool] = Field(default=None)
ventilation_pressure_test: Optional[str] = Field(default=None)
ventilation_open_flues_count: Optional[int] = Field(default=None)
ventilation_closed_flues_count: Optional[int] = Field(default=None)
ventilation_boiler_flues_count: Optional[int] = Field(default=None)
ventilation_other_flues_count: Optional[int] = Field(default=None)
ventilation_extract_fans_count: Optional[int] = Field(default=None)
ventilation_passive_vents_count: Optional[int] = Field(default=None)
ventilation_flueless_gas_fires_count: Optional[int] = Field(default=None)
ventilation_in_pcdf_database: Optional[bool] = Field(default=None)
# SAP 10.2 §2 lodgements + a presence flag so an all-None SapVentilation
# round-trips as present (not collapsed to None).
ventilation_present: bool = Field(default=False)
ventilation_sheltered_sides: Optional[int] = Field(default=None)
ventilation_has_suspended_timber_floor: Optional[bool] = Field(default=None)
ventilation_suspended_timber_floor_sealed: Optional[bool] = Field(default=None)
ventilation_has_draught_lobby: Optional[bool] = Field(default=None)
ventilation_air_permeability_ap4_m3_h_m2: Optional[float] = Field(default=None)
ventilation_mechanical_ventilation_kind: Optional[str] = Field(default=None)
mechanical_ventilation: Optional[int] = Field(default=None)
mechanical_vent_duct_type: Optional[int] = Field(default=None)
mechanical_vent_duct_placement: Optional[int] = Field(default=None)
mechanical_vent_duct_insulation: Optional[int] = Field(default=None)
mechanical_ventilation_index_number: Optional[int] = Field(default=None)
mechanical_vent_measured_installation: Optional[str] = Field(default=None)
@classmethod
def from_epc_property_data(
cls,
data: EpcPropertyData,
property_id: Optional[int] = None,
portfolio_id: Optional[int] = None,
) -> EpcPropertyModel:
es = data.sap_energy_source
h = data.sap_heating
v = data.sap_ventilation
shower = h.shower_outlets.shower_outlet if h.shower_outlets else None
pv = es.photovoltaic_supply
wt = es.wind_turbine_details
pvb = es.pv_batteries
return cls(
property_id=property_id,
portfolio_id=portfolio_id,
uprn=data.uprn,
uprn_source=data.uprn_source,
report_reference=data.report_reference,
report_type=data.report_type,
assessment_type=data.assessment_type,
sap_version=data.sap_version,
schema_type=data.schema_type,
schema_versions_original=data.schema_versions_original,
status=data.status,
calculation_software_version=data.calculation_software_version,
address_line_1=data.address_line_1,
address_line_2=data.address_line_2,
post_town=data.post_town,
postcode=data.postcode,
region_code=data.region_code,
country_code=data.country_code,
language_code=data.language_code,
dwelling_type=data.dwelling_type,
property_type=data.property_type,
built_form=data.built_form,
tenure=data.tenure,
transaction_type=data.transaction_type,
inspection_date=data.inspection_date.isoformat(),
completion_date=(
data.completion_date.isoformat() if data.completion_date else None
),
registration_date=(
data.registration_date.isoformat() if data.registration_date else None
),
total_floor_area_m2=data.total_floor_area_m2,
measurement_type=data.measurement_type,
solar_water_heating=data.solar_water_heating,
has_hot_water_cylinder=data.has_hot_water_cylinder,
has_fixed_air_conditioning=data.has_fixed_air_conditioning,
has_conservatory=data.has_conservatory,
has_heated_separate_conservatory=data.has_heated_separate_conservatory,
conservatory_type=data.conservatory_type,
door_count=data.door_count,
wet_rooms_count=data.wet_rooms_count,
extensions_count=data.extensions_count,
heated_rooms_count=data.heated_rooms_count,
open_chimneys_count=data.open_chimneys_count,
habitable_rooms_count=data.habitable_rooms_count,
insulated_door_count=data.insulated_door_count,
cfl_fixed_lighting_bulbs_count=data.cfl_fixed_lighting_bulbs_count,
led_fixed_lighting_bulbs_count=data.led_fixed_lighting_bulbs_count,
incandescent_fixed_lighting_bulbs_count=data.incandescent_fixed_lighting_bulbs_count,
blocked_chimneys_count=data.blocked_chimneys_count,
draughtproofed_door_count=data.draughtproofed_door_count,
energy_rating_average=data.energy_rating_average,
low_energy_fixed_lighting_bulbs_count=data.low_energy_fixed_lighting_bulbs_count,
fixed_lighting_outlets_count=data.fixed_lighting_outlets_count,
low_energy_fixed_lighting_outlets_count=data.low_energy_fixed_lighting_outlets_count,
number_of_storeys=data.number_of_storeys,
any_unheated_rooms=data.any_unheated_rooms,
mechanical_vent_duct_insulation_level=data.mechanical_vent_duct_insulation_level,
addendum_stone_walls=data.addendum.stone_walls if data.addendum else None,
addendum_system_build=(
data.addendum.system_build if data.addendum else None
),
addendum_numbers=data.addendum.addendum_numbers if data.addendum else None,
hydro=data.hydro,
photovoltaic_array=data.photovoltaic_array,
waste_water_heat_recovery=data.waste_water_heat_recovery,
pressure_test=data.pressure_test,
pressure_test_certificate_number=data.pressure_test_certificate_number,
percent_draughtproofed=data.percent_draughtproofed,
insulated_door_u_value=data.insulated_door_u_value,
multiple_glazed_proportion=data.multiple_glazed_proportion,
windows_transmission_u_value=(
data.windows_transmission_details.u_value
if data.windows_transmission_details
else None
),
windows_transmission_data_source=(
data.windows_transmission_details.data_source
if data.windows_transmission_details
else None
),
windows_transmission_solar_transmittance=(
data.windows_transmission_details.solar_transmittance
if data.windows_transmission_details
else None
),
energy_mains_gas=es.mains_gas,
energy_meter_type=str(es.meter_type),
energy_pv_battery_count=es.pv_battery_count,
energy_wind_turbines_count=es.wind_turbines_count,
energy_gas_smart_meter_present=es.gas_smart_meter_present,
energy_is_dwelling_export_capable=es.is_dwelling_export_capable,
energy_wind_turbines_terrain_type=str(es.wind_turbines_terrain_type),
energy_electricity_smart_meter_present=es.electricity_smart_meter_present,
energy_pv_connection=es.pv_connection,
energy_pv_percent_roof_area=(
pv.none_or_no_details.percent_roof_area if pv else None
),
energy_pv_battery_capacity=pvb.pv_battery.battery_capacity if pvb else None,
energy_wind_turbine_hub_height=wt.hub_height if wt else None,
energy_wind_turbine_rotor_diameter=wt.rotor_diameter if wt else None,
heating_cylinder_size=h.cylinder_size,
heating_water_heating_code=h.water_heating_code,
heating_water_heating_fuel=h.water_heating_fuel,
heating_immersion_heating_type=h.immersion_heating_type,
heating_cylinder_insulation_type=h.cylinder_insulation_type,
heating_cylinder_thermostat=h.cylinder_thermostat,
heating_secondary_fuel_type=h.secondary_fuel_type,
heating_secondary_heating_type=h.secondary_heating_type,
heating_cylinder_insulation_thickness_mm=h.cylinder_insulation_thickness_mm,
heating_wwhrs_index_number_1=h.instantaneous_wwhrs.wwhrs_index_number1,
heating_wwhrs_index_number_2=h.instantaneous_wwhrs.wwhrs_index_number2,
heating_shower_outlet_type=shower.shower_outlet_type if shower else None,
heating_shower_wwhrs=shower.shower_wwhrs if shower else None,
heating_number_baths=h.number_baths,
heating_number_baths_wwhrs=h.number_baths_wwhrs,
heating_electric_shower_count=h.electric_shower_count,
heating_mixer_shower_count=h.mixer_shower_count,
ventilation_type=v.ventilation_type if v else None,
ventilation_draught_lobby=v.draught_lobby if v else None,
ventilation_pressure_test=v.pressure_test if v else None,
ventilation_open_flues_count=v.open_flues_count if v else None,
ventilation_closed_flues_count=v.closed_flues_count if v else None,
ventilation_boiler_flues_count=v.boiler_flues_count if v else None,
ventilation_other_flues_count=v.other_flues_count if v else None,
ventilation_extract_fans_count=v.extract_fans_count if v else None,
ventilation_passive_vents_count=v.passive_vents_count if v else None,
ventilation_flueless_gas_fires_count=(
v.flueless_gas_fires_count if v else None
),
ventilation_in_pcdf_database=v.ventilation_in_pcdf_database if v else None,
ventilation_present=v is not None,
ventilation_sheltered_sides=v.sheltered_sides if v else None,
ventilation_has_suspended_timber_floor=(
v.has_suspended_timber_floor if v else None
),
ventilation_suspended_timber_floor_sealed=(
v.suspended_timber_floor_sealed if v else None
),
ventilation_has_draught_lobby=v.has_draught_lobby if v else None,
ventilation_air_permeability_ap4_m3_h_m2=(
v.air_permeability_ap4_m3_h_m2 if v else None
),
ventilation_mechanical_ventilation_kind=(
v.mechanical_ventilation_kind if v else None
),
mechanical_ventilation=data.mechanical_ventilation,
mechanical_vent_duct_type=data.mechanical_vent_duct_type,
mechanical_vent_duct_placement=data.mechanical_vent_duct_placement,
mechanical_vent_duct_insulation=data.mechanical_vent_duct_insulation,
mechanical_ventilation_index_number=data.mechanical_ventilation_index_number,
mechanical_vent_measured_installation=data.mechanical_vent_measured_installation,
)
class EpcPropertyEnergyPerformanceModel(SQLModel, table=True):
__tablename__: ClassVar[str] = "epc_property_energy_performance" # pyright: ignore[reportIncompatibleVariableOverride]
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(
foreign_key="epc_property.id", nullable=False, unique=True
)
energy_rating_current: Optional[int] = Field(default=None)
energy_consumption_current: Optional[int] = Field(default=None)
environmental_impact_current: Optional[int] = Field(default=None)
heating_cost_current: Optional[float] = Field(default=None)
lighting_cost_current: Optional[float] = Field(default=None)
hot_water_cost_current: Optional[float] = Field(default=None)
co2_emissions_current: Optional[float] = Field(default=None)
co2_emissions_current_per_floor_area: Optional[int] = Field(default=None)
current_energy_efficiency_band: Optional[str] = Field(default=None)
energy_rating_potential: Optional[float] = Field(default=None)
energy_consumption_potential: Optional[int] = Field(default=None)
environmental_impact_potential: Optional[int] = Field(default=None)
heating_cost_potential: Optional[float] = Field(default=None)
lighting_cost_potential: Optional[float] = Field(default=None)
hot_water_cost_potential: Optional[float] = Field(default=None)
co2_emissions_potential: Optional[float] = Field(default=None)
potential_energy_efficiency_band: Optional[str] = Field(default=None)
@classmethod
def from_epc_property_data(
cls, data: EpcPropertyData, epc_property_id: int
) -> EpcPropertyEnergyPerformanceModel:
return cls(
epc_property_id=epc_property_id,
energy_rating_current=data.energy_rating_current,
energy_consumption_current=data.energy_consumption_current,
environmental_impact_current=data.environmental_impact_current,
heating_cost_current=data.heating_cost_current,
lighting_cost_current=data.lighting_cost_current,
hot_water_cost_current=data.hot_water_cost_current,
co2_emissions_current=data.co2_emissions_current,
co2_emissions_current_per_floor_area=data.co2_emissions_current_per_floor_area,
current_energy_efficiency_band=(
data.current_energy_efficiency_band.value
if data.current_energy_efficiency_band
else None
),
energy_rating_potential=data.energy_rating_potential,
energy_consumption_potential=data.energy_consumption_potential,
environmental_impact_potential=data.environmental_impact_potential,
heating_cost_potential=data.heating_cost_potential,
lighting_cost_potential=data.lighting_cost_potential,
hot_water_cost_potential=data.hot_water_cost_potential,
co2_emissions_potential=data.co2_emissions_potential,
potential_energy_efficiency_band=(
data.potential_energy_efficiency_band.value
if data.potential_energy_efficiency_band
else None
),
)
class EpcFlatDetailsModel(SQLModel, table=True):
__tablename__: ClassVar[str] = "epc_flat_details" # pyright: ignore[reportIncompatibleVariableOverride]
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(
foreign_key="epc_property.id", nullable=False, unique=True
)
level: int
top_storey: str
flat_location: int
heat_loss_corridor: int
storey_count: Optional[int] = Field(default=None)
unheated_corridor_length_m: Optional[int] = Field(default=None)
@classmethod
def from_domain(
cls, flat: SapFlatDetails, epc_property_id: int
) -> EpcFlatDetailsModel:
return cls(
epc_property_id=epc_property_id,
level=flat.level,
top_storey=flat.top_storey,
flat_location=flat.flat_location,
heat_loss_corridor=flat.heat_loss_corridor,
storey_count=flat.storey_count,
unheated_corridor_length_m=flat.unheated_corridor_length_m,
)
class EpcMainHeatingDetailModel(SQLModel, table=True):
__tablename__: ClassVar[str] = "epc_main_heating_detail" # pyright: ignore[reportIncompatibleVariableOverride]
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(foreign_key="epc_property.id", nullable=False)
has_fghrs: bool
# Union[int, str] code fields — JSONB to preserve int/str on round-trip.
main_fuel_type: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
heat_emitter_type: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
emitter_temperature: Union[int, str] = Field(
sa_column=Column(JSONB, nullable=False)
)
main_heating_control: Union[int, str] = Field(
sa_column=Column(JSONB, nullable=False)
)
fan_flue_present: Optional[bool] = Field(default=None)
boiler_flue_type: Optional[int] = Field(default=None)
boiler_ignition_type: Optional[int] = Field(default=None)
central_heating_pump_age: Optional[int] = Field(default=None)
central_heating_pump_age_str: Optional[str] = Field(default=None)
main_heating_index_number: Optional[int] = Field(default=None)
sap_main_heating_code: Optional[int] = Field(default=None)
main_heating_number: Optional[int] = Field(default=None)
main_heating_category: Optional[int] = Field(default=None)
main_heating_fraction: Optional[int] = Field(default=None)
main_heating_data_source: Optional[int] = Field(default=None)
condensing: Optional[bool] = Field(default=None)
weather_compensator: Optional[bool] = Field(default=None)
@classmethod
def from_domain(
cls, detail: MainHeatingDetail, epc_property_id: int
) -> EpcMainHeatingDetailModel:
return cls(
epc_property_id=epc_property_id,
has_fghrs=detail.has_fghrs,
main_fuel_type=detail.main_fuel_type,
heat_emitter_type=detail.heat_emitter_type,
emitter_temperature=detail.emitter_temperature,
main_heating_control=detail.main_heating_control,
fan_flue_present=detail.fan_flue_present,
boiler_flue_type=detail.boiler_flue_type,
boiler_ignition_type=detail.boiler_ignition_type,
central_heating_pump_age=detail.central_heating_pump_age,
central_heating_pump_age_str=detail.central_heating_pump_age_str,
main_heating_index_number=detail.main_heating_index_number,
sap_main_heating_code=detail.sap_main_heating_code,
main_heating_number=detail.main_heating_number,
main_heating_category=detail.main_heating_category,
main_heating_fraction=detail.main_heating_fraction,
main_heating_data_source=detail.main_heating_data_source,
condensing=detail.condensing,
weather_compensator=detail.weather_compensator,
)
class EpcBuildingPartModel(SQLModel, table=True):
__tablename__: ClassVar[str] = "epc_building_part" # pyright: ignore[reportIncompatibleVariableOverride]
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(foreign_key="epc_property.id", nullable=False)
identifier: str
construction_age_band: str
# Union[int, str] code fields — JSONB to preserve int/str on round-trip.
wall_construction: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
wall_insulation_type: Union[int, str] = Field(
sa_column=Column(JSONB, nullable=False)
)
wall_thickness_measured: bool
party_wall_construction: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
building_part_number: Optional[int] = Field(default=None)
wall_dry_lined: Optional[bool] = Field(default=None)
wall_thickness_mm: Optional[int] = Field(default=None)
wall_insulation_thickness: Optional[str] = Field(default=None)
floor_heat_loss: Optional[int] = Field(default=None)
floor_insulation_thickness: Optional[str] = Field(default=None)
flat_roof_insulation_thickness: Optional[Union[str, int]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
floor_type: Optional[str] = Field(default=None)
floor_construction_type: Optional[str] = Field(default=None)
floor_insulation_type_str: Optional[str] = Field(default=None)
floor_u_value_known: Optional[bool] = Field(default=None)
roof_construction: Optional[int] = Field(default=None)
roof_construction_type: Optional[str] = Field(default=None)
curtain_wall_age: Optional[str] = Field(default=None)
roof_insulation_location: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
roof_insulation_thickness: Optional[Union[str, int]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
room_in_roof_floor_area: Optional[float] = Field(default=None)
room_in_roof_construction_age_band: Optional[str] = Field(default=None)
alt_wall_1_area: Optional[float] = Field(default=None)
alt_wall_1_dry_lined: Optional[str] = Field(default=None)
alt_wall_1_construction: Optional[int] = Field(default=None)
alt_wall_1_insulation_type: Optional[int] = Field(default=None)
alt_wall_1_thickness_measured: Optional[str] = Field(default=None)
alt_wall_1_insulation_thickness: Optional[str] = Field(default=None)
alt_wall_2_area: Optional[float] = Field(default=None)
alt_wall_2_dry_lined: Optional[str] = Field(default=None)
alt_wall_2_construction: Optional[int] = Field(default=None)
alt_wall_2_insulation_type: Optional[int] = Field(default=None)
alt_wall_2_thickness_measured: Optional[str] = Field(default=None)
alt_wall_2_insulation_thickness: Optional[str] = Field(default=None)
@classmethod
def from_domain(
cls, part: SapBuildingPart, epc_property_id: int
) -> EpcBuildingPartModel:
rir = part.sap_room_in_roof
aw1 = part.sap_alternative_wall_1
aw2 = part.sap_alternative_wall_2
return cls(
epc_property_id=epc_property_id,
identifier=part.identifier.value,
construction_age_band=part.construction_age_band,
wall_construction=part.wall_construction,
wall_insulation_type=part.wall_insulation_type,
wall_thickness_measured=part.wall_thickness_measured,
party_wall_construction=part.party_wall_construction,
building_part_number=part.building_part_number,
wall_dry_lined=part.wall_dry_lined,
wall_thickness_mm=part.wall_thickness_mm,
wall_insulation_thickness=part.wall_insulation_thickness,
floor_heat_loss=part.floor_heat_loss,
floor_insulation_thickness=part.floor_insulation_thickness,
flat_roof_insulation_thickness=part.flat_roof_insulation_thickness,
floor_type=part.floor_type,
floor_construction_type=part.floor_construction_type,
floor_insulation_type_str=part.floor_insulation_type_str,
floor_u_value_known=part.floor_u_value_known,
roof_construction=part.roof_construction,
roof_construction_type=part.roof_construction_type,
curtain_wall_age=part.curtain_wall_age,
roof_insulation_location=part.roof_insulation_location,
roof_insulation_thickness=part.roof_insulation_thickness,
room_in_roof_floor_area=float(rir.floor_area) if rir else None,
room_in_roof_construction_age_band=(
rir.construction_age_band if rir else None
),
alt_wall_1_area=aw1.wall_area if aw1 else None,
alt_wall_1_dry_lined=aw1.wall_dry_lined if aw1 else None,
alt_wall_1_construction=aw1.wall_construction if aw1 else None,
alt_wall_1_insulation_type=aw1.wall_insulation_type if aw1 else None,
alt_wall_1_thickness_measured=aw1.wall_thickness_measured if aw1 else None,
alt_wall_1_insulation_thickness=(
aw1.wall_insulation_thickness if aw1 else None
),
alt_wall_2_area=aw2.wall_area if aw2 else None,
alt_wall_2_dry_lined=aw2.wall_dry_lined if aw2 else None,
alt_wall_2_construction=aw2.wall_construction if aw2 else None,
alt_wall_2_insulation_type=aw2.wall_insulation_type if aw2 else None,
alt_wall_2_thickness_measured=aw2.wall_thickness_measured if aw2 else None,
alt_wall_2_insulation_thickness=(
aw2.wall_insulation_thickness if aw2 else None
),
)
class EpcFloorDimensionModel(SQLModel, table=True):
__tablename__: ClassVar[str] = "epc_floor_dimension" # pyright: ignore[reportIncompatibleVariableOverride]
id: Optional[int] = Field(default=None, primary_key=True)
epc_building_part_id: int = Field(
foreign_key="epc_building_part.id", nullable=False
)
floor: Optional[int] = Field(default=None)
room_height_m: float
total_floor_area_m2: float
party_wall_length_m: float
heat_loss_perimeter_m: float
floor_insulation: Optional[int] = Field(default=None)
floor_construction: Optional[int] = Field(default=None)
@classmethod
def from_domain(
cls, dim: SapFloorDimension, epc_building_part_id: int
) -> EpcFloorDimensionModel:
return cls(
epc_building_part_id=epc_building_part_id,
floor=dim.floor,
room_height_m=dim.room_height_m,
total_floor_area_m2=dim.total_floor_area_m2,
party_wall_length_m=dim.party_wall_length_m,
heat_loss_perimeter_m=dim.heat_loss_perimeter_m,
floor_insulation=dim.floor_insulation,
floor_construction=dim.floor_construction,
)
class EpcWindowModel(SQLModel, table=True):
__tablename__: ClassVar[str] = "epc_window" # pyright: ignore[reportIncompatibleVariableOverride]
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(foreign_key="epc_property.id", nullable=False)
frame_material: Optional[str] = Field(default=None)
# Union[int, str] / Union[bool, str] code fields — JSONB to preserve type on round-trip.
glazing_gap: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
orientation: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
window_type: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
glazing_type: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
window_width: float
window_height: float
draught_proofed: Union[bool, str] = Field(sa_column=Column(JSONB, nullable=False))
window_location: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
window_wall_type: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
permanent_shutters_present: Union[bool, str] = Field(
sa_column=Column(JSONB, nullable=False)
)
frame_factor: Optional[float] = Field(default=None)
permanent_shutters_insulated: Optional[str] = Field(default=None)
transmission_u_value: Optional[float] = Field(default=None)
transmission_data_source: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
transmission_solar_transmittance: Optional[float] = Field(default=None)
@classmethod
def from_domain(cls, window: SapWindow, epc_property_id: int) -> EpcWindowModel:
td = window.window_transmission_details
return cls(
epc_property_id=epc_property_id,
frame_material=window.frame_material,
glazing_gap=window.glazing_gap,
orientation=window.orientation,
window_type=window.window_type,
glazing_type=window.glazing_type,
window_width=window.window_width,
window_height=window.window_height,
draught_proofed=window.draught_proofed,
window_location=window.window_location,
window_wall_type=window.window_wall_type,
permanent_shutters_present=window.permanent_shutters_present,
frame_factor=window.frame_factor,
permanent_shutters_insulated=window.permanent_shutters_insulated,
transmission_u_value=td.u_value if td else None,
transmission_data_source=td.data_source if td else None,
transmission_solar_transmittance=td.solar_transmittance if td else None,
)
class EpcEnergyElementModel(SQLModel, table=True):
__tablename__: ClassVar[str] = "epc_energy_element" # pyright: ignore[reportIncompatibleVariableOverride]
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(foreign_key="epc_property.id", nullable=False)
element_type: str # roof | wall | floor | main_heating | window | lighting | hot_water | secondary_heating | main_heating_controls
description: str
energy_efficiency_rating: int
environmental_efficiency_rating: int
@classmethod
def from_domain(
cls, element: EnergyElement, element_type: str, epc_property_id: int
) -> EpcEnergyElementModel:
return cls(
epc_property_id=epc_property_id,
element_type=element_type,
description=element.description,
energy_efficiency_rating=element.energy_efficiency_rating,
environmental_efficiency_rating=element.environmental_efficiency_rating,
)