new application to trigger e2e for a single property and scenario

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
+}

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"
+}

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"
+}

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}"
+}

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
+}
+

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

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"]

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()

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

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()