mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-30 13:10:47 +00:00
Merge pull request #1253 from Hestia-Homes/feature/trigger-e2e-lamnda
e2e lambda
This commit is contained in:
Daniel Roth
GitHub
commit
04aa5db1e2
21 changed files with 1459 additions and 4 deletions
24
applications/modelling_e2e/Dockerfile
Normal file
24
applications/modelling_e2e/Dockerfile
Normal file
|
|
@ -0,0 +1,24 @@
|
|||
FROM public.ecr.aws/lambda/python:3.11
|
||||
|
||||
WORKDIR /var/task
|
||||
|
||||
COPY applications/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 applications/ applications/
|
||||
|
||||
CMD ["applications.modelling_e2e.handler.handler"]
|
||||
358
applications/modelling_e2e/handler.py
Normal file
358
applications/modelling_e2e/handler.py
Normal file
|
|
@ -0,0 +1,358 @@
|
|||
"""SQS-triggered Lambda: fetch EPC (or predict) → run modelling → persist plan.
|
||||
|
||||
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
|
||||
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 collections.abc import Callable
|
||||
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 (
|
||||
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
|
||||
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 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.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.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
|
||||
_cohort_cache: dict[str, list[ComparableProperty]] = {}
|
||||
|
||||
logger = setup_logger()
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
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_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_ids={property_ids} 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())
|
||||
solar_client = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"])
|
||||
|
||||
with engine.connect() as conn:
|
||||
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()
|
||||
|
||||
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))
|
||||
prediction_attrs_reader = OverrideBackedPredictionAttributesReader(overrides_reader)
|
||||
comparables_repo = EpcComparablePropertiesRepository(epc_client, geospatial)
|
||||
predictor = EpcPrediction()
|
||||
|
||||
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]
|
||||
|
||||
read_session = Session(engine)
|
||||
try:
|
||||
scenario = ScenarioPostgresRepository(read_session).get_many([scenario_id])[0]
|
||||
products = ProductPostgresRepository(read_session)
|
||||
|
||||
errors: list[int] = []
|
||||
|
||||
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}")
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
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}
|
||||
- {MeasureType.SYSTEM_TUNE_UP_ZONED}
|
||||
)
|
||||
|
||||
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()
|
||||
36
applications/modelling_e2e/local_handler/.env.local.example
Normal file
36
applications/modelling_e2e/local_handler/.env.local.example
Normal file
|
|
@ -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)
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
services:
|
||||
modelling-e2e:
|
||||
build:
|
||||
context: ../../../
|
||||
dockerfile: applications/modelling_e2e/Dockerfile
|
||||
ports:
|
||||
- "9004:8080"
|
||||
env_file:
|
||||
- ../../../.env
|
||||
|
|
@ -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_ids": [722473],
|
||||
"portfolio_id": 796,
|
||||
"scenario_id": 1268,
|
||||
"no_solar": False,
|
||||
"dry_run": False,
|
||||
}
|
||||
)
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
response = requests.post(LAMBDA_URL, json=payload)
|
||||
|
||||
print("Status code:", response.status_code)
|
||||
print("Response:")
|
||||
print(response.text)
|
||||
11
applications/modelling_e2e/modelling_e2e_trigger_body.py
Normal file
11
applications/modelling_e2e/modelling_e2e_trigger_body.py
Normal file
|
|
@ -0,0 +1,11 @@
|
|||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
|
||||
class ModellingE2ETriggerBody(BaseModel):
|
||||
model_config = ConfigDict(extra="allow")
|
||||
|
||||
property_ids: list[int]
|
||||
portfolio_id: int
|
||||
scenario_id: int
|
||||
no_solar: bool = False
|
||||
dry_run: bool = False
|
||||
11
applications/modelling_e2e/requirements.txt
Normal file
11
applications/modelling_e2e/requirements.txt
Normal file
|
|
@ -0,0 +1,11 @@
|
|||
awslambdaric
|
||||
boto3
|
||||
numpy==2.1.2
|
||||
pandas==2.2.2
|
||||
pyarrow==17.0.0
|
||||
httpx
|
||||
requests
|
||||
pydantic
|
||||
sqlalchemy==2.0.36
|
||||
sqlmodel
|
||||
psycopg2-binary==2.9.10
|
||||
51
deployment/terraform/lambda/modelling_e2e/main.tf
Normal file
51
deployment/terraform/lambda/modelling_e2e/main.tf
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
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
|
||||
maximum_concurrency = var.maximum_concurrency
|
||||
|
||||
timeout = 900
|
||||
memory_size = 3008
|
||||
|
||||
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
|
||||
}
|
||||
9
deployment/terraform/lambda/modelling_e2e/outputs.tf
Normal file
9
deployment/terraform/lambda/modelling_e2e/outputs.tf
Normal file
|
|
@ -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"
|
||||
}
|
||||
20
deployment/terraform/lambda/modelling_e2e/provider.tf
Normal file
20
deployment/terraform/lambda/modelling_e2e/provider.tf
Normal file
|
|
@ -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"
|
||||
}
|
||||
65
deployment/terraform/lambda/modelling_e2e/variables.tf
Normal file
65
deployment/terraform/lambda/modelling_e2e/variables.tf
Normal file
|
|
@ -0,0 +1,65 @@
|
|||
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 = "Reserved concurrency for the Lambda. -1 = unreserved."
|
||||
}
|
||||
|
||||
variable "maximum_concurrency" {
|
||||
type = number
|
||||
default = 2
|
||||
description = "Maximum concurrent Lambda invocations from the SQS trigger."
|
||||
}
|
||||
|
||||
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}"
|
||||
}
|
||||
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ class TaskStatus(str, Enum):
|
|||
class Source(str, Enum):
|
||||
PORTFOLIO = "portfolio_id"
|
||||
HUBSPOT_DEAL = "hubspot_deal_id"
|
||||
PROPERTY = "property_id"
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
105
scripts/trigger_modelling_e2e_sqs.py
Normal file
105
scripts/trigger_modelling_e2e_sqs.py
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
"""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()
|
||||
0
tests/applications/modelling_e2e/__init__.py
Normal file
0
tests/applications/modelling_e2e/__init__.py
Normal file
679
tests/applications/modelling_e2e/test_handler.py
Normal file
679
tests/applications/modelling_e2e/test_handler.py
Normal file
|
|
@ -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()
|
||||
0
tests/lambdas/__init__.py
Normal file
0
tests/lambdas/__init__.py
Normal file
0
tests/lambdas/modelling_e2e/__init__.py
Normal file
0
tests/lambdas/modelling_e2e/__init__.py
Normal file
|
|
@ -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})
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue