various fixes

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(
+            logger.info(f"[dry_run] measures=[{measure_types}] — skipping DB write")
-                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,
@@ -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")

applications/modelling_e2e/local_handler/invoke_local_lambda.py

@@ -12,9 +12,9 @@ payload = {
         {
             "body": json.dumps(
                 {
-                    "property_id": 0,
+                    "property_id": 709634,
-                    "portfolio_id": 0,
+                    "portfolio_id": 785,
-                    "scenario_id": 0,
+                    "scenario_id": 1266,
                     "no_solar": False,
                     "dry_run": True,
                 }

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

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