junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

17.1 and 18 done by claude

Browse source
This commit is contained in:
Jun-te Kim 2026-06-12 12:52:36 +00:00
parent 1ff50374e7
commit 32de7f6c3f
18 changed files with 367 additions and 117 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
harness/console.py
View file

@ -54,6 +54,7 @@ DEFAULT_CATALOGUE = Path(__file__).resolve().parent / "sample_catalogue.json"
_PROPERTY_ID = 1
_SCENARIO_ID = 7
_PORTFOLIO_ID = 1
_UPRN = 12345
@dataclass
@ -108,7 +109,7 @@ def run_one(
portfolio_id=_PORTFOLIO_ID,
postcode="A0 0AA",
address="1 Some Street",
uprn=12345,
uprn=_UPRN,
),
current_market_value=current_market_value,
)
@ -211,7 +212,7 @@ def run_modelling(
portfolio_id=_PORTFOLIO_ID,
postcode="A0 0AA",
address="1 Some Street",
uprn=12345,
uprn=_UPRN,
),
epc=epc,
current_market_value=current_market_value,
@ -222,7 +223,7 @@ def run_modelling(
unit = FakeUnitOfWork(
property=property_repo,
solar=FakeSolarRepo(
by_property={_PROPERTY_ID: solar_insights}
by_uprn={_UPRN: solar_insights}
if solar_insights is not None
else None
),

25
infrastructure/postgres/epc_property_table.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
from typing import ClassVar, Optional, Union
from sqlalchemy import Column
from sqlalchemy import Enum as SAEnum
from sqlalchemy.dialects.postgresql import JSONB
from sqlmodel import SQLModel, Field
@ -735,7 +736,29 @@ class EpcEnergyElementModel(SQLModel, table=True):
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
# The live column is the ``energy_element_type`` Postgres enum (owned by the
# Drizzle schema). Binding it as a bare ``str`` makes SQLAlchemy cast the
# value to VARCHAR, which Postgres won't implicitly coerce to the enum on
# INSERT — so declare the native enum explicitly (``create_type=False``: the
# type already exists). The values stay plain strings on the domain side.
element_type: str = Field(
sa_column=Column(
SAEnum(
"roof",
"wall",
"floor",
"main_heating",
"window",
"lighting",
"hot_water",
"secondary_heating",
"main_heating_controls",
name="energy_element_type",
create_type=False,
),
nullable=False,
)
)
description: str
energy_efficiency_rating: int
environmental_efficiency_rating: int

30
infrastructure/postgres/property_baseline_performance_table.py
View file

@ -1,7 +1,9 @@
from __future__ import annotations
from typing import ClassVar, Optional, cast
from typing import ClassVar, Optional, cast, get_args
from sqlalchemy import Column
from sqlalchemy import Enum as SAEnum
from sqlmodel import Field, SQLModel
from datatypes.epc.domain.epc import Epc
@ -10,6 +12,11 @@ from domain.property_baseline.property_baseline_performance import PropertyBasel
from domain.property_baseline.performance import Performance
from domain.property_baseline.rebaseliner import RebaselineReason
# Native-enum labels for the ``epc`` / ``rebaseline_reason`` columns, sourced
# from the domain types so the mirror can't drift from them.
_EPC_BANDS: tuple[str, ...] = tuple(band.value for band in Epc)
_REBASELINE_REASONS: tuple[str, ...] = get_args(RebaselineReason)
# Each Bill section's flat-column stem (``bill_{stem}_kwh`` / ``bill_{stem}_cost_gbp``).
_SECTION_COLUMN_STEM: dict[BillSection, str] = {
BillSection.HEATING: "heating",
@ -35,17 +42,32 @@ class PropertyBaselinePerformanceModel(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
property_id: int = Field(unique=True, index=True)
# ``*_epc_band`` and ``rebaseline_reason`` are native Postgres enums on the
# live table (``epc`` / ``rebaseline_reason``, owned by Drizzle); declaring
# them as bare ``str`` makes SQLAlchemy bind VARCHAR on INSERT, which
# Postgres won't coerce to the enum. Bind the native enums explicitly
# (``create_type=False``: the types already exist). Values stay plain
# strings on the domain side (``Epc(...)`` / the RebaselineReason Literal).
lodged_sap_score: int
lodged_epc_band: str
lodged_epc_band: str = Field(
sa_column=Column(SAEnum(*_EPC_BANDS, name="epc", create_type=False), nullable=False)
)
lodged_co2_emissions_t_per_yr: float
lodged_primary_energy_intensity_kwh_per_m2_yr: int
effective_sap_score: int
effective_epc_band: str
effective_epc_band: str = Field(
sa_column=Column(SAEnum(*_EPC_BANDS, name="epc", create_type=False), nullable=False)
)
effective_co2_emissions_t_per_yr: float
effective_primary_energy_intensity_kwh_per_m2_yr: int
rebaseline_reason: str
rebaseline_reason: str = Field(
sa_column=Column(
SAEnum(*_REBASELINE_REASONS, name="rebaseline_reason", create_type=False),
nullable=False,
)
)
space_heating_kwh: float
water_heating_kwh: float

29
infrastructure/postgres/solar_table.py
View file

@ -2,21 +2,30 @@ from __future__ import annotations
from typing import Any, ClassVar, Optional
from sqlalchemy import Column
from sqlalchemy import BigInteger, Column, Float
from sqlalchemy.dialects.postgresql import JSONB
from sqlmodel import Field, SQLModel
class SolarBuildingInsightsRow(SQLModel, table=True):
"""Persisted Google Solar `buildingInsights` response for one Property.
class SolarRow(SQLModel, table=True):
"""Mirror of the live ``solar`` table (owned by the Drizzle schema): the raw
Google Solar ``buildingInsights`` response for one UPRN, stored whole as
JSONB so a future SolarPotential projection can be derived without
re-fetching. Keyed by ``uprn`` (the live table carries no ``property_id``);
``longitude``/``latitude`` are the coordinates the fetch was made against.
Stored as JSONB — the raw fetched insights are retained whole so the
structured projection a future SolarPotential type needs can be derived
without re-fetching. One row per Property.
Only the columns this repo reads/writes are mirrored — ``created_at`` /
``updated_at`` are left to the database's ``DEFAULT now()``.
"""
__tablename__: ClassVar[str] = "solar_building_insights" # pyright: ignore[reportIncompatibleVariableOverride]
__tablename__: ClassVar[str] = "solar" # pyright: ignore[reportIncompatibleVariableOverride]
id: Optional[int] = Field(default=None, primary_key=True)
property_id: int = Field(index=True, unique=True)
insights: dict[str, Any] = Field(sa_column=Column(JSONB, nullable=False))
id: Optional[int] = Field(
default=None, sa_column=Column(BigInteger, primary_key=True)
)
uprn: int = Field(sa_column=Column(BigInteger, nullable=False, index=True))
longitude: float = Field(sa_column=Column(Float, nullable=False))
latitude: float = Field(sa_column=Column(Float, nullable=False))
google_api_response: dict[str, Any] = Field(
sa_column=Column(JSONB, nullable=False)
)

16
orchestration/ingestion_orchestrator.py
View file

@ -99,8 +99,20 @@ class IngestionOrchestrator:
for item in fetched:
if item.epc is not None:
uow.epc.save(item.epc, property_id=item.property_id)
if item.solar_insights is not None:
uow.solar.save(item.property_id, item.solar_insights)
# The live `solar` table is keyed by UPRN and needs the fetch's
# coordinates; insights are only set when those coordinates were
# resolved, so spatial.coordinates is non-None alongside them.
if (
item.solar_insights is not None
and item.spatial is not None
and item.spatial.coordinates is not None
):
uow.solar.save(
item.uprn,
longitude=item.spatial.coordinates.longitude,
latitude=item.spatial.coordinates.latitude,
insights=item.solar_insights,
)
if item.spatial is not None:
uow.spatial.save(item.uprn, item.spatial)
uow.commit()

15
orchestration/modelling_orchestrator.py
View file

@ -118,7 +118,7 @@ class ModellingOrchestrator:
# threaded into the solar Generator (ADR-0026). None when no
# solar data was fetched — the Generator then offers nothing.
solar_potential: Optional[SolarPotential] = _solar_potential_for(
uow.solar, property_id
uow.solar, prop.identity.uprn
)
for scenario in scenarios:
plan = self._plan_for(
@ -225,12 +225,15 @@ def _bill_for(bill_derivation: BillDerivation, score: Score) -> Bill:
def _solar_potential_for(
solar_repo: SolarRepository, property_id: int
solar_repo: SolarRepository, uprn: Optional[int]
) -> Optional[SolarPotential]:
"""Project the Property's persisted Google Solar `buildingInsights` JSON
into a typed `SolarPotential` (ADR-0026), or None when none was fetched /
the lookup returned an error payload (no `solarPotential` block)."""
insights = solar_repo.get(property_id)
"""Project the UPRN's persisted Google Solar `buildingInsights` JSON
into a typed `SolarPotential` (ADR-0026), or None when there is no UPRN /
none was fetched / the lookup returned an error payload (no `solarPotential`
block). Solar is keyed by UPRN to match the live ``solar`` table."""
if uprn is None:
return None
insights = solar_repo.get(uprn)
if not insights or "solarPotential" not in insights:
return None
return SolarPotential.from_building_insights(insights)

20
repositories/product/product_postgres_repository.py
View file

@ -8,6 +8,21 @@ from infrastructure.postgres.product_table import MaterialRow
from repositories.product.product_repository import ProductRepository
# The domain ``MeasureType`` vocabulary and the catalogue's ``material.type``
# pgEnum drifted apart: these five measures are spelled differently on the
# catalogue side (and querying the domain spelling raises a pgEnum DataError
# that poisons the session's transaction). Translate them to the catalogue's
# own vocabulary at this boundary so the domain enum stays stable. Every other
# MeasureType already matches its material.type and maps to itself.
_MATERIAL_TYPE_BY_MEASURE: dict[str, str] = {
"low_energy_lighting": "low_energy_lighting_installation",
"gas_boiler_upgrade": "boiler_upgrade",
"system_tune_up": "roomstat_programmer_trvs",
"system_tune_up_zoned": "time_temperature_zone_control",
"sloping_ceiling_insulation": "room_roof_insulation",
}
class ProductPostgresRepository(ProductRepository):
"""Reads the ``material`` catalogue table and maps an active row to a
Product: `total_cost` becomes the fully-loaded `unit_cost_per_m2`, and the
@ -17,13 +32,16 @@ class ProductPostgresRepository(ProductRepository):
self._session = session
def get(self, measure_type: str) -> Product:
# Resolve the domain MeasureType to the catalogue's ``material.type``
# spelling (identity for all but the five drifted types above).
catalogue_type = _MATERIAL_TYPE_BY_MEASURE.get(measure_type, measure_type)
# The live catalogue holds many active rows per type; order by id so the
# pick is deterministic (a re-seed prices the same) rather than relying
# on the database's physical row order.
row: MaterialRow | None = self._session.exec(
select(MaterialRow)
.where(
col(MaterialRow.type) == measure_type,
col(MaterialRow.type) == catalogue_type,
col(MaterialRow.is_active).is_(True),
)
.order_by(col(MaterialRow.id))

29
repositories/solar/solar_postgres_repository.py
View file

@ -4,7 +4,7 @@ from typing import Any, Optional
from sqlmodel import Session, select
from infrastructure.postgres.solar_table import SolarBuildingInsightsRow
from infrastructure.postgres.solar_table import SolarRow
from repositories.solar.solar_repository import SolarRepository
@ -12,24 +12,29 @@ class SolarPostgresRepository(SolarRepository):
def __init__(self, session: Session) -> None:
self._session = session
def save(self, property_id: int, insights: dict[str, Any]) -> None:
def save(
self, uprn: int, *, longitude: float, latitude: float, insights: dict[str, Any]
) -> None:
existing = self._session.exec(
select(SolarBuildingInsightsRow).where(
SolarBuildingInsightsRow.property_id == property_id
)
select(SolarRow).where(SolarRow.uprn == uprn)
).first()
if existing is None:
self._session.add(
SolarBuildingInsightsRow(property_id=property_id, insights=insights)
SolarRow(
uprn=uprn,
longitude=longitude,
latitude=latitude,
google_api_response=insights,
)
)
else:
existing.insights = insights
existing.longitude = longitude
existing.latitude = latitude
existing.google_api_response = insights
self._session.add(existing)
def get(self, property_id: int) -> Optional[dict[str, Any]]:
def get(self, uprn: int) -> Optional[dict[str, Any]]:
row = self._session.exec(
select(SolarBuildingInsightsRow).where(
SolarBuildingInsightsRow.property_id == property_id
)
select(SolarRow).where(SolarRow.uprn == uprn)
).first()
return row.insights if row is not None else None
return row.google_api_response if row is not None else None

12
repositories/solar/solar_repository.py
View file

@ -5,15 +5,19 @@ from typing import Any, Optional
class SolarRepository(ABC):
"""Persists and loads a Property's Google Solar building insights.
"""Persists and loads a UPRN's Google Solar building insights.
Thin save/get over the raw fetched insights (a future SolarPotential domain
type will derive its fields from these). Written by Ingestion, read by
Baseline/Modelling — never re-fetched downstream (ADR-0003).
Baseline/Modelling — never re-fetched downstream (ADR-0003). Keyed by
``uprn`` to match the live ``solar`` table; ``longitude``/``latitude`` are
the coordinates the fetch was made against (NOT NULL on the live table).
"""
@abstractmethod
def save(self, property_id: int, insights: dict[str, Any]) -> None: ...
def save(
self, uprn: int, *, longitude: float, latitude: float, insights: dict[str, Any]
) -> None: ...
@abstractmethod
def get(self, property_id: int) -> Optional[dict[str, Any]]: ...
def get(self, uprn: int) -> Optional[dict[str, Any]]: ...

20
scripts/eon/find_epc_data.py
View file

@ -17,17 +17,17 @@ from infrastructure.epc_client.epc_client_service import EpcClientService
# Reduced-Field Synthesis mapper (ADR-0027) re-maps so the SAP10 calculator can
# re-score them. The commented rows are non-20.0.0 neighbours kept for context.
UPRNS: list[int] = [
10003318624, # 20.0.0 Flat 1, 6 Alexandra Gardens, PO38 1EE
10003318625, # 20.0.0 Flat 2, 6 Alexandra Gardens, PO38 1EE
10003318626, # 20.0.0 Flat 3, 6 Alexandra Gardens, PO38 1EE
# 10003318698, # 17.1 Flat 4, 6 Alexandra Gardens, PO38 1EE
100062430247, # 20.0.0 Flat 5, Adelaide Court, Adelaide Place, PO33 3DG
100062430248, # 20.0.0 Flat 6, Adelaide Court, Adelaide Place, PO33 3DG
100062430250, # 20.0.0 Flat 8, Adelaide Court, Adelaide Place, PO33 3DG
100062429797, # 20.0.0 Flat 1, 10-11 Cross Street, PO33 2AD
10003320577, # 20.0.0 Flat 3, 10-11 Cross Street, PO33 2AD
# 10003318624, # 20.0.0 Flat 1, 6 Alexandra Gardens, PO38 1EE
# 10003318625, # 20.0.0 Flat 2, 6 Alexandra Gardens, PO38 1EE
# 10003318626, # 20.0.0 Flat 3, 6 Alexandra Gardens, PO38 1EE
10003318698, # 17.1 Flat 4, 6 Alexandra Gardens, PO38 1EE
# 100062430247, # 20.0.0 Flat 5, Adelaide Court, Adelaide Place, PO33 3DG
# 100062430248, # 20.0.0 Flat 6, Adelaide Court, Adelaide Place, PO33 3DG
# 100062430250, # 20.0.0 Flat 8, Adelaide Court, Adelaide Place, PO33 3DG
# 100062429797, # 20.0.0 Flat 1, 10-11 Cross Street, PO33 2AD
# 10003320577, # 20.0.0 Flat 3, 10-11 Cross Street, PO33 2AD
# 10003320573, # 18.0 Flat 7, 10-11 Cross Street, PO33 2AD
10024248769, # 20.0.0 Flat 8, 10-11 Cross Street, PO33 2AD
# 10024248769, # 20.0.0 Flat 8, 10-11 Cross Street, PO33 2AD
# 10024248772, # 18.0 Flat 9, 10-11 Cross Street, PO33 2AD
]

16
scripts/run_modelling_e2e.py
View file

@ -240,8 +240,17 @@ def _persist(
uow.epc.save(epc, property_id=property_id, portfolio_id=portfolio_id)
if spatial is not None:
uow.spatial.save(uprn, spatial)
# The live `solar` table is keyed by UPRN and needs the fetch's
# coordinates; insights are only present when those coordinates were
# (see `_solar_insights_for`), so `spatial.coordinates` is non-None here.
if solar_insights is not None:
uow.solar.save(property_id, solar_insights)
assert spatial is not None and spatial.coordinates is not None
uow.solar.save(
uprn,
longitude=spatial.coordinates.longitude,
latitude=spatial.coordinates.latitude,
insights=solar_insights,
)
uow.plan.save(
plan,
property_id=property_id,
@ -368,6 +377,11 @@ def main() -> None:
except (
Exception
) as error: # noqa: BLE001 — one bad property must not stop the run
# A failed catalogue query (e.g. a `material.type` enum mismatch)
# aborts the shared session's transaction; without a rollback every
# subsequent property reports `InFailedSqlTransaction` and masks its
# own real error. Reset so each property surfaces what's wrong.
catalogue_session.rollback()
line = f"property {property_id} (uprn {uprn}): ERROR — {type(error).__name__}: {error}"
print(line + "\n")
md_lines.append(f"## Property {property_id}\n\n`{line}`\n")

172
sfr/principal_pitch/2_export_data.py
View file

@ -2,9 +2,19 @@
This script prepares the data for the financial model
"""
import os
from datetime import date, datetime
from pathlib import Path
from typing import Any, Optional
from dotenv import load_dotenv
load_dotenv(".env.local")
# The retired `property_details_epc` table is no longer populated under the new
# backend, so the EPC descriptive fields are sourced live from the EPC service
# instead (which needs OPEN_EPC_API_TOKEN — also lives in backend/.env).
_REPO_ROOT = Path(__file__).resolve().parents[2]
load_dotenv(_REPO_ROOT / "backend" / ".env")
import pandas as pd
import numpy as np
@ -14,25 +24,97 @@ from backend.app.db.connection import db_engine, db_read_session
from backend.app.db.models.recommendations import (
Recommendation,
PlanModel,
PlanRecommendations,
RecommendationMaterials,
)
from backend.app.db.models.portfolio import (
PropertyModel,
PropertyDetailsEpcModel,
PropertyDetailsSpatial,
)
from backend.app.db.functions.materials_functions import get_materials
from infrastructure.epc_client.epc_client_service import EpcClientService
from collections import defaultdict
from sqlalchemy import func
PORTFOLIO_ID = 632
SCENARIOS = [1144]
def _description_text(item: Any) -> str:
"""Display text for one raw-cert EPC feature. Handles both schema shapes:
20.0.0 stores ``description`` as a plain string; 17.1 wraps it as a
``{"value": ..., "language": ...}`` LanguageString."""
if not isinstance(item, dict):
return ""
desc = item.get("description")
if isinstance(desc, dict):
desc = desc.get("value")
return str(desc or "")
def _join_descriptions(value: Any) -> str:
"""Flatten a raw-cert EPC feature into a display string. The new EPC API
returns these as a list of feature dicts (walls/roofs/floors/main_heating),
a single feature dict (hot_water/window/lighting), or null."""
if isinstance(value, list):
return "; ".join(t for t in (_description_text(d) for d in value) if t)
return _description_text(value)
def _is_expired(registration_date: Optional[str]) -> Optional[bool]:
"""An EPC is valid for 10 years from its lodgement (registration) date."""
if not registration_date:
return None
try:
lodged = datetime.fromisoformat(registration_date[:10]).date()
except ValueError:
return None
return (date.today() - lodged).days > 365 * 10
def epc_details_from_service(svc: EpcClientService, uprn: Optional[int]) -> dict[str, Any]:
"""Mock the retired ``property_details_epc`` row from the live EPC service:
fetch the UPRN's latest raw certificate and flatten the descriptive fields
the export needs. Returns ``{}`` when the UPRN has no EPC (the property then
carries blank EPC columns rather than being dropped)."""
if uprn is None:
return {}
results = svc._search(uprn=uprn) # pyright: ignore[reportPrivateUsage]
if not results:
return {}
latest = max(results, key=lambda r: r.registration_date)
raw = svc._fetch_certificate(latest.certificate_number) # pyright: ignore[reportPrivateUsage]
def _to_int(value: Any) -> Optional[int]:
try:
return int(value)
except (TypeError, ValueError):
return None
current_sap = _to_int(raw.get("energy_rating_current"))
return {
"walls": _join_descriptions(raw.get("walls")),
"roof": _join_descriptions(raw.get("roofs")),
"floor": _join_descriptions(raw.get("floors")),
"windows": _join_descriptions(raw.get("window")),
"heating": _join_descriptions(raw.get("main_heating")),
"heating_controls": _join_descriptions(raw.get("main_heating_controls")),
"hot_water": _join_descriptions(raw.get("hot_water")),
"lighting": _join_descriptions(raw.get("lighting")),
"total_floor_area": raw.get("total_floor_area"),
"lodgement_date": raw.get("registration_date"),
"is_expired": _is_expired(raw.get("registration_date")),
# Baseline SAP/band/postcode aren't on the new `property` table, so take
# the lodged figures off the cert (the assessment re-scores from these).
"postcode": raw.get("postcode"),
"current_epc_rating": raw.get("current_energy_efficiency_band"),
"current_sap_points": current_sap,
"original_sap_points": current_sap,
}
PORTFOLIO_ID = 785
SCENARIOS = [1266]
scenario_names = {
1144: "EPC C",
1266: "EPC C",
}
project_name = "Calico Refresh"
project_name = "Small request for EON"
def get_data(portfolio_id, scenario_ids):
@ -42,29 +124,29 @@ def get_data(portfolio_id, scenario_ids):
# --------------------
# Properties
# --------------------
# `property_details_epc` is dead under the new backend, so read the base
# Property rows and source the EPC descriptive fields live from the EPC
# service (one cert fetch per property).
properties_query = (
session.query(PropertyModel, PropertyDetailsEpcModel)
.join(
PropertyDetailsEpcModel,
PropertyModel.id == PropertyDetailsEpcModel.property_id,
)
session.query(PropertyModel)
.filter(PropertyModel.portfolio_id == portfolio_id)
.all()
)
properties_data = [
{
**{
col.name: getattr(p.PropertyModel, col.name)
for col in PropertyModel.__table__.columns
},
**{
col.name: getattr(p.PropertyDetailsEpcModel, col.name)
for col in PropertyDetailsEpcModel.__table__.columns
},
}
for p in properties_query
]
epc_service = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"])
properties_data = []
for p in properties_query:
base = {col.name: getattr(p, col.name) for col in PropertyModel.__table__.columns}
# `property_id` is the key the recommendations merge joins on; the
# Property's own PK is its `id`.
base["property_id"] = p.id
# Fill EPC fields from the service; for columns that also exist on the
# Property row (postcode, SAP points, rating), only fill when the row's
# value is missing so genuine Property data is never clobbered.
for key, value in epc_details_from_service(epc_service, p.uprn).items():
if base.get(key) is None:
base[key] = value
properties_data.append(base)
# --------------------
# Plans
@ -115,17 +197,13 @@ def get_data(portfolio_id, scenario_ids):
# --------------------
# Recommendations (NO materials yet)
# --------------------
# The `plan_recommendations` m2m is retired (ADR-0017): a Recommendation
# links to its Plan directly via `recommendation.plan_id`.
recommendations_query = (
session.query(
Recommendation, PlanModel.scenario_id, PlanRecommendations.plan_id
)
.join(
PlanRecommendations,
Recommendation.id == PlanRecommendations.recommendation_id,
)
.join(PlanModel, PlanModel.id == PlanRecommendations.plan_id)
session.query(Recommendation, PlanModel.scenario_id)
.join(PlanModel, PlanModel.id == Recommendation.plan_id)
.filter(
PlanRecommendations.plan_id.in_(plan_ids),
Recommendation.plan_id.in_(plan_ids),
Recommendation.default.is_(True),
Recommendation.already_installed.is_(False),
)
@ -135,7 +213,7 @@ def get_data(portfolio_id, scenario_ids):
recommendations_data = [
{
**{
col.name: getattr(r.Recommendation, col.name)
col.name: getattr(r[0], col.name)
for col in Recommendation.__table__.columns
},
"scenario_id": r.scenario_id,
@ -328,7 +406,33 @@ for scenario_id in SCENARIOS:
if col not in df.columns:
df[col] = ""
# A per-recommendation detail sheet (one row per recommended measure) so the
# measures and their costs are readable directly, not just pivoted into the
# wide `properties` sheet.
recs_detail = recommendations_df[
recommendations_df["scenario_id"] == scenario_id
].copy()
recs_detail = recs_detail[recs_detail["default"]]
detail_cols = [
c
for c in [
"property_id",
"measure_type",
"description",
"estimated_cost",
"sap_points",
"co2_equivalent_savings",
"kwh_savings",
"energy_cost_savings",
]
if c in recs_detail.columns
]
recs_detail = recs_detail[detail_cols].sort_values(
["property_id", "estimated_cost"], ascending=[True, False]
)
# Create excel to store to
filename = f"{scenario_names[scenario_id]} - {project_name}.xlsx"
with pd.ExcelWriter(filename) as writer:
df.to_excel(writer, sheet_name="properties", index=False)
recs_detail.to_excel(writer, sheet_name="recommendations", index=False)

29
tests/orchestration/fakes.py
View file

@ -105,23 +105,24 @@ class FakeEpcRepo(EpcRepository):
class FakeSolarRepo(SolarRepository):
"""In-memory Google Solar insights store. Seed `by_property` to hydrate a
Property's potential for a Modelling read; `saved` records writes for an
Ingestion-side assertion. Returns None for an unseeded Property (no solar
data fetched) — the solar Generator then offers nothing."""
"""In-memory Google Solar insights store keyed by UPRN. Seed `by_uprn` to
hydrate a UPRN's potential for a Modelling read; `saved` records writes
(uprn, longitude, latitude, insights) for an Ingestion-side assertion.
Returns None for an unseeded UPRN (no solar data fetched) — the solar
Generator then offers nothing."""
def __init__(
self, by_property: Optional[dict[int, dict[str, Any]]] = None
def __init__(self, by_uprn: Optional[dict[int, dict[str, Any]]] = None) -> None:
self.saved: list[tuple[int, float, float, dict[str, Any]]] = []
self._by_uprn: dict[int, dict[str, Any]] = dict(by_uprn or {})
def save(
self, uprn: int, *, longitude: float, latitude: float, insights: dict[str, Any]
) -> None:
self.saved: list[tuple[int, dict[str, Any]]] = []
self._by_property: dict[int, dict[str, Any]] = dict(by_property or {})
self.saved.append((uprn, longitude, latitude, insights))
self._by_uprn[uprn] = insights
def save(self, property_id: int, insights: dict[str, Any]) -> None:
self.saved.append((property_id, insights))
self._by_property[property_id] = insights
def get(self, property_id: int) -> Optional[dict[str, Any]]:
return self._by_property.get(property_id)
def get(self, uprn: int) -> Optional[dict[str, Any]]:
return self._by_uprn.get(uprn)
class FakeSpatialRepo(SpatialRepository):

12
tests/orchestration/test_ara_first_run_pipeline_integration.py
View file

@ -162,7 +162,7 @@ def test_first_run_baselines_through_repos_and_is_idempotent_on_rerun(
),
MaterialRow(
id=4,
type="low_energy_lighting",
type="low_energy_lighting_installation",
total_cost=8.0,
cost_unit="gbp_per_unit",
is_active=True,
@ -170,7 +170,7 @@ def test_first_run_baselines_through_repos_and_is_idempotent_on_rerun(
),
MaterialRow(
id=6,
type="gas_boiler_upgrade",
type="boiler_upgrade",
total_cost=3000.0,
cost_unit="gbp_per_unit",
is_active=True,
@ -178,7 +178,7 @@ def test_first_run_baselines_through_repos_and_is_idempotent_on_rerun(
),
MaterialRow(
id=7,
type="system_tune_up",
type="roomstat_programmer_trvs",
total_cost=500.0,
cost_unit="gbp_per_unit",
is_active=True,
@ -186,7 +186,7 @@ def test_first_run_baselines_through_repos_and_is_idempotent_on_rerun(
),
MaterialRow(
id=8,
type="system_tune_up_zoned",
type="time_temperature_zone_control",
total_cost=900.0,
cost_unit="gbp_per_unit",
is_active=True,
@ -310,7 +310,7 @@ def test_modelling_optimises_and_persists_a_multi_measure_plan(
),
MaterialRow(
id=4,
type="low_energy_lighting",
type="low_energy_lighting_installation",
total_cost=8.0,
cost_unit="gbp_per_unit",
is_active=True,
@ -485,7 +485,7 @@ def test_modelling_recommends_nothing_when_already_at_the_target_band(
),
MaterialRow(
id=13,
type="low_energy_lighting",
type="low_energy_lighting_installation",
total_cost=8.0,
cost_unit="gbp_per_unit",
is_active=True,

2
tests/orchestration/test_ingestion_orchestrator.py
View file

@ -98,7 +98,7 @@ def test_ingestion_persists_epc_and_threads_coords_into_solar() -> None:
# fetcher, solar persisted, batch committed once.
assert epc_repo.saved == [(epc, 10)]
assert solar_fetcher.calls == [(-0.1278, 51.5074)]
assert solar_repo.saved == [(10, insights)]
assert solar_repo.saved == [(12345, -0.1278, 51.5074, insights)]
assert uow.commits == 1

14
tests/orchestration/test_modelling_solar_threading.py
View file

@ -62,24 +62,24 @@ def test_solar_potential_for_returns_none_when_no_insights() -> None:
solar = FakeSolarRepo()
# Act / Assert
assert _solar_potential_for(solar, property_id=42) is None
assert _solar_potential_for(solar, uprn=42) is None
def test_solar_potential_for_returns_none_for_an_error_payload() -> None:
# Arrange — the Solar API found no building; Ingestion persisted the error
# dict (no `solarPotential` block).
solar = FakeSolarRepo(by_property={7: {"error": "ENTITY_NOT_FOUND"}})
solar = FakeSolarRepo(by_uprn={7: {"error": "ENTITY_NOT_FOUND"}})
# Act / Assert
assert _solar_potential_for(solar, property_id=7) is None
assert _solar_potential_for(solar, uprn=7) is None
def test_solar_potential_for_projects_valid_insights() -> None:
# Arrange
solar = FakeSolarRepo(by_property={7: _insights()})
solar = FakeSolarRepo(by_uprn={7: _insights()})
# Act
potential = _solar_potential_for(solar, property_id=7)
potential = _solar_potential_for(solar, uprn=7)
# Assert — the real London example projects to the 46-rung ladder.
assert potential is not None
@ -91,7 +91,7 @@ def test_candidate_recommendations_includes_solar_when_potential_present() -> No
# Arrange — a solar-eligible house with a feasible potential.
epc = _eligible_house()
potential = _solar_potential_for(
FakeSolarRepo(by_property={1: _insights()}), property_id=1
FakeSolarRepo(by_uprn={1: _insights()}), uprn=1
)
# Act
@ -121,7 +121,7 @@ def test_considered_measures_restricts_candidates_to_the_allowlist() -> None:
# unrestricted run offers Solar PV alongside any fabric/heating candidates.
epc = _eligible_house()
potential = _solar_potential_for(
FakeSolarRepo(by_property={1: json.loads(_INSIGHTS_FIXTURE.read_text())}), 1
FakeSolarRepo(by_uprn={1: json.loads(_INSIGHTS_FIXTURE.read_text())}), 1
)
# Act — restrict the run to Solar PV only.

32
tests/repositories/product/test_product_postgres_repository.py
View file

@ -42,6 +42,38 @@ def test_get_maps_active_material_to_product_with_contingency(
assert abs(product.contingency_rate - 0.10) <= 1e-9
def test_get_resolves_a_drifted_measure_type_to_the_catalogue_spelling(
db_engine: Engine,
) -> None:
# Arrange — the domain MeasureType `low_energy_lighting` is spelled
# `low_energy_lighting_installation` on the live `material.type` enum; the
# repo must translate it so the lookup finds the catalogue row. The returned
# Product keeps the *domain* measure type (what the Plan/Recommendation
# store), not the catalogue spelling.
with Session(db_engine) as session:
session.add(
MaterialRow(
id=1,
type="low_energy_lighting_installation",
total_cost=5.0,
cost_unit="gbp_per_unit",
is_active=True,
description="Low energy lighting",
)
)
session.commit()
# Act
with Session(db_engine) as session:
product: Product = ProductPostgresRepository(session).get(
"low_energy_lighting"
)
# Assert
assert product.measure_type == "low_energy_lighting"
assert abs(product.unit_cost_per_m2 - 5.0) <= 1e-9
def test_get_picks_the_lowest_id_when_several_active_rows_share_a_type(
db_engine: Engine,
) -> None:

4
tests/repositories/solar/test_solar_repository.py
View file

@ -23,7 +23,9 @@ def test_building_insights_round_trip(db_engine: Engine) -> None:
# Act
with Session(db_engine) as session:
SolarPostgresRepository(session).save(property_id=5, insights=insights)
SolarPostgresRepository(session).save(
5, longitude=-0.1, latitude=51.5, insights=insights
)
session.commit()
with Session(db_engine) as session:
reloaded = SolarPostgresRepository(session).get(5)