mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
handler creates one child SubTask per property ID in the batch 🟩
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Daniel Roth
parent
2d2abc016b
commit
91c2d8a8fd
1 changed files with 24 additions and 28 deletions
|
|
@ -297,10 +297,14 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
|||
failures: list[dict[str, Any]] = []
|
||||
|
||||
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}")
|
||||
child = orchestrator.create_child_subtask(
|
||||
task_id, inputs={"property_id": property_id}
|
||||
)
|
||||
|
||||
def _work(pid: int = property_id) -> None:
|
||||
uprn = uprns[pid]
|
||||
postcode = postcodes.get(pid, "")
|
||||
logger.info(f"property={pid} uprn={uprn} postcode={postcode!r}")
|
||||
|
||||
spatial = _spatial_for(geospatial, uprn)
|
||||
restrictions = (
|
||||
|
|
@ -313,11 +317,11 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
|||
)
|
||||
|
||||
epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn)
|
||||
overrides = overlays_from(overrides_reader.overrides_for(property_id))
|
||||
overrides = overlays_from(overrides_reader.overrides_for(pid))
|
||||
predicted_epc: Optional[EpcPropertyData] = None
|
||||
|
||||
if epc is not None:
|
||||
logger.info(f"property={property_id} lodged EPC found")
|
||||
logger.info(f"property={pid} lodged EPC found")
|
||||
effective_epc = Property(
|
||||
identity=PropertyIdentity(
|
||||
portfolio_id=portfolio_id,
|
||||
|
|
@ -330,10 +334,10 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
|||
).effective_epc
|
||||
else:
|
||||
logger.info(
|
||||
f"property={property_id} no lodged EPC — attempting prediction"
|
||||
f"property={pid} no lodged EPC — attempting prediction"
|
||||
)
|
||||
predicted_epc = _predict_epc(
|
||||
property_id=property_id,
|
||||
property_id=pid,
|
||||
uprn=uprn,
|
||||
postcode=postcode,
|
||||
portfolio_id=portfolio_id,
|
||||
|
|
@ -361,9 +365,6 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
|||
landlord_overrides=overrides,
|
||||
).effective_epc
|
||||
|
||||
# Read-before-fetch: the Google Solar call is paid, so skip it
|
||||
# when this UPRN's insights are already persisted. Only a cache
|
||||
# miss hits Google — re-runs cost nothing for solar.
|
||||
solar_insights: Optional[dict[str, Any]]
|
||||
solar_was_fetched = False
|
||||
if no_solar:
|
||||
|
|
@ -374,9 +375,6 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
|||
solar_insights = _solar_insights_for(solar_client, spatial)
|
||||
solar_was_fetched = solar_insights is not None
|
||||
|
||||
# All Measure Types are considered: the off-catalogue overlay
|
||||
# (catalogue_with_off_catalogue_overrides) prices the measures the
|
||||
# live material catalogue cannot supply, so none need excluding.
|
||||
plan = run_modelling(
|
||||
effective_epc,
|
||||
planning_restrictions=restrictions,
|
||||
|
|
@ -387,7 +385,7 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
|||
print_table=False,
|
||||
)
|
||||
logger.info(
|
||||
f"property={property_id} modelling complete "
|
||||
f"property={pid} modelling complete "
|
||||
f"measures={len(plan.measures)}"
|
||||
)
|
||||
|
||||
|
|
@ -396,15 +394,15 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
|||
", ".join(m.measure_type for m in plan.measures) or "none"
|
||||
)
|
||||
logger.info(
|
||||
f"[dry_run] property={property_id} "
|
||||
f"[dry_run] property={pid} "
|
||||
f"measures=[{measure_types}] — skipping DB write"
|
||||
)
|
||||
continue
|
||||
return
|
||||
|
||||
with PostgresUnitOfWork(lambda: Session(engine)) as uow:
|
||||
if epc is not None:
|
||||
uow.epc.save(
|
||||
epc, property_id=property_id, portfolio_id=portfolio_id
|
||||
epc, property_id=pid, portfolio_id=portfolio_id
|
||||
)
|
||||
elif predicted_epc is not None:
|
||||
# Persist the synthesised EPC in the predicted slot (ADR-0031),
|
||||
|
|
@ -412,7 +410,7 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
|||
# the picture the Plan was modelled from.
|
||||
uow.epc.save(
|
||||
predicted_epc,
|
||||
property_id=property_id,
|
||||
property_id=pid,
|
||||
portfolio_id=portfolio_id,
|
||||
source="predicted",
|
||||
)
|
||||
|
|
@ -432,24 +430,22 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
|||
)
|
||||
uow.plan.save(
|
||||
plan,
|
||||
property_id=property_id,
|
||||
property_id=pid,
|
||||
scenario_id=scenario_id,
|
||||
portfolio_id=portfolio_id,
|
||||
is_default=scenario.is_default,
|
||||
)
|
||||
uow.property.mark_modelled(
|
||||
property_id, has_recommendations=bool(plan.measures)
|
||||
pid, has_recommendations=bool(plan.measures)
|
||||
)
|
||||
uow.commit()
|
||||
logger.info(f"property={property_id} plan saved")
|
||||
logger.info(f"property={pid} plan saved")
|
||||
|
||||
# Baseline Performance is re-established from the persisted EPC
|
||||
# (lodged or predicted), so it runs after the Plan UoW commits. By
|
||||
# here the property always has a persisted EPC — a property that
|
||||
# could be neither fetched nor predicted raised earlier.
|
||||
baseline_orchestrator.run([property_id])
|
||||
logger.info(f"property={property_id} baseline saved")
|
||||
baseline_orchestrator.run([pid])
|
||||
logger.info(f"property={pid} baseline saved")
|
||||
|
||||
try:
|
||||
orchestrator.run_subtask(child.id, work=_work)
|
||||
except Exception as error: # noqa: BLE001
|
||||
logger.error(
|
||||
f"property={property_id} uprn={uprns.get(property_id)}: "
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue