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