diff --git a/src/aeros_simulation/schema.py b/src/aeros_simulation/schema.py index 9b06fce..f29e605 100644 --- a/src/aeros_simulation/schema.py +++ b/src/aeros_simulation/schema.py @@ -28,6 +28,7 @@ class SimulationInput(DefultBase): AhmJobId: Optional[str] = Field(None, max_length=50) deratings: Optional[List[dict]] = [] maintenance_outages: Optional[List[dict]] = [] + CallbackWorkflowId: Optional[str] = Field(None, max_length=100) class SimulationNode(DefultBase): id: UUID diff --git a/temporal/workflow.py b/temporal/workflow.py index 3f5b2c7..47cd05e 100644 --- a/temporal/workflow.py +++ b/temporal/workflow.py @@ -1,10 +1,11 @@ from datetime import timedelta from temporalio import workflow - +from temporalio.client import Client +import os 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 - +TEMPORAL_URL = os.environ.get("TEMPORAL_URL", "http://192.168.1.86:7233") @workflow.defn class SimulationWorkflow: @@ -64,8 +65,8 @@ class SimulationWorkflow: sim_data["HubCnnId"], start_to_close_timeout=timedelta(days=1) ) - - + + if sim_data.get("AhmJobId"): await workflow.execute_activity( call_callback_ahm, @@ -79,9 +80,21 @@ class SimulationWorkflow: self.status = "COMPLETED" # etc… self.isDone = True + await workflow.execute_activity( release_lock, sim_data["HubCnnId"], start_to_close_timeout=timedelta(minutes=10), ) + + if sim_data.get("CallbackWorkflowId"): + client = await Client.connect(TEMPORAL_URL) + handle = client.get_workflow_handle(sim_data["CallbackWorkflowId"]) + + await handle.signal("notify_done", { + "simulation_id": sim_data["HubCnnId"], + "status": "success" + }) + + return sim_data["HubCnnId"] # simulation_id