feat: replace direct AHM callback activity with external workflow signaling for simulation completion

main
Cizz22 1 month ago
parent aeb8362ece
commit 372ff30a93

@ -5,6 +5,7 @@ import os
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
from temporal.activity import calculate_plant_eaf_activity, execute_simulation_activity, update_contribution_bulk_mappings_activity, update_equipment_for_simulation_activity,call_callback_ahm, acquire_lock, release_lock,execute_update_node from temporal.activity import calculate_plant_eaf_activity, execute_simulation_activity, update_contribution_bulk_mappings_activity, update_equipment_for_simulation_activity,call_callback_ahm, acquire_lock, release_lock,execute_update_node
TEMPORAL_URL = os.environ.get("TEMPORAL_URL", "http://192.168.1.86:7233") TEMPORAL_URL = os.environ.get("TEMPORAL_URL", "http://192.168.1.86:7233")
@workflow.defn @workflow.defn
@ -68,14 +69,26 @@ class SimulationWorkflow:
if sim_data.get("AhmJobId"): if sim_data.get("AhmJobId"):
await workflow.execute_activity( job_id = sim_data["AhmJobId"]
call_callback_ahm, sim_id = sim_data["HubCnnId"]
{ if "|" in job_id:
"simulation_id": sim_data["HubCnnId"], wf_id, sim_type = job_id.split("|", 1)
"job_id": sim_data["AhmJobId"] else:
}, wf_id = job_id
start_to_close_timeout=timedelta(days=1) sim_type = "unknown"
)
try:
handle = workflow.get_external_workflow_handle(wf_id)
await handle.signal(
"rbd_simulation_completed",
{
"sim_id": sim_id,
"type": sim_type,
"status": "completed"
}
)
except Exception as e:
workflow.logger.error(f"Failed to signal OpsDock: {e}")
self.status = "COMPLETED" self.status = "COMPLETED"
# etc… # etc…

Loading…
Cancel
Save