From dc830fea63f19a95a8e6b75ad10439efd190064c Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 14:21:52 +0000 Subject: [PATCH] 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 ebea872e5..6b276f0c6 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 3d2379735..808200c62 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 51ee9050f..bac381847 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 82c7198ea..093d6bd6c 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})