feat: migrate EAF forecasting to Temporal workflows and add simulation result retrieval to dashboard service

src/aeros_simulation/router.py

@@ -497,13 +497,13 @@ async def get_simulation_report(db_session: DbSession, simulation_id: str):
     )
 
 
-@router.get("/forecast/eaf", response_model=StandardResponse[dict])
+@router.get("/forecast/eaf", response_model=StandardResponse[str])
 async def get_forecast_eaf(
     db_session: DbSession,
     time_range: str = Query(..., alias="range", regex="^(weekly|monthly)$"),
 ):
     """
-    Generate a forecasted EAF value for the next week or month.
+    Trigger a forecasted EAF simulation for the next week or month using Temporal.
     Calculates offset from 'Last Overhaul Date' in master data.
     """
     from src.dashboard_model.service import _get_or_create_master_data
@@ -528,19 +528,19 @@ async def get_forecast_eaf(
         oh_date = datetime(now.year, 1, 1).replace(tzinfo=None)
 
     # 3. Calculate Age in Hours (Offset)
-    # Using the helper from utils.py
     offset_hours = hours_between(oh_date, now)
 
-    # start date is oh + offset
-    start_date = oh_date + timedelta(hours=offset_hours)
-
     # 4. Define Forecast Window
     duration = 168 if time_range == "weekly" else 720
+    
+    start_date = oh_date + timedelta(hours=offset_hours)
+    end_date = start_date + timedelta(hours=duration)
+    project = await get_project(db_session=db_session)
 
     # 5. Create Simulation Request
     sim_in = SimulationInput(
         SchematicName="- TJB - Unit 3 -",
-        SimulationName=f"Simulation{time_range}_{now.strftime('%Y%m%d')}",
+        SimulationName=f"Simulation_{time_range}_{now.strftime('%Y%m%d')}",
         SimDuration=duration,
         DurationUnit="UHour",
         OffSet=offset_hours,
@@ -549,52 +549,37 @@ async def get_forecast_eaf(
         IsDefault=False
     )
 
-    # 6. Run simulation (blocking call to engine)
+    # 6. Initialize Simulation in DB
-    try:
+    simulation = await create_simulation(
-        raw_result = await execute_simulation(
+        db_session=db_session, 
-            db_session=db_session,
+        simulation_in=sim_in, 
-            sim_data=sim_in.model_dump(),
+        current_user=None
-            is_saved=False
+    )
-        )
+    simulation_id = simulation.id
-        
-        # 7. Extract Root Node Results
-        node_results = raw_result.get("nodeResultOuts", [])
-        plant_result = next((r for r in node_results if r.get("nodeName") == "- TJB - Unit 3 -"), None)
-        
-        if not plant_result:
-             # Fallback to any schematic result if name doesn't match perfectly
-             plant_result = next((r for r in node_results if r.get("nodeType") == "Schematic"), None)
-
-        if not plant_result:
-            raise HTTPException(status_code=500, detail="Plant node results not found in simulation output")
-
-        # 8. Return predicted metrics
-        eaf = plant_result.get("eaf")
-        total_dt = plant_result.get("totalDowntime", 0)
-        total_ut = plant_result.get("totalUpTime", 0)
-        efor = (total_dt / (total_dt + total_ut)) * 100 if (total_dt + total_ut) > 0 else 0
-
-        forecast_msg = (
-            f"Forecast generated successfully, "
-            f"start date: {start_date.strftime('%Y-%m-%d %H:%M:%S')}, "
-            f"end_date: {(start_date + timedelta(hours=duration)).strftime('%Y-%m-%d %H:%M:%S')}"
-        )
 
-        return StandardResponse(
+    # 7. Start Temporal Workflow
-            data={
+    temporal_client = await Client.connect(TEMPORAL_URL)
-                "range": time_range,
+
-                "eaf_predicted": float(eaf) if eaf is not None else 0.0,
+    sim_data = sim_in.model_dump()
-                "efor_predicted": float(efor),
+    sim_data["HubCnnId"] = str(simulation_id)
-                "reliability_age_hours": offset_hours,
+    sim_data["projectName"] = project.project_name
-                "sim_duration_hours": duration,
+
-                "reference_oh_date": oh_date_str
+    await temporal_client.start_workflow(
-            },
+        SimulationWorkflow.run,
-            message=forecast_msg
+        sim_data,
-        )
+        id=f"simulation-{simulation_id}",
-    except Exception as e:
+        task_queue="simulation-task-queue",
-        import logging
+    )
-        logging.getLogger(__name__).error(f"Forecast failed: {str(e)}")
+
-        raise HTTPException(status_code=500, detail=f"Simulation failed: {str(e)}")
+    forecast_msg = (
+        f"Forecast simulation started via Temporal. "
+        f"Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}"
+    )
+
+    return StandardResponse(
+        data=str(simulation_id),
+        message=forecast_msg
+    )
     
     
 

src/aeros_simulation/service.py

@@ -997,10 +997,10 @@ def convert_id_to_none_if_negative(value):
     return None if value < 0 else value
 
 
-async def create_simulation(*, db_session: DbSession, simulation_in: SimulationInput, current_user):
+async def create_simulation(*, db_session: DbSession, simulation_in: SimulationInput, current_user=None):
     """Create a new simulation."""
     input = simulation_in.model_dump(exclude={"SimulationName"})
-    user_id = current_user.user_id
+    user_id = current_user.user_id if current_user else None
     
     # Check if is default
     if simulation_in.IsDefault:

src/dashboard_model/service.py

@@ -129,6 +129,10 @@ async def get_model_data(*, db_session: DbSession, simulation_id: Optional[UUID]
         "SSB": 98,
         "WTP": 98,
     }
+    
+    # Fetch Forecasts
+    forecast_weekly = await get_latest_sim_results_by_pattern(db_session, "Simulation_weekly_%")
+    forecast_monthly = await get_latest_sim_results_by_pattern(db_session, "Simulation_monthly_%")
 
     return {
     "id": str(simulation.id), 
@@ -152,5 +156,49 @@ async def get_model_data(*, db_session: DbSession, simulation_id: Optional[UUID]
          "Prediction_Derating": Derating,
          "Real_Derating": Derating
      },
-    "powerplant_reliability": powerplant_reliability
+    "powerplant_reliability": powerplant_reliability,
+    "forecast_weekly": forecast_weekly,
+    "forecast_monthly": forecast_monthly
 }
+
+async def get_latest_sim_results_by_pattern(db_session: DbSession, pattern: str):
+    from sqlalchemy import select
+    from src.aeros_simulation.model import AerosSimulation, AerosSimulationCalcResult, AerosNode
+    from src.aeros_simulation.service import get_plant_calc_result
+
+    query = select(AerosSimulation).where(AerosSimulation.simulation_name.like(pattern)).order_by(AerosSimulation.created_at.desc()).limit(1)
+    result = await db_session.execute(query)
+    simulation = result.scalars().first()
+    
+    if not simulation:
+        return None
+        
+    # Get plant node result
+    plant_res = await get_plant_calc_result(db_session=db_session, simulation_id=simulation.id)
+    
+    # Get top 5 equipment causing trips (num_events)
+    trip_query = select(AerosSimulationCalcResult).join(AerosNode).where(
+        AerosSimulationCalcResult.aeros_simulation_id == simulation.id,
+        AerosNode.node_type == 'RegularNode'
+    ).order_by(AerosSimulationCalcResult.num_events.desc()).limit(5)
+    
+    trip_res_exec = await db_session.execute(trip_query)
+    trip_results = trip_res_exec.scalars().all()
+    
+    top_trips = []
+    for tr in trip_results:
+        if tr.num_events > 0:
+            top_trips.append({
+                "name": tr.aeros_node.node_name,
+                "events": tr.num_events,
+                "tag": tr.aeros_node.node_name
+            })
+            
+    return {
+        "simulation_id": str(simulation.id),
+        "name": simulation.simulation_name,
+        "eaf": plant_res.eaf if plant_res else 0,
+        "efor": plant_res.efor if plant_res else 0,
+        "trips": int(plant_res.num_events) if plant_res else 0,
+        "top_trip_equipment": top_trips
+    }