From d05e5bd1f3670c287911b5dc6f0408aec27b541c Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 12:54:53 +0000 Subject: [PATCH 01/14] new application to trigger e2e for a single property and scenario --- .../terraform/lambda/modelling_e2e/main.tf | 50 +++++ .../terraform/lambda/modelling_e2e/outputs.tf | 9 + .../lambda/modelling_e2e/provider.tf | 20 ++ .../lambda/modelling_e2e/variables.tf | 59 ++++++ deployment/terraform/shared/main.tf | 32 ++++ domain/tasks/tasks.py | 1 + lambdas/__init__.py | 0 lambdas/modelling_e2e/Dockerfile | 32 ++++ lambdas/modelling_e2e/__init__.py | 0 lambdas/modelling_e2e/handler.py | 179 ++++++++++++++++++ lambdas/modelling_e2e/requirements.txt | 8 + scripts/trigger_modelling_e2e_sqs.py | 103 ++++++++++ tests/lambdas/__init__.py | 0 tests/lambdas/modelling_e2e/__init__.py | 0 14 files changed, 493 insertions(+) create mode 100644 deployment/terraform/lambda/modelling_e2e/main.tf create mode 100644 deployment/terraform/lambda/modelling_e2e/outputs.tf create mode 100644 deployment/terraform/lambda/modelling_e2e/provider.tf create mode 100644 deployment/terraform/lambda/modelling_e2e/variables.tf create mode 100644 lambdas/__init__.py create mode 100644 lambdas/modelling_e2e/Dockerfile create mode 100644 lambdas/modelling_e2e/__init__.py create mode 100644 lambdas/modelling_e2e/handler.py create mode 100644 lambdas/modelling_e2e/requirements.txt create mode 100644 scripts/trigger_modelling_e2e_sqs.py create mode 100644 tests/lambdas/__init__.py create mode 100644 tests/lambdas/modelling_e2e/__init__.py diff --git a/deployment/terraform/lambda/modelling_e2e/main.tf b/deployment/terraform/lambda/modelling_e2e/main.tf new file mode 100644 index 00000000..31f90e1f --- /dev/null +++ b/deployment/terraform/lambda/modelling_e2e/main.tf @@ -0,0 +1,50 @@ +data "terraform_remote_state" "shared" { + backend = "s3" + config = { + bucket = "assessment-model-terraform-state" + key = "env:/${var.stage}/terraform.tfstate" + region = "eu-west-2" + } +} + +data "aws_secretsmanager_secret_version" "db_credentials" { + secret_id = "${var.stage}/assessment_model/db_credentials" +} + +locals { + db_credentials = jsondecode(data.aws_secretsmanager_secret_version.db_credentials.secret_string) +} + +module "lambda" { + source = "../../modules/lambda_with_sqs" + + name = var.lambda_name + stage = var.stage + + image_uri = local.image_uri + + reserved_concurrent_executions = var.reserved_concurrent_executions + + batch_size = var.batch_size + + timeout = 60 + memory_size = 1024 + + environment = { + STAGE = var.stage + LOG_LEVEL = "info" + POSTGRES_USERNAME = local.db_credentials.db_assessment_model_username + POSTGRES_PASSWORD = local.db_credentials.db_assessment_model_password + POSTGRES_HOST = var.db_host + POSTGRES_DATABASE = var.db_name + POSTGRES_PORT = var.db_port + OPEN_EPC_API_TOKEN = var.open_epc_api_token + GOOGLE_SOLAR_API_KEY = var.google_solar_api_key + DATA_BUCKET = "retrofit-data-${var.stage}" + } +} + +resource "aws_iam_role_policy_attachment" "modelling_e2e_s3_read" { + role = module.lambda.role_name + policy_arn = data.terraform_remote_state.shared.outputs.modelling_e2e_s3_read_arn +} diff --git a/deployment/terraform/lambda/modelling_e2e/outputs.tf b/deployment/terraform/lambda/modelling_e2e/outputs.tf new file mode 100644 index 00000000..34cf4cef --- /dev/null +++ b/deployment/terraform/lambda/modelling_e2e/outputs.tf @@ -0,0 +1,9 @@ +output "modelling_e2e_queue_url" { + value = module.lambda.queue_url + description = "URL of the modelling-e2e SQS queue (pass to trigger_modelling_e2e_sqs.py --sqs-url)" +} + +output "modelling_e2e_queue_arn" { + value = module.lambda.queue_arn + description = "ARN of the modelling-e2e SQS queue" +} diff --git a/deployment/terraform/lambda/modelling_e2e/provider.tf b/deployment/terraform/lambda/modelling_e2e/provider.tf new file mode 100644 index 00000000..7100be24 --- /dev/null +++ b/deployment/terraform/lambda/modelling_e2e/provider.tf @@ -0,0 +1,20 @@ +terraform { + required_providers { + aws = { + source = "hashicorp/aws" + version = ">= 5.0" + } + } + + backend "s3" { + bucket = "modelling-e2e-terraform-state" + key = "terraform.tfstate" + region = "eu-west-2" + } + + required_version = ">= 1.2.0" +} + +provider "aws" { + region = "eu-west-2" +} diff --git a/deployment/terraform/lambda/modelling_e2e/variables.tf b/deployment/terraform/lambda/modelling_e2e/variables.tf new file mode 100644 index 00000000..7fd1ecde --- /dev/null +++ b/deployment/terraform/lambda/modelling_e2e/variables.tf @@ -0,0 +1,59 @@ +variable "lambda_name" { + type = string + description = "Logical name of the lambda" +} + +variable "stage" { + description = "Deployment stage (e.g. dev, prod)" + type = string +} + +variable "ecr_repo_url" { + type = string + description = "ECR repository URL (no tag, no digest)" +} + +variable "image_digest" { + type = string + description = "Image digest (sha256:...)" +} + +variable "reserved_concurrent_executions" { + type = number + default = 1 + description = "Start at 1 to validate correctness before scaling up." +} + +variable "batch_size" { + type = number + default = 1 +} + +variable "db_host" { + type = string + sensitive = true +} + +variable "db_name" { + type = string + sensitive = true +} + +variable "db_port" { + type = string + sensitive = true +} + +variable "open_epc_api_token" { + type = string + sensitive = true +} + +variable "google_solar_api_key" { + type = string + sensitive = true +} + +locals { + image_uri = "${var.ecr_repo_url}@${var.image_digest}" +} diff --git a/deployment/terraform/shared/main.tf b/deployment/terraform/shared/main.tf index 3d6bbd39..bca65bb3 100644 --- a/deployment/terraform/shared/main.tf +++ b/deployment/terraform/shared/main.tf @@ -858,3 +858,35 @@ module "sharepoint_renamer_registry" { stage = var.stage } +################################################ +# Modelling E2E – Lambda +################################################ +module "modelling_e2e_state_bucket" { + source = "../modules/tf_state_bucket" + bucket_name = "modelling-e2e-terraform-state" +} + +module "modelling_e2e_registry" { + source = "../modules/container_registry" + name = "modelling-e2e" + stage = var.stage +} + +module "modelling_e2e_s3_read" { + source = "../modules/s3_iam_policy" + + policy_name = "ModellingE2EReadS3" + policy_description = "Allow modelling-e2e Lambda to read spatial parquet from the data bucket" + bucket_arns = ["arn:aws:s3:::retrofit-data-${var.stage}"] + actions = ["s3:GetObject", "s3:ListBucket"] + resource_paths = ["/*"] +} + +output "modelling_e2e_s3_read_arn" { + value = module.modelling_e2e_s3_read.policy_arn +} + +output "modelling_e2e_ecr_url" { + value = module.modelling_e2e_registry.repository_url +} + diff --git a/domain/tasks/tasks.py b/domain/tasks/tasks.py index 177258d6..ead253ab 100644 --- a/domain/tasks/tasks.py +++ b/domain/tasks/tasks.py @@ -17,6 +17,7 @@ class TaskStatus(str, Enum): class Source(str, Enum): PORTFOLIO = "portfolio_id" HUBSPOT_DEAL = "hubspot_deal_id" + PROPERTY = "property_id" @dataclass diff --git a/lambdas/__init__.py b/lambdas/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/lambdas/modelling_e2e/Dockerfile b/lambdas/modelling_e2e/Dockerfile new file mode 100644 index 00000000..c577bdf0 --- /dev/null +++ b/lambdas/modelling_e2e/Dockerfile @@ -0,0 +1,32 @@ +FROM public.ecr.aws/lambda/python:3.11 + +ARG DEV_DB_HOST +ARG DEV_DB_PORT +ARG DEV_DB_NAME + +ENV POSTGRES_HOST=${DEV_DB_HOST} +ENV POSTGRES_PORT=${DEV_DB_PORT} +ENV POSTGRES_DATABASE=${DEV_DB_NAME} + +WORKDIR /var/task + +COPY lambdas/modelling_e2e/requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY datatypes/ datatypes/ +COPY domain/ domain/ +COPY infrastructure/ infrastructure/ +COPY orchestration/ orchestration/ +COPY repositories/ repositories/ +COPY utilities/ utilities/ +COPY harness/ harness/ + +# harness/console.py imports in-memory fakes from tests/orchestration/ at module +# load time; the fakes have no pytest dependency and are safe to ship. +COPY tests/__init__.py tests/__init__.py +COPY tests/orchestration/__init__.py tests/orchestration/__init__.py +COPY tests/orchestration/fakes.py tests/orchestration/fakes.py + +COPY lambdas/ lambdas/ + +CMD ["lambdas.modelling_e2e.handler.handler"] diff --git a/lambdas/modelling_e2e/__init__.py b/lambdas/modelling_e2e/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/lambdas/modelling_e2e/handler.py b/lambdas/modelling_e2e/handler.py new file mode 100644 index 00000000..0fed0fb4 --- /dev/null +++ b/lambdas/modelling_e2e/handler.py @@ -0,0 +1,179 @@ +"""SQS-triggered Lambda: fetch EPC → run modelling → persist plan. + +One SQS message = one property. The handler reads ``property_id``, +``portfolio_id``, ``scenario_id``, and ``no_solar`` from the message body, +fetches the property's EPC from the gov API, runs the full modelling pipeline +(SAP10 → optimiser) via ``harness.console.run_modelling``, and persists the +resulting Plan via ``PlanPostgresRepository.save()``. + +``secondary_heating_removal`` is excluded unconditionally: the live ``material`` +catalogue does not yet carry this measure type, causing a crash during catalogue +reads for properties with a lodged secondary heater. + +DB engine is module-scoped so the connection pool is reused across warm +invocations (ADR-0012). +""" + +from __future__ import annotations + +import io +import os +from typing import Any, Optional, cast + +import boto3 +import pandas as pd # pyright: ignore[reportMissingTypeStubs] +from sqlalchemy import Engine, text +from sqlmodel import Session + +from datatypes.epc.domain.epc_property_data import EpcPropertyData +from domain.geospatial.planning_restrictions import PlanningRestrictions +from domain.geospatial.spatial_reference import SpatialReference +from domain.modelling.measure_type import MeasureType +from domain.property.property import Property, PropertyIdentity +from domain.tasks.tasks import Source +from harness.console import run_modelling +from infrastructure.epc_client.epc_client_service import EpcClientService +from infrastructure.postgres.config import PostgresConfig +from infrastructure.postgres.engine import make_engine +from infrastructure.solar.google_solar_api_client import ( + BuildingInsightsNotFoundError, + GoogleSolarApiClient, +) +from repositories.geospatial.geospatial_s3_repository import ( + GeospatialS3Repository, + ParquetReader, +) +from repositories.plan.plan_postgres_repository import PlanPostgresRepository +from repositories.product.product_postgres_repository import ProductPostgresRepository +from repositories.property.landlord_override_overlays import overlays_from +from repositories.property.property_overrides_postgres_reader import ( + PropertyOverridesPostgresReader, +) +from repositories.property.property_postgres_repository import ( + PropertyPostgresRepository, +) +from repositories.scenario.scenario_postgres_repository import ( + ScenarioPostgresRepository, +) +from utilities.aws_lambda.task_handler import task_handler + +_engine: Optional[Engine] = None + + +def _get_engine() -> Engine: + global _engine + if _engine is None: + _engine = make_engine(PostgresConfig.from_env(dict(os.environ))) + return _engine + + +def _s3_parquet_reader() -> ParquetReader: + bucket = os.environ["DATA_BUCKET"] + + def read(key: str) -> pd.DataFrame: + s3: Any = cast(Any, boto3.client("s3")) # pyright: ignore[reportUnknownMemberType] + raw = cast(bytes, s3.get_object(Bucket=bucket, Key=key)["Body"].read()) + return pd.read_parquet(io.BytesIO(raw)) # type: ignore[return-value] + + return read + + +def _spatial_for( + geospatial: GeospatialS3Repository, uprn: int +) -> Optional[SpatialReference]: + try: + return geospatial.spatial_for(uprn) + except Exception: # noqa: BLE001 + return None + + +def _solar_insights_for( + solar_client: GoogleSolarApiClient, spatial: Optional[SpatialReference] +) -> Optional[dict[str, Any]]: + if spatial is None or spatial.coordinates is None: + return None + try: + return solar_client.get_building_insights( + spatial.coordinates.longitude, spatial.coordinates.latitude + ) + except BuildingInsightsNotFoundError: + return None + + +@task_handler(task_source="modelling_e2e", source=Source.PROPERTY) +def handler(body: dict[str, Any], context: Any) -> None: + property_id = int(body["property_id"]) + portfolio_id = int(body["portfolio_id"]) + scenario_id = int(body["scenario_id"]) + no_solar = bool(body.get("no_solar", False)) + dry_run = bool(body.get("dry_run", False)) + + engine = _get_engine() + epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"]) + geospatial = GeospatialS3Repository(_s3_parquet_reader()) + solar_client = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"]) + + with engine.connect() as conn: + row = conn.execute( + text("SELECT uprn FROM property WHERE id = :pid"), + {"pid": property_id}, + ).one() + uprn = int(row[0]) + + epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn) + if epc is None: + raise ValueError(f"no EPC found for UPRN {uprn} (property {property_id})") + + overrides_reader = PropertyOverridesPostgresReader(lambda: Session(engine)) + overlaid = Property( + identity=PropertyIdentity( + portfolio_id=portfolio_id, postcode="", address="", uprn=uprn + ), + epc=epc, + landlord_overrides=overlays_from(overrides_reader.overrides_for(property_id)), + ) + effective_epc = overlaid.effective_epc + + spatial = _spatial_for(geospatial, uprn) + restrictions = spatial.restrictions if spatial is not None else PlanningRestrictions() + solar_insights = None if no_solar else _solar_insights_for(solar_client, spatial) + + with Session(engine) as session: + scenario = ScenarioPostgresRepository(session).get_many([scenario_id])[0] + products = ProductPostgresRepository(session) + + # secondary_heating_removal is absent from the live material.type enum; + # exclude it unconditionally until the catalogue gap is resolved. + considered: Optional[frozenset[MeasureType]] = ( + frozenset(MeasureType) - {MeasureType.SECONDARY_HEATING_REMOVAL} + ) + + plan = run_modelling( + effective_epc, + planning_restrictions=restrictions, + solar_insights=solar_insights, + considered_measures=considered, + products=products, + scenario=scenario, + print_table=False, + ) + + if dry_run: + measure_types = ", ".join(m.measure_type for m in plan.measures) or "none" + print( + f"[dry_run] property={property_id} scenario={scenario_id} " + f"SAP {plan.baseline.sap_continuous:.1f}→{plan.post_sap_continuous:.1f} " + f"measures=[{measure_types}] cost=£{plan.cost_of_works:,.0f}" + ) + return + PlanPostgresRepository(session).save( + plan, + property_id=property_id, + scenario_id=scenario_id, + portfolio_id=portfolio_id, + is_default=scenario.is_default, + ) + PropertyPostgresRepository(session).mark_modelled( + property_id, has_recommendations=bool(plan.measures) + ) + session.commit() diff --git a/lambdas/modelling_e2e/requirements.txt b/lambdas/modelling_e2e/requirements.txt new file mode 100644 index 00000000..51ee9050 --- /dev/null +++ b/lambdas/modelling_e2e/requirements.txt @@ -0,0 +1,8 @@ +awslambdaric +boto3 +pandas==2.2.2 +pyarrow +pydantic +sqlalchemy==2.0.36 +sqlmodel +psycopg2-binary==2.9.10 diff --git a/scripts/trigger_modelling_e2e_sqs.py b/scripts/trigger_modelling_e2e_sqs.py new file mode 100644 index 00000000..d220e799 --- /dev/null +++ b/scripts/trigger_modelling_e2e_sqs.py @@ -0,0 +1,103 @@ +"""Enqueue one SQS message per property for the modelling_e2e Lambda. + +Reads all property IDs for the given portfolio from the DB and sends a batch of +SQS messages, one per property. The Lambda then processes each message +independently, enabling concurrent modelling at scale. + +Edit the CONFIG block below, then run via VSCode Run button or Jupyter. +AWS creds come from the ambient ~/.aws profile; DB creds from backend/.env. +""" + +from __future__ import annotations + +# --------------------------------------------------------------------------- +# CONFIG — edit these before running +# --------------------------------------------------------------------------- +PORTFOLIO_ID: int = 785 +SCENARIO_ID: int = 1266 +SQS_URL: str = "https://sqs.eu-west-2.amazonaws.com/ACCOUNT_ID/modelling-e2e-STAGE" + +# Set to a positive integer to enqueue only the first N properties (trial run). +LIMIT: int | None = 10 + +# True → Lambda runs the full pipeline but skips all DB writes (safe for testing). +DRY_RUN: bool = True + +# True → Lambda skips the Google Solar fetch. +NO_SOLAR: bool = False +# --------------------------------------------------------------------------- + +import json +import sys +from pathlib import Path +from typing import Any, cast +from uuid import uuid4 + +_REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(_REPO_ROOT)) + +import boto3 # noqa: E402 +from sqlalchemy import text # noqa: E402 + +from scripts.e2e_common import ENV_PATH, build_engine, load_env # noqa: E402 + +_BATCH_SIZE = 10 + + +def _property_ids(portfolio_id: int, limit: int | None, engine: object) -> list[int]: + from sqlalchemy.engine import Engine + + assert isinstance(engine, Engine) + query = "SELECT id FROM property WHERE portfolio_id = :pid ORDER BY id" + if limit is not None: + query += f" LIMIT {int(limit)}" + with engine.connect() as conn: + rows = conn.execute(text(query), {"pid": portfolio_id}).fetchall() + return [int(r[0]) for r in rows] + + +def _batches(items: list[int], size: int) -> list[list[int]]: + return [items[i : i + size] for i in range(0, len(items), size)] + + +def main() -> None: + load_env(ENV_PATH) + engine = build_engine() + + ids = _property_ids(PORTFOLIO_ID, LIMIT, engine) + if not ids: + print(f"no properties found for portfolio {PORTFOLIO_ID}") + return + + print( + f"enqueuing {len(ids)} properties " + f"(portfolio={PORTFOLIO_ID}, scenario={SCENARIO_ID}, " + f"no_solar={NO_SOLAR}, dry_run={DRY_RUN}) → {SQS_URL}" + ) + + sqs: Any = cast(Any, boto3.client("sqs")) # pyright: ignore[reportUnknownMemberType] + sent = 0 + for batch in _batches(ids, _BATCH_SIZE): + entries = [ + { + "Id": str(uuid4()).replace("-", "")[:8] + str(i), + "MessageBody": json.dumps( + { + "property_id": pid, + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": NO_SOLAR, + "dry_run": DRY_RUN, + } + ), + } + for i, pid in enumerate(batch) + ] + sqs.send_message_batch(QueueUrl=SQS_URL, Entries=entries) + sent += len(batch) + print(f" sent {sent}/{len(ids)}", end="\r") + + print(f"\ndone — {sent} messages enqueued") + + +main() diff --git a/tests/lambdas/__init__.py b/tests/lambdas/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/lambdas/modelling_e2e/__init__.py b/tests/lambdas/modelling_e2e/__init__.py new file mode 100644 index 00000000..e69de29b From a54671e5695d672d0ed8823929d0e6058589b824 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 13:26:57 +0000 Subject: [PATCH 02/14] don't use ARG in dockerfile --- lambdas/modelling_e2e/Dockerfile | 8 -------- 1 file changed, 8 deletions(-) diff --git a/lambdas/modelling_e2e/Dockerfile b/lambdas/modelling_e2e/Dockerfile index c577bdf0..0f17e99a 100644 --- a/lambdas/modelling_e2e/Dockerfile +++ b/lambdas/modelling_e2e/Dockerfile @@ -1,13 +1,5 @@ FROM public.ecr.aws/lambda/python:3.11 -ARG DEV_DB_HOST -ARG DEV_DB_PORT -ARG DEV_DB_NAME - -ENV POSTGRES_HOST=${DEV_DB_HOST} -ENV POSTGRES_PORT=${DEV_DB_PORT} -ENV POSTGRES_DATABASE=${DEV_DB_NAME} - WORKDIR /var/task COPY lambdas/modelling_e2e/requirements.txt . From c45a27adb08efae0c79fbec5a2b1d45b73aaf1d0 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 13:35:23 +0000 Subject: [PATCH 03/14] move handler stuff to applications directory --- .../modelling_e2e/Dockerfile | 6 ++--- .../modelling_e2e/handler.py | 22 +++++++++++-------- .../modelling_e2e_trigger_body.py | 11 ++++++++++ .../modelling_e2e/requirements.txt | 0 lambdas/__init__.py | 0 lambdas/modelling_e2e/__init__.py | 0 6 files changed, 27 insertions(+), 12 deletions(-) rename {lambdas => applications}/modelling_e2e/Dockerfile (82%) rename {lambdas => applications}/modelling_e2e/handler.py (91%) create mode 100644 applications/modelling_e2e/modelling_e2e_trigger_body.py rename {lambdas => applications}/modelling_e2e/requirements.txt (100%) delete mode 100644 lambdas/__init__.py delete mode 100644 lambdas/modelling_e2e/__init__.py diff --git a/lambdas/modelling_e2e/Dockerfile b/applications/modelling_e2e/Dockerfile similarity index 82% rename from lambdas/modelling_e2e/Dockerfile rename to applications/modelling_e2e/Dockerfile index 0f17e99a..fa13f8ef 100644 --- a/lambdas/modelling_e2e/Dockerfile +++ b/applications/modelling_e2e/Dockerfile @@ -2,7 +2,7 @@ FROM public.ecr.aws/lambda/python:3.11 WORKDIR /var/task -COPY lambdas/modelling_e2e/requirements.txt . +COPY applications/modelling_e2e/requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY datatypes/ datatypes/ @@ -19,6 +19,6 @@ COPY tests/__init__.py tests/__init__.py COPY tests/orchestration/__init__.py tests/orchestration/__init__.py COPY tests/orchestration/fakes.py tests/orchestration/fakes.py -COPY lambdas/ lambdas/ +COPY applications/ applications/ -CMD ["lambdas.modelling_e2e.handler.handler"] +CMD ["applications.modelling_e2e.handler.handler"] diff --git a/lambdas/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py similarity index 91% rename from lambdas/modelling_e2e/handler.py rename to applications/modelling_e2e/handler.py index 0fed0fb4..ebea872e 100644 --- a/lambdas/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -39,6 +39,7 @@ from infrastructure.solar.google_solar_api_client import ( BuildingInsightsNotFoundError, GoogleSolarApiClient, ) +from applications.modelling_e2e.modelling_e2e_trigger_body import ModellingE2ETriggerBody from repositories.geospatial.geospatial_s3_repository import ( GeospatialS3Repository, ParquetReader, @@ -102,11 +103,12 @@ def _solar_insights_for( @task_handler(task_source="modelling_e2e", source=Source.PROPERTY) def handler(body: dict[str, Any], context: Any) -> None: - property_id = int(body["property_id"]) - portfolio_id = int(body["portfolio_id"]) - scenario_id = int(body["scenario_id"]) - no_solar = bool(body.get("no_solar", False)) - dry_run = bool(body.get("dry_run", False)) + trigger = ModellingE2ETriggerBody.model_validate(body) + property_id = trigger.property_id + portfolio_id = trigger.portfolio_id + scenario_id = trigger.scenario_id + no_solar = trigger.no_solar + dry_run = trigger.dry_run engine = _get_engine() epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"]) @@ -135,7 +137,9 @@ def handler(body: dict[str, Any], context: Any) -> None: effective_epc = overlaid.effective_epc spatial = _spatial_for(geospatial, uprn) - restrictions = spatial.restrictions if spatial is not None else PlanningRestrictions() + restrictions = ( + spatial.restrictions if spatial is not None else PlanningRestrictions() + ) solar_insights = None if no_solar else _solar_insights_for(solar_client, spatial) with Session(engine) as session: @@ -144,9 +148,9 @@ def handler(body: dict[str, Any], context: Any) -> None: # secondary_heating_removal is absent from the live material.type enum; # exclude it unconditionally until the catalogue gap is resolved. - considered: Optional[frozenset[MeasureType]] = ( - frozenset(MeasureType) - {MeasureType.SECONDARY_HEATING_REMOVAL} - ) + considered: Optional[frozenset[MeasureType]] = frozenset(MeasureType) - { + MeasureType.SECONDARY_HEATING_REMOVAL + } plan = run_modelling( effective_epc, diff --git a/applications/modelling_e2e/modelling_e2e_trigger_body.py b/applications/modelling_e2e/modelling_e2e_trigger_body.py new file mode 100644 index 00000000..908f9bd2 --- /dev/null +++ b/applications/modelling_e2e/modelling_e2e_trigger_body.py @@ -0,0 +1,11 @@ +from pydantic import BaseModel, ConfigDict + + +class ModellingE2ETriggerBody(BaseModel): + model_config = ConfigDict(extra="allow") + + property_id: int + portfolio_id: int + scenario_id: int + no_solar: bool = False + dry_run: bool = False diff --git a/lambdas/modelling_e2e/requirements.txt b/applications/modelling_e2e/requirements.txt similarity index 100% rename from lambdas/modelling_e2e/requirements.txt rename to applications/modelling_e2e/requirements.txt diff --git a/lambdas/__init__.py b/lambdas/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/lambdas/modelling_e2e/__init__.py b/lambdas/modelling_e2e/__init__.py deleted file mode 100644 index e69de29b..00000000 From 409fb8485cfeca9def8ba182968059957af5152e Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 13:43:52 +0000 Subject: [PATCH 04/14] local handler --- .../local_handler/.env.local.example | 36 +++++++++++++++++++ .../local_handler/docker-compose.yml | 9 +++++ .../local_handler/invoke_local_lambda.py | 30 ++++++++++++++++ 3 files changed, 75 insertions(+) create mode 100644 applications/modelling_e2e/local_handler/.env.local.example create mode 100644 applications/modelling_e2e/local_handler/docker-compose.yml create mode 100644 applications/modelling_e2e/local_handler/invoke_local_lambda.py diff --git a/applications/modelling_e2e/local_handler/.env.local.example b/applications/modelling_e2e/local_handler/.env.local.example new file mode 100644 index 00000000..8054ecd8 --- /dev/null +++ b/applications/modelling_e2e/local_handler/.env.local.example @@ -0,0 +1,36 @@ +# Local-test environment for the modelling_e2e Lambda. +# +# cp .env.local.example .env.local then fill in the values below. +# +# .env.local is gitignored. The container hits REAL AWS and a REAL Postgres, +# so every value here points at infrastructure that exists. +# +# Set dry_run=true in invoke_local_lambda.py to run the full pipeline without +# writing anything to the DB — safe for local testing. +# +# Keep comments on their own lines — docker-compose's env_file parser folds a +# trailing "# ..." into the value. + +# --- Postgres (infrastructure/postgres/config.py -> PostgresConfig.from_env) --- +# POSTGRES_HOST <- DB_HOST, PORT <- DB_PORT, USERNAME <- DB_USERNAME, +# PASSWORD <- DB_PASSWORD, DATABASE <- DB_NAME. +POSTGRES_HOST= +POSTGRES_PORT=5432 +POSTGRES_USERNAME= +POSTGRES_PASSWORD= +POSTGRES_DATABASE= +# POSTGRES_DRIVER=psycopg2 (optional; defaults to psycopg2) + +# --- Handler config (applications/modelling_e2e/handler.py) --- +# OPEN_EPC_API_TOKEN: gov.uk EPC API token (root .env: OPEN_EPC_API_TOKEN). +# GOOGLE_SOLAR_API_KEY: Google Solar API key (root .env: GOOGLE_SOLAR_API_KEY). +# DATA_BUCKET: S3 bucket holding geospatial parquet files (root .env: DATA_BUCKET). +OPEN_EPC_API_TOKEN= +GOOGLE_SOLAR_API_KEY= +DATA_BUCKET= + +# --- AWS credentials for boto3 (S3 + EPC client) --- +AWS_ACCESS_KEY_ID= +AWS_SECRET_ACCESS_KEY= +AWS_DEFAULT_REGION=eu-west-2 +# AWS_SESSION_TOKEN= (only if using temporary/SSO credentials) diff --git a/applications/modelling_e2e/local_handler/docker-compose.yml b/applications/modelling_e2e/local_handler/docker-compose.yml new file mode 100644 index 00000000..09e41d65 --- /dev/null +++ b/applications/modelling_e2e/local_handler/docker-compose.yml @@ -0,0 +1,9 @@ +services: + modelling-e2e: + build: + context: ../../../ + dockerfile: applications/modelling_e2e/Dockerfile + ports: + - "9004:8080" + env_file: + - ../../../.env diff --git a/applications/modelling_e2e/local_handler/invoke_local_lambda.py b/applications/modelling_e2e/local_handler/invoke_local_lambda.py new file mode 100644 index 00000000..3d237973 --- /dev/null +++ b/applications/modelling_e2e/local_handler/invoke_local_lambda.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 +import json +import requests + +HOST = "localhost" +PORT = "9004" + +LAMBDA_URL = f"http://{HOST}:{PORT}/2015-03-31/functions/function/invocations" + +payload = { + "Records": [ + { + "body": json.dumps( + { + "property_id": 0, + "portfolio_id": 0, + "scenario_id": 0, + "no_solar": False, + "dry_run": True, + } + ) + } + ] +} + +response = requests.post(LAMBDA_URL, json=payload) + +print("Status code:", response.status_code) +print("Response:") +print(response.text) From dc830fea63f19a95a8e6b75ad10439efd190064c Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 14:21:52 +0000 Subject: [PATCH 05/14] various fixes --- applications/modelling_e2e/handler.py | 41 +++++++++++++++---- .../local_handler/invoke_local_lambda.py | 6 +-- applications/modelling_e2e/requirements.txt | 5 ++- utilities/aws_lambda/task_handler.py | 8 ++++ 4 files changed, 48 insertions(+), 12 deletions(-) diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index ebea872e..6b276f0c 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -39,7 +39,9 @@ from infrastructure.solar.google_solar_api_client import ( BuildingInsightsNotFoundError, GoogleSolarApiClient, ) -from applications.modelling_e2e.modelling_e2e_trigger_body import ModellingE2ETriggerBody +from applications.modelling_e2e.modelling_e2e_trigger_body import ( + ModellingE2ETriggerBody, +) from repositories.geospatial.geospatial_s3_repository import ( GeospatialS3Repository, ParquetReader, @@ -57,9 +59,12 @@ from repositories.scenario.scenario_postgres_repository import ( ScenarioPostgresRepository, ) from utilities.aws_lambda.task_handler import task_handler +from utilities.logger import setup_logger _engine: Optional[Engine] = None +logger = setup_logger() + def _get_engine() -> Engine: global _engine @@ -72,7 +77,9 @@ def _s3_parquet_reader() -> ParquetReader: bucket = os.environ["DATA_BUCKET"] def read(key: str) -> pd.DataFrame: - s3: Any = cast(Any, boto3.client("s3")) # pyright: ignore[reportUnknownMemberType] + s3: Any = cast( + Any, boto3.client("s3") + ) # pyright: ignore[reportUnknownMemberType] raw = cast(bytes, s3.get_object(Bucket=bucket, Key=key)["Body"].read()) return pd.read_parquet(io.BytesIO(raw)) # type: ignore[return-value] @@ -110,6 +117,11 @@ def handler(body: dict[str, Any], context: Any) -> None: no_solar = trigger.no_solar dry_run = trigger.dry_run + logger.info( + f"start property={property_id} portfolio={portfolio_id} " + f"scenario={scenario_id} no_solar={no_solar} dry_run={dry_run}" + ) + engine = _get_engine() epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"]) geospatial = GeospatialS3Repository(_s3_parquet_reader()) @@ -121,10 +133,12 @@ def handler(body: dict[str, Any], context: Any) -> None: {"pid": property_id}, ).one() uprn = int(row[0]) + logger.info(f"resolved uprn={uprn}") epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn) if epc is None: raise ValueError(f"no EPC found for UPRN {uprn} (property {property_id})") + logger.info(f"fetched EPC (energy_rating_current={epc.energy_rating_current})") overrides_reader = PropertyOverridesPostgresReader(lambda: Session(engine)) overlaid = Property( @@ -140,10 +154,18 @@ def handler(body: dict[str, Any], context: Any) -> None: restrictions = ( spatial.restrictions if spatial is not None else PlanningRestrictions() ) - solar_insights = None if no_solar else _solar_insights_for(solar_client, spatial) + logger.info(f"spatial={'found' if spatial is not None else 'not found'}") + + if no_solar: + solar_insights = None + logger.info("solar skipped (no_solar=True)") + else: + solar_insights = _solar_insights_for(solar_client, spatial) + logger.info(f"solar={'found' if solar_insights is not None else 'not found'}") with Session(engine) as session: scenario = ScenarioPostgresRepository(session).get_many([scenario_id])[0] + logger.info(f"loaded scenario goal={scenario.goal!r} goal_value={scenario.goal_value!r}") products = ProductPostgresRepository(session) # secondary_heating_removal is absent from the live material.type enum; @@ -152,6 +174,7 @@ def handler(body: dict[str, Any], context: Any) -> None: MeasureType.SECONDARY_HEATING_REMOVAL } + logger.info("running modelling pipeline") plan = run_modelling( effective_epc, planning_restrictions=restrictions, @@ -161,14 +184,15 @@ def handler(body: dict[str, Any], context: Any) -> None: scenario=scenario, print_table=False, ) + logger.info( + f"modelling complete: SAP {plan.baseline.sap_continuous:.1f}→" + f"{plan.post_sap_continuous:.1f} measures={len(plan.measures)} " + f"cost=£{plan.cost_of_works:,.0f}" + ) if dry_run: measure_types = ", ".join(m.measure_type for m in plan.measures) or "none" - print( - f"[dry_run] property={property_id} scenario={scenario_id} " - f"SAP {plan.baseline.sap_continuous:.1f}→{plan.post_sap_continuous:.1f} " - f"measures=[{measure_types}] cost=£{plan.cost_of_works:,.0f}" - ) + logger.info(f"[dry_run] measures=[{measure_types}] — skipping DB write") return PlanPostgresRepository(session).save( plan, @@ -181,3 +205,4 @@ def handler(body: dict[str, Any], context: Any) -> None: property_id, has_recommendations=bool(plan.measures) ) session.commit() + logger.info("plan saved") diff --git a/applications/modelling_e2e/local_handler/invoke_local_lambda.py b/applications/modelling_e2e/local_handler/invoke_local_lambda.py index 3d237973..808200c6 100644 --- a/applications/modelling_e2e/local_handler/invoke_local_lambda.py +++ b/applications/modelling_e2e/local_handler/invoke_local_lambda.py @@ -12,9 +12,9 @@ payload = { { "body": json.dumps( { - "property_id": 0, - "portfolio_id": 0, - "scenario_id": 0, + "property_id": 709634, + "portfolio_id": 785, + "scenario_id": 1266, "no_solar": False, "dry_run": True, } diff --git a/applications/modelling_e2e/requirements.txt b/applications/modelling_e2e/requirements.txt index 51ee9050..bac38184 100644 --- a/applications/modelling_e2e/requirements.txt +++ b/applications/modelling_e2e/requirements.txt @@ -1,7 +1,10 @@ awslambdaric boto3 +numpy==2.1.2 pandas==2.2.2 -pyarrow +pyarrow==17.0.0 +httpx +requests pydantic sqlalchemy==2.0.36 sqlmodel diff --git a/utilities/aws_lambda/task_handler.py b/utilities/aws_lambda/task_handler.py index 82c7198e..093d6bd6 100644 --- a/utilities/aws_lambda/task_handler.py +++ b/utilities/aws_lambda/task_handler.py @@ -5,6 +5,7 @@ TaskOrchestrator.create_task_with_subtask(...) + run_subtask(...). """ import json +import logging from contextlib import AbstractContextManager from functools import wraps from typing import Any, Callable, Optional, cast @@ -13,6 +14,8 @@ from utilities.aws_lambda.default_orchestrator import default_orchestrator from domain.tasks.tasks import Source from orchestration.task_orchestrator import TaskOrchestrator +logger = logging.getLogger(__name__) + OrchestratorCM = Callable[[], AbstractContextManager[TaskOrchestrator]] @@ -63,6 +66,11 @@ def task_handler( ) results.append(result) except Exception: + logger.exception( + "subtask failed (task_source=%s source_id=%s)", + task_source, + source_id, + ) if "Records" in event: message_id = record.get("messageId", "") failures.append({"itemIdentifier": message_id}) From 8ee53a3a952894d7600df65c58283fb8ed261998 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 14:45:03 +0000 Subject: [PATCH 06/14] persist EPC data --- applications/modelling_e2e/handler.py | 4 ++++ .../modelling_e2e/local_handler/invoke_local_lambda.py | 8 ++++---- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index 6b276f0c..7c8bc4f9 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -46,6 +46,7 @@ from repositories.geospatial.geospatial_s3_repository import ( GeospatialS3Repository, ParquetReader, ) +from repositories.epc.epc_postgres_repository import EpcPostgresRepository from repositories.plan.plan_postgres_repository import PlanPostgresRepository from repositories.product.product_postgres_repository import ProductPostgresRepository from repositories.property.landlord_override_overlays import overlays_from @@ -194,6 +195,9 @@ def handler(body: dict[str, Any], context: Any) -> None: measure_types = ", ".join(m.measure_type for m in plan.measures) or "none" logger.info(f"[dry_run] measures=[{measure_types}] — skipping DB write") return + EpcPostgresRepository(session).save( + epc, property_id=property_id, portfolio_id=portfolio_id + ) PlanPostgresRepository(session).save( plan, property_id=property_id, diff --git a/applications/modelling_e2e/local_handler/invoke_local_lambda.py b/applications/modelling_e2e/local_handler/invoke_local_lambda.py index 808200c6..e28d6e84 100644 --- a/applications/modelling_e2e/local_handler/invoke_local_lambda.py +++ b/applications/modelling_e2e/local_handler/invoke_local_lambda.py @@ -12,11 +12,11 @@ payload = { { "body": json.dumps( { - "property_id": 709634, - "portfolio_id": 785, - "scenario_id": 1266, + "property_id": 709774, + "portfolio_id": 796, + "scenario_id": 1268, "no_solar": False, - "dry_run": True, + "dry_run": False, } ) } From 307217eb5dfa54d48b362f9b02017def5d7d1d71 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 14:45:14 +0000 Subject: [PATCH 07/14] dry run true --- applications/modelling_e2e/local_handler/invoke_local_lambda.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/applications/modelling_e2e/local_handler/invoke_local_lambda.py b/applications/modelling_e2e/local_handler/invoke_local_lambda.py index e28d6e84..c88d4362 100644 --- a/applications/modelling_e2e/local_handler/invoke_local_lambda.py +++ b/applications/modelling_e2e/local_handler/invoke_local_lambda.py @@ -16,7 +16,7 @@ payload = { "portfolio_id": 796, "scenario_id": 1268, "no_solar": False, - "dry_run": False, + "dry_run": True, } ) } From 233dace245e1997f657b737c48f45dacb393cfe4 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 14:49:28 +0000 Subject: [PATCH 08/14] 2 concurrent executions --- deployment/terraform/lambda/modelling_e2e/variables.tf | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/deployment/terraform/lambda/modelling_e2e/variables.tf b/deployment/terraform/lambda/modelling_e2e/variables.tf index 7fd1ecde..e45ec68a 100644 --- a/deployment/terraform/lambda/modelling_e2e/variables.tf +++ b/deployment/terraform/lambda/modelling_e2e/variables.tf @@ -20,8 +20,8 @@ variable "image_digest" { variable "reserved_concurrent_executions" { type = number - default = 1 - description = "Start at 1 to validate correctness before scaling up." + default = 2 + description = "Concurrent executions" } variable "batch_size" { From 2156109e99467fc882c940e3deceaca5f005704d Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 14:54:23 +0000 Subject: [PATCH 09/14] concurrency = 2 --- deployment/terraform/lambda/modelling_e2e/main.tf | 3 ++- deployment/terraform/lambda/modelling_e2e/variables.tf | 8 +++++++- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/deployment/terraform/lambda/modelling_e2e/main.tf b/deployment/terraform/lambda/modelling_e2e/main.tf index 31f90e1f..e7b4e127 100644 --- a/deployment/terraform/lambda/modelling_e2e/main.tf +++ b/deployment/terraform/lambda/modelling_e2e/main.tf @@ -25,7 +25,8 @@ module "lambda" { reserved_concurrent_executions = var.reserved_concurrent_executions - batch_size = var.batch_size + batch_size = var.batch_size + maximum_concurrency = var.maximum_concurrency timeout = 60 memory_size = 1024 diff --git a/deployment/terraform/lambda/modelling_e2e/variables.tf b/deployment/terraform/lambda/modelling_e2e/variables.tf index e45ec68a..417f2780 100644 --- a/deployment/terraform/lambda/modelling_e2e/variables.tf +++ b/deployment/terraform/lambda/modelling_e2e/variables.tf @@ -19,9 +19,15 @@ variable "image_digest" { } variable "reserved_concurrent_executions" { + type = number + default = -1 + description = "Reserved concurrency for the Lambda. -1 = unreserved." +} + +variable "maximum_concurrency" { type = number default = 2 - description = "Concurrent executions" + description = "Maximum concurrent Lambda invocations from the SQS trigger." } variable "batch_size" { From 7e5af6c8f4144a5c89e21bf34e47064615ee2ead Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 15:46:18 +0000 Subject: [PATCH 10/14] process multiple properties in one message --- applications/modelling_e2e/handler.py | 322 ++++++--- .../local_handler/invoke_local_lambda.py | 4 +- .../modelling_e2e_trigger_body.py | 2 +- tests/applications/modelling_e2e/__init__.py | 0 .../modelling_e2e/test_handler.py | 679 ++++++++++++++++++ 5 files changed, 918 insertions(+), 89 deletions(-) create mode 100644 tests/applications/modelling_e2e/__init__.py create mode 100644 tests/applications/modelling_e2e/test_handler.py diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index 7c8bc4f9..95af5269 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -1,10 +1,16 @@ -"""SQS-triggered Lambda: fetch EPC → run modelling → persist plan. +"""SQS-triggered Lambda: fetch EPC (or predict) → run modelling → persist plan. -One SQS message = one property. The handler reads ``property_id``, -``portfolio_id``, ``scenario_id``, and ``no_solar`` from the message body, -fetches the property's EPC from the gov API, runs the full modelling pipeline -(SAP10 → optimiser) via ``harness.console.run_modelling``, and persists the -resulting Plan via ``PlanPostgresRepository.save()``. +One SQS message = one batch of properties sharing a portfolio, scenario, and +(by caller convention) postcode. The handler reads ``property_ids``, +``portfolio_id``, ``scenario_id``, ``no_solar``, and ``dry_run`` from the +message body, fetches or predicts each property's EPC, runs the full modelling +pipeline (SAP10 → optimiser) via ``harness.console.run_modelling``, and +persists the resulting Plan via ``PostgresUnitOfWork`` in one atomic transaction +per property. + +When no lodged EPC is found, EPC Prediction (Path 3, ADR-0031) synthesises one +from the postcode cohort. ``_cohort_cache`` is module-level so warm Lambda +containers re-processing the same postcode avoid redundant fetches. ``secondary_heating_removal`` is excluded unconditionally: the live ``material`` catalogue does not yet carry this measure type, causing a crash during catalogue @@ -18,6 +24,7 @@ from __future__ import annotations import io import os +from collections.abc import Callable from typing import Any, Optional, cast import boto3 @@ -25,7 +32,17 @@ import pandas as pd # pyright: ignore[reportMissingTypeStubs] from sqlalchemy import Engine, text from sqlmodel import Session -from datatypes.epc.domain.epc_property_data import EpcPropertyData +from datatypes.epc.domain.epc_property_data import ( + BuildingPartIdentifier, + EpcPropertyData, +) +from domain.epc_prediction.comparable_properties import ( + ComparableProperty, + select_comparables, +) +from domain.epc_prediction.epc_prediction import EpcPrediction +from domain.epc_prediction.prediction_target import build_prediction_target +from domain.geospatial.coordinates import Coordinates from domain.geospatial.planning_restrictions import PlanningRestrictions from domain.geospatial.spatial_reference import SpatialReference from domain.modelling.measure_type import MeasureType @@ -42,20 +59,22 @@ from infrastructure.solar.google_solar_api_client import ( from applications.modelling_e2e.modelling_e2e_trigger_body import ( ModellingE2ETriggerBody, ) +from repositories.comparable_properties.epc_comparable_properties_repository import ( + EpcComparablePropertiesRepository, +) from repositories.geospatial.geospatial_s3_repository import ( GeospatialS3Repository, ParquetReader, ) -from repositories.epc.epc_postgres_repository import EpcPostgresRepository -from repositories.plan.plan_postgres_repository import PlanPostgresRepository +from repositories.postgres_unit_of_work import PostgresUnitOfWork from repositories.product.product_postgres_repository import ProductPostgresRepository from repositories.property.landlord_override_overlays import overlays_from +from repositories.property.override_backed_prediction_attributes_reader import ( + OverrideBackedPredictionAttributesReader, +) from repositories.property.property_overrides_postgres_reader import ( PropertyOverridesPostgresReader, ) -from repositories.property.property_postgres_repository import ( - PropertyPostgresRepository, -) from repositories.scenario.scenario_postgres_repository import ( ScenarioPostgresRepository, ) @@ -63,6 +82,7 @@ from utilities.aws_lambda.task_handler import task_handler from utilities.logger import setup_logger _engine: Optional[Engine] = None +_cohort_cache: dict[str, list[ComparableProperty]] = {} logger = setup_logger() @@ -109,17 +129,53 @@ def _solar_insights_for( return None +def _predict_epc( + *, + property_id: int, + uprn: int, + postcode: str, + portfolio_id: int, + attributes_reader: OverrideBackedPredictionAttributesReader, + coordinates: Optional[Coordinates], + cohort_for: Callable[[str], list[ComparableProperty]], + predictor: EpcPrediction, +) -> Optional[EpcPropertyData]: + """Synthesise an EpcPropertyData for an EPC-less property from its postcode + cohort (EPC Prediction Path 3, ADR-0031), or None when ineligible. + + Returns None when property_type is unresolvable (hard cohort filter cannot + fire) or when the postcode cohort is empty after filtering. + """ + attributes = attributes_reader.attributes_for(property_id) + identity = PropertyIdentity( + portfolio_id=portfolio_id, postcode=postcode, address="", uprn=uprn + ) + target = build_prediction_target(identity, coordinates, attributes) + if target is None: + return None + comparables = select_comparables(target, cohort_for(target.postcode)) + if not comparables.members: + return None + predicted = predictor.predict(target, comparables) + if not any( + part.identifier is BuildingPartIdentifier.MAIN + for part in predicted.sap_building_parts + ): + return None + return predicted + + @task_handler(task_source="modelling_e2e", source=Source.PROPERTY) def handler(body: dict[str, Any], context: Any) -> None: trigger = ModellingE2ETriggerBody.model_validate(body) - property_id = trigger.property_id + property_ids = trigger.property_ids portfolio_id = trigger.portfolio_id scenario_id = trigger.scenario_id no_solar = trigger.no_solar dry_run = trigger.dry_run logger.info( - f"start property={property_id} portfolio={portfolio_id} " + f"start property_ids={property_ids} portfolio={portfolio_id} " f"scenario={scenario_id} no_solar={no_solar} dry_run={dry_run}" ) @@ -129,84 +185,178 @@ def handler(body: dict[str, Any], context: Any) -> None: solar_client = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"]) with engine.connect() as conn: - row = conn.execute( - text("SELECT uprn FROM property WHERE id = :pid"), - {"pid": property_id}, - ).one() - uprn = int(row[0]) - logger.info(f"resolved uprn={uprn}") + uprn_rows = conn.execute( + text("SELECT id, uprn FROM property WHERE id = ANY(:ids)"), + {"ids": property_ids}, + ).fetchall() + postcode_rows = conn.execute( + text("SELECT id, postcode FROM property WHERE id = ANY(:ids)"), + {"ids": property_ids}, + ).fetchall() - epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn) - if epc is None: - raise ValueError(f"no EPC found for UPRN {uprn} (property {property_id})") - logger.info(f"fetched EPC (energy_rating_current={epc.energy_rating_current})") + uprns: dict[int, int] = {int(row[0]): int(row[1]) for row in uprn_rows} + postcodes: dict[int, str] = { + int(row[0]): (row[1] or "") for row in postcode_rows + } overrides_reader = PropertyOverridesPostgresReader(lambda: Session(engine)) - overlaid = Property( - identity=PropertyIdentity( - portfolio_id=portfolio_id, postcode="", address="", uprn=uprn - ), - epc=epc, - landlord_overrides=overlays_from(overrides_reader.overrides_for(property_id)), - ) - effective_epc = overlaid.effective_epc + prediction_attrs_reader = OverrideBackedPredictionAttributesReader(overrides_reader) + comparables_repo = EpcComparablePropertiesRepository(epc_client, geospatial) + predictor = EpcPrediction() - spatial = _spatial_for(geospatial, uprn) - restrictions = ( - spatial.restrictions if spatial is not None else PlanningRestrictions() - ) - logger.info(f"spatial={'found' if spatial is not None else 'not found'}") + def _get_cohort(postcode: str) -> list[ComparableProperty]: + if postcode not in _cohort_cache: + _cohort_cache[postcode] = ( + comparables_repo.candidates_for(postcode) if postcode else [] + ) + return _cohort_cache[postcode] - if no_solar: - solar_insights = None - logger.info("solar skipped (no_solar=True)") - else: - solar_insights = _solar_insights_for(solar_client, spatial) - logger.info(f"solar={'found' if solar_insights is not None else 'not found'}") + read_session = Session(engine) + try: + scenario = ScenarioPostgresRepository(read_session).get_many([scenario_id])[0] + products = ProductPostgresRepository(read_session) - with Session(engine) as session: - scenario = ScenarioPostgresRepository(session).get_many([scenario_id])[0] - logger.info(f"loaded scenario goal={scenario.goal!r} goal_value={scenario.goal_value!r}") - products = ProductPostgresRepository(session) + errors: list[int] = [] - # secondary_heating_removal is absent from the live material.type enum; - # exclude it unconditionally until the catalogue gap is resolved. - considered: Optional[frozenset[MeasureType]] = frozenset(MeasureType) - { - MeasureType.SECONDARY_HEATING_REMOVAL - } + for property_id in property_ids: + try: + uprn = uprns[property_id] + postcode = postcodes.get(property_id, "") + logger.info( + f"property={property_id} uprn={uprn} postcode={postcode!r}" + ) - logger.info("running modelling pipeline") - plan = run_modelling( - effective_epc, - planning_restrictions=restrictions, - solar_insights=solar_insights, - considered_measures=considered, - products=products, - scenario=scenario, - print_table=False, - ) - logger.info( - f"modelling complete: SAP {plan.baseline.sap_continuous:.1f}→" - f"{plan.post_sap_continuous:.1f} measures={len(plan.measures)} " - f"cost=£{plan.cost_of_works:,.0f}" - ) + spatial = _spatial_for(geospatial, uprn) + restrictions = ( + spatial.restrictions + if spatial is not None + else PlanningRestrictions() + ) + coordinates: Optional[Coordinates] = ( + spatial.coordinates if spatial is not None else None + ) - if dry_run: - measure_types = ", ".join(m.measure_type for m in plan.measures) or "none" - logger.info(f"[dry_run] measures=[{measure_types}] — skipping DB write") - return - EpcPostgresRepository(session).save( - epc, property_id=property_id, portfolio_id=portfolio_id - ) - PlanPostgresRepository(session).save( - plan, - property_id=property_id, - scenario_id=scenario_id, - portfolio_id=portfolio_id, - is_default=scenario.is_default, - ) - PropertyPostgresRepository(session).mark_modelled( - property_id, has_recommendations=bool(plan.measures) - ) - session.commit() - logger.info("plan saved") + epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn) + overrides = overlays_from(overrides_reader.overrides_for(property_id)) + + if epc is not None: + logger.info(f"property={property_id} lodged EPC found") + effective_epc = Property( + identity=PropertyIdentity( + portfolio_id=portfolio_id, + postcode=postcode, + address="", + uprn=uprn, + ), + epc=epc, + landlord_overrides=overrides, + ).effective_epc + else: + logger.info( + f"property={property_id} no lodged EPC — attempting prediction" + ) + predicted_epc = _predict_epc( + property_id=property_id, + uprn=uprn, + postcode=postcode, + portfolio_id=portfolio_id, + attributes_reader=prediction_attrs_reader, + coordinates=coordinates, + cohort_for=_get_cohort, + predictor=predictor, + ) + if predicted_epc is None: + raise ValueError( + f"no EPC for UPRN {uprn} and not predictable " + f"(unresolved property_type or empty '{postcode}' cohort)" + ) + effective_epc = Property( + identity=PropertyIdentity( + portfolio_id=portfolio_id, + postcode=postcode, + address="", + uprn=uprn, + ), + epc=None, + predicted_epc=predicted_epc, + landlord_overrides=overrides, + ).effective_epc + + solar_insights: Optional[dict[str, Any]] = ( + None + if no_solar + else _solar_insights_for(solar_client, spatial) + ) + + # secondary_heating_removal is absent from the live material.type + # enum; exclude unconditionally until the catalogue gap is resolved. + considered: Optional[frozenset[MeasureType]] = frozenset( + MeasureType + ) - {MeasureType.SECONDARY_HEATING_REMOVAL} + + plan = run_modelling( + effective_epc, + planning_restrictions=restrictions, + solar_insights=solar_insights, + considered_measures=considered, + products=products, + scenario=scenario, + print_table=False, + ) + logger.info( + f"property={property_id} modelling complete " + f"measures={len(plan.measures)}" + ) + + if dry_run: + measure_types = ( + ", ".join(m.measure_type for m in plan.measures) or "none" + ) + logger.info( + f"[dry_run] property={property_id} " + f"measures=[{measure_types}] — skipping DB write" + ) + continue + + with PostgresUnitOfWork(lambda: Session(engine)) as uow: + if epc is not None: + uow.epc.save( + epc, property_id=property_id, portfolio_id=portfolio_id + ) + if spatial is not None: + uow.spatial.save(uprn, spatial) + if ( + solar_insights is not None + and 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, + scenario_id=scenario_id, + portfolio_id=portfolio_id, + is_default=scenario.is_default, + ) + uow.property.mark_modelled( + property_id, has_recommendations=bool(plan.measures) + ) + uow.commit() + logger.info(f"property={property_id} plan saved") + + except Exception as error: # noqa: BLE001 + logger.error( + f"property={property_id}: {type(error).__name__}: {error}", + exc_info=True, + ) + errors.append(property_id) + + if errors: + raise RuntimeError(f"failed property_ids: {errors}") + finally: + read_session.close() diff --git a/applications/modelling_e2e/local_handler/invoke_local_lambda.py b/applications/modelling_e2e/local_handler/invoke_local_lambda.py index c88d4362..c667ffc6 100644 --- a/applications/modelling_e2e/local_handler/invoke_local_lambda.py +++ b/applications/modelling_e2e/local_handler/invoke_local_lambda.py @@ -12,11 +12,11 @@ payload = { { "body": json.dumps( { - "property_id": 709774, + "property_ids": [709773, 709774], "portfolio_id": 796, "scenario_id": 1268, "no_solar": False, - "dry_run": True, + "dry_run": False, } ) } diff --git a/applications/modelling_e2e/modelling_e2e_trigger_body.py b/applications/modelling_e2e/modelling_e2e_trigger_body.py index 908f9bd2..cb83118a 100644 --- a/applications/modelling_e2e/modelling_e2e_trigger_body.py +++ b/applications/modelling_e2e/modelling_e2e_trigger_body.py @@ -4,7 +4,7 @@ from pydantic import BaseModel, ConfigDict class ModellingE2ETriggerBody(BaseModel): model_config = ConfigDict(extra="allow") - property_id: int + property_ids: list[int] portfolio_id: int scenario_id: int no_solar: bool = False diff --git a/tests/applications/modelling_e2e/__init__.py b/tests/applications/modelling_e2e/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py new file mode 100644 index 00000000..5979fe1b --- /dev/null +++ b/tests/applications/modelling_e2e/test_handler.py @@ -0,0 +1,679 @@ +"""Tests for the modelling_e2e Lambda handler. + +Tests exercise the handler's external behaviour through handler.__wrapped__, +patching I/O boundaries (EPC client, DB reads, UoW) so no real DB or network +is needed. One test per distinct behaviour path. +""" + +from __future__ import annotations + +from contextlib import ExitStack +from typing import Any +from unittest.mock import MagicMock, call, patch + +import pytest +from pydantic import ValidationError + +from applications.modelling_e2e.modelling_e2e_trigger_body import ( + ModellingE2ETriggerBody, +) + +PROPERTY_ID = 12345 +UPRN = 987654321 +POSTCODE = "SW1A 1AA" +PORTFOLIO_ID = 100 +SCENARIO_ID = 200 + +_ENV = { + "OPEN_EPC_API_TOKEN": "test-token", + "DATA_BUCKET": "test-bucket", + "GOOGLE_SOLAR_API_KEY": "test-solar-key", +} + +_BODY = { + "property_ids": [PROPERTY_ID], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": True, + "dry_run": False, +} + + +def _call_handler(body: dict[str, Any]) -> Any: + from applications.modelling_e2e.handler import handler + + return handler.__wrapped__(body, None) # type: ignore[attr-defined] + + +def _engine_mock( + property_ids: list[int], + uprns: list[int], + postcodes: list[str], +) -> MagicMock: + """Mock engine whose connect() returns UPRN then postcode rows.""" + mock_engine = MagicMock() + mock_conn = mock_engine.connect.return_value.__enter__.return_value + + uprn_result = MagicMock() + uprn_result.fetchall.return_value = list(zip(property_ids, uprns)) + + postcode_result = MagicMock() + postcode_result.fetchall.return_value = list(zip(property_ids, postcodes)) + + mock_conn.execute.side_effect = [uprn_result, postcode_result] + return mock_engine + + +def _plan_mock() -> MagicMock: + plan = MagicMock() + plan.measures = [] + plan.cost_of_works = 0.0 + return plan + + +# --------------------------------------------------------------------------- +# Fixture: clear module-level cohort cache between tests +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _clear_cohort_cache() -> None: + import applications.modelling_e2e.handler as h + + h._cohort_cache.clear() + + +# --------------------------------------------------------------------------- +# Trigger body validation +# --------------------------------------------------------------------------- + + +def test_trigger_body_requires_property_ids() -> None: + """property_ids (plural) must be provided and must be a list of ints.""" + # Arrange + body = { + "property_ids": [1, 2], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + } + + # Act + result = ModellingE2ETriggerBody.model_validate(body) + + # Assert + assert result.property_ids == [1, 2] + + +def test_trigger_body_rejects_missing_property_ids() -> None: + with pytest.raises(ValidationError): + ModellingE2ETriggerBody.model_validate( + {"portfolio_id": PORTFOLIO_ID, "scenario_id": SCENARIO_ID} + ) + + +# --------------------------------------------------------------------------- +# Lodged EPC path +# --------------------------------------------------------------------------- + + +def test_lodged_epc_path_saves_epc_plan_and_marks_modelled() -> None: + """When get_by_uprn returns an EPC the handler saves it, saves the plan, + and marks the property as modelled — all inside one UoW per property.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + mock_epc = MagicMock() + mock_plan = _plan_mock() + mock_uow = MagicMock() + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = mock_epc + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._solar_insights_for", + return_value=None, + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.Session") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=mock_plan, + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + MockUoW.return_value.__enter__.return_value = mock_uow + MockUoW.return_value.__exit__.return_value = False + + # Act + _call_handler(_BODY) + + # Assert — EPC saved (lodged path), plan saved, property marked modelled + mock_uow.epc.save.assert_called_once_with( + mock_epc, property_id=PROPERTY_ID, portfolio_id=PORTFOLIO_ID + ) + mock_uow.plan.save.assert_called_once() + mock_uow.property.mark_modelled.assert_called_once_with( + PROPERTY_ID, has_recommendations=False + ) + mock_uow.commit.assert_called_once() + + +# --------------------------------------------------------------------------- +# EPC Prediction path +# --------------------------------------------------------------------------- + + +def test_prediction_path_saves_plan_without_epc_save() -> None: + """When get_by_uprn returns None the handler synthesises an EPC via + prediction and saves the plan — but never calls epc.save.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + mock_plan = _plan_mock() + mock_uow = MagicMock() + + mock_predicted_epc = MagicMock() + # _predict_epc checks for a MAIN building part + from datatypes.epc.domain.epc_property_data import BuildingPartIdentifier + + mock_part = MagicMock() + mock_part.identifier = BuildingPartIdentifier.MAIN + mock_predicted_epc.sap_building_parts = [mock_part] + + mock_comparables = MagicMock() + mock_comparables.members = [MagicMock()] # non-empty cohort + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = None # no lodged EPC + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._solar_insights_for", + return_value=None, + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + # Prediction infrastructure + from domain.epc_prediction.prediction_target import PredictionTargetAttributes + + stack.enter_context( + patch( + "applications.modelling_e2e.handler.OverrideBackedPredictionAttributesReader" + ) + ).return_value.attributes_for.return_value = PredictionTargetAttributes( + property_type="2" + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.select_comparables") + ).return_value = mock_comparables + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcPrediction") + ).return_value.predict.return_value = mock_predicted_epc + stack.enter_context( + patch( + "applications.modelling_e2e.handler.EpcComparablePropertiesRepository" + ) + ).return_value.candidates_for.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.Session") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=mock_plan, + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + MockUoW.return_value.__enter__.return_value = mock_uow + MockUoW.return_value.__exit__.return_value = False + + # Act + _call_handler(_BODY) + + # Assert — epc.save NOT called (no lodged cert), plan IS saved + mock_uow.epc.save.assert_not_called() + mock_uow.plan.save.assert_called_once() + mock_uow.commit.assert_called_once() + + +# --------------------------------------------------------------------------- +# Prediction gate-out (empty cohort) +# --------------------------------------------------------------------------- + + +def test_empty_cohort_gates_property_out_and_raises() -> None: + """When candidates_for returns an empty list the property cannot be + predicted; the handler records it as an error and raises RuntimeError.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + + empty_comparables = MagicMock() + empty_comparables.members = [] # empty cohort → gate-out + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = None + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + from domain.epc_prediction.prediction_target import PredictionTargetAttributes + + stack.enter_context( + patch( + "applications.modelling_e2e.handler.OverrideBackedPredictionAttributesReader" + ) + ).return_value.attributes_for.return_value = PredictionTargetAttributes( + property_type="2" + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.select_comparables") + ).return_value = empty_comparables + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcPrediction") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.EpcComparablePropertiesRepository" + ) + ).return_value.candidates_for.return_value = [] + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + + # Act + with pytest.raises(RuntimeError, match=str(PROPERTY_ID)): + _call_handler(_BODY) + + # UoW never entered — the property errored before the write block + MockUoW.return_value.__enter__.assert_not_called() + + +# --------------------------------------------------------------------------- +# Partial batch failure +# --------------------------------------------------------------------------- + + +def test_partial_batch_failure_raises_runtime_error_listing_failed_ids() -> None: + """Two properties: property 1 succeeds, property 2 raises during modelling. + Handler raises RuntimeError naming only the failed property; property 1's + UoW was committed.""" + # Arrange + pid1, pid2 = 111, 222 + mock_engine = _engine_mock([pid1, pid2], [1001, 1002], [POSTCODE, POSTCODE]) + mock_plan = _plan_mock() + mock_uow = MagicMock() + + def _run_modelling_side_effect(*args: Any, **kwargs: Any) -> Any: + # Fail on second call (pid2) + if not hasattr(_run_modelling_side_effect, "_calls"): + _run_modelling_side_effect._calls = 0 # type: ignore[attr-defined] + _run_modelling_side_effect._calls += 1 # type: ignore[attr-defined] + if _run_modelling_side_effect._calls == 2: # type: ignore[attr-defined] + raise ValueError("modelling exploded") + return mock_plan + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = MagicMock() # lodged EPC + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._solar_insights_for", + return_value=None, + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + side_effect=_run_modelling_side_effect, + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + MockUoW.return_value.__enter__.return_value = mock_uow + MockUoW.return_value.__exit__.return_value = False + + # Act + with pytest.raises(RuntimeError, match=str(pid2)): + _call_handler( + { + "property_ids": [pid1, pid2], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": True, + "dry_run": False, + } + ) + + # Property 1 succeeded — its UoW was committed + mock_uow.commit.assert_called_once() + + +# --------------------------------------------------------------------------- +# Cohort cache hit +# --------------------------------------------------------------------------- + + +def test_cohort_cache_prevents_duplicate_candidates_for_calls() -> None: + """Two properties sharing a postcode: candidates_for is fetched once and + cached — EpcComparablePropertiesRepository.candidates_for called once.""" + # Arrange + pid1, pid2 = 301, 302 + mock_engine = _engine_mock( + [pid1, pid2], [3001, 3002], [POSTCODE, POSTCODE] + ) + mock_plan = _plan_mock() + mock_uow = MagicMock() + + mock_predicted_epc = MagicMock() + from datatypes.epc.domain.epc_property_data import BuildingPartIdentifier + + mock_part = MagicMock() + mock_part.identifier = BuildingPartIdentifier.MAIN + mock_predicted_epc.sap_building_parts = [mock_part] + + mock_comparables = MagicMock() + mock_comparables.members = [MagicMock()] + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = None # force prediction path + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + from domain.epc_prediction.prediction_target import PredictionTargetAttributes + + stack.enter_context( + patch( + "applications.modelling_e2e.handler.OverrideBackedPredictionAttributesReader" + ) + ).return_value.attributes_for.return_value = PredictionTargetAttributes( + property_type="2" + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.select_comparables") + ).return_value = mock_comparables + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcPrediction") + ).return_value.predict.return_value = mock_predicted_epc + MockCandidates = stack.enter_context( + patch( + "applications.modelling_e2e.handler.EpcComparablePropertiesRepository" + ) + ) + MockCandidates.return_value.candidates_for.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=mock_plan, + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + MockUoW.return_value.__enter__.return_value = mock_uow + MockUoW.return_value.__exit__.return_value = False + + # Act + _call_handler( + { + "property_ids": [pid1, pid2], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": True, + "dry_run": False, + } + ) + + # Assert — cohort fetched exactly once despite two properties + MockCandidates.return_value.candidates_for.assert_called_once_with(POSTCODE) + + +# --------------------------------------------------------------------------- +# Dry-run +# --------------------------------------------------------------------------- + + +def test_dry_run_skips_all_db_writes() -> None: + """dry_run=True: run_modelling executes but PostgresUnitOfWork is never + entered — no DB writes occur for any property in the batch.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = MagicMock() + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._solar_insights_for", + return_value=None, + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=_plan_mock(), + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + + # Act + _call_handler({**_BODY, "dry_run": True}) + + # Assert — UoW never entered + MockUoW.return_value.__enter__.assert_not_called() From 25c695186b11dcbffad20bb6685a60aface4de19 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 16:12:16 +0000 Subject: [PATCH 11/14] measured excluded plus get geospatial nearby stuff working --- applications/modelling_e2e/handler.py | 20 ++++++++----------- .../local_handler/invoke_local_lambda.py | 2 +- .../generators/heating_recommendation.py | 12 ++++++++--- orchestration/modelling_orchestrator.py | 2 +- 4 files changed, 19 insertions(+), 17 deletions(-) diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index 95af5269..d45951e3 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -195,9 +195,7 @@ def handler(body: dict[str, Any], context: Any) -> None: ).fetchall() uprns: dict[int, int] = {int(row[0]): int(row[1]) for row in uprn_rows} - postcodes: dict[int, str] = { - int(row[0]): (row[1] or "") for row in postcode_rows - } + postcodes: dict[int, str] = {int(row[0]): (row[1] or "") for row in postcode_rows} overrides_reader = PropertyOverridesPostgresReader(lambda: Session(engine)) prediction_attrs_reader = OverrideBackedPredictionAttributesReader(overrides_reader) @@ -222,9 +220,7 @@ def handler(body: dict[str, Any], context: Any) -> None: try: uprn = uprns[property_id] postcode = postcodes.get(property_id, "") - logger.info( - f"property={property_id} uprn={uprn} postcode={postcode!r}" - ) + logger.info(f"property={property_id} uprn={uprn} postcode={postcode!r}") spatial = _spatial_for(geospatial, uprn) restrictions = ( @@ -283,16 +279,16 @@ def handler(body: dict[str, Any], context: Any) -> None: ).effective_epc solar_insights: Optional[dict[str, Any]] = ( - None - if no_solar - else _solar_insights_for(solar_client, spatial) + None if no_solar else _solar_insights_for(solar_client, spatial) ) # secondary_heating_removal is absent from the live material.type # enum; exclude unconditionally until the catalogue gap is resolved. - considered: Optional[frozenset[MeasureType]] = frozenset( - MeasureType - ) - {MeasureType.SECONDARY_HEATING_REMOVAL} + considered: Optional[frozenset[MeasureType]] = ( + frozenset(MeasureType) + - {MeasureType.SECONDARY_HEATING_REMOVAL} + - {MeasureType.SYSTEM_TUNE_UP_ZONED} + ) plan = run_modelling( effective_epc, diff --git a/applications/modelling_e2e/local_handler/invoke_local_lambda.py b/applications/modelling_e2e/local_handler/invoke_local_lambda.py index c667ffc6..b9a0d584 100644 --- a/applications/modelling_e2e/local_handler/invoke_local_lambda.py +++ b/applications/modelling_e2e/local_handler/invoke_local_lambda.py @@ -12,7 +12,7 @@ payload = { { "body": json.dumps( { - "property_ids": [709773, 709774], + "property_ids": [722473], "portfolio_id": 796, "scenario_id": 1268, "no_solar": False, diff --git a/domain/modelling/generators/heating_recommendation.py b/domain/modelling/generators/heating_recommendation.py index 475039df..9eae0353 100644 --- a/domain/modelling/generators/heating_recommendation.py +++ b/domain/modelling/generators/heating_recommendation.py @@ -284,6 +284,7 @@ def recommend_heating( epc: EpcPropertyData, products: ProductRepository, restrictions: PlanningRestrictions = PlanningRestrictions(), + considered_measures: Optional[frozenset[MeasureType]] = None, ) -> Optional[Recommendation]: """Return a "Heating & Hot Water" Recommendation of competing whole-system bundles for the dwelling, else None when no bundle is eligible. ASHP is @@ -302,7 +303,7 @@ def recommend_heating( if boiler_option is not None: options.append(boiler_option) - options.extend(_system_tune_up_options(epc, products)) + options.extend(_system_tune_up_options(epc, products, considered_measures)) if not options: return None @@ -310,7 +311,9 @@ def recommend_heating( def _system_tune_up_options( - epc: EpcPropertyData, products: ProductRepository + epc: EpcPropertyData, + products: ProductRepository, + considered_measures: Optional[frozenset[MeasureType]] = None, ) -> list[MeasureOption]: """The system tune-up options: keep the existing wet boiler but install better heating controls (standard 2106 and/or zone 2110, as competing @@ -338,7 +341,10 @@ def _system_tune_up_options( ), ) ) - if control_code not in _ZONE_CONTROL_CODES: + if control_code not in _ZONE_CONTROL_CODES and ( + considered_measures is None + or _SYSTEM_TUNE_UP_ZONED_MEASURE_TYPE in considered_measures + ): options.append( _tune_up_option( epc, diff --git a/orchestration/modelling_orchestrator.py b/orchestration/modelling_orchestrator.py index 55ae531d..e3180a4d 100644 --- a/orchestration/modelling_orchestrator.py +++ b/orchestration/modelling_orchestrator.py @@ -331,7 +331,7 @@ def _candidate_recommendations( MeasureType.SYSTEM_TUNE_UP, MeasureType.SYSTEM_TUNE_UP_ZONED, ), - lambda: recommend_heating(effective_epc, products, planning_restrictions), + lambda: recommend_heating(effective_epc, products, planning_restrictions, considered_measures), ), ( admitted(MeasureType.SECONDARY_HEATING_REMOVAL), From f1a2bbc4678948d946dc746d1d7469f48f85b204 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 16:13:10 +0000 Subject: [PATCH 12/14] list of properties when triggering locally --- scripts/trigger_modelling_e2e_sqs.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/scripts/trigger_modelling_e2e_sqs.py b/scripts/trigger_modelling_e2e_sqs.py index d220e799..8ab868c3 100644 --- a/scripts/trigger_modelling_e2e_sqs.py +++ b/scripts/trigger_modelling_e2e_sqs.py @@ -75,7 +75,9 @@ def main() -> None: f"no_solar={NO_SOLAR}, dry_run={DRY_RUN}) → {SQS_URL}" ) - sqs: Any = cast(Any, boto3.client("sqs")) # pyright: ignore[reportUnknownMemberType] + sqs: Any = cast( + Any, boto3.client("sqs") + ) # pyright: ignore[reportUnknownMemberType] sent = 0 for batch in _batches(ids, _BATCH_SIZE): entries = [ @@ -83,7 +85,7 @@ def main() -> None: "Id": str(uuid4()).replace("-", "")[:8] + str(i), "MessageBody": json.dumps( { - "property_id": pid, + "property_id": [pid], "portfolio_id": PORTFOLIO_ID, "scenario_id": SCENARIO_ID, "no_solar": NO_SOLAR, From 6b8eb88a38d03887168323a037753dde5c0b6971 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 16:17:30 +0000 Subject: [PATCH 13/14] increase lambda timeout to maximum --- deployment/terraform/lambda/modelling_e2e/main.tf | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/deployment/terraform/lambda/modelling_e2e/main.tf b/deployment/terraform/lambda/modelling_e2e/main.tf index e7b4e127..826e83aa 100644 --- a/deployment/terraform/lambda/modelling_e2e/main.tf +++ b/deployment/terraform/lambda/modelling_e2e/main.tf @@ -28,8 +28,8 @@ module "lambda" { batch_size = var.batch_size maximum_concurrency = var.maximum_concurrency - timeout = 60 - memory_size = 1024 + timeout = 900 + memory_size = 3008 environment = { STAGE = var.stage From 625edbefa471d36c30e0f657b6ea00efbdef7151 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Tue, 23 Jun 2026 08:26:19 +0000 Subject: [PATCH 14/14] trigger workflows --- scripts/trigger_modelling_e2e_sqs.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/trigger_modelling_e2e_sqs.py b/scripts/trigger_modelling_e2e_sqs.py index 8ab868c3..3bcfeaf9 100644 --- a/scripts/trigger_modelling_e2e_sqs.py +++ b/scripts/trigger_modelling_e2e_sqs.py @@ -10,7 +10,7 @@ AWS creds come from the ambient ~/.aws profile; DB creds from backend/.env. from __future__ import annotations -# --------------------------------------------------------------------------- +# -------------------------------------------------------------------------- # CONFIG — edit these before running # --------------------------------------------------------------------------- PORTFOLIO_ID: int = 785