"""
Fully AI generated module that enables call of execute_process_step() of the worker on an individual process step in the workflow regardles of the processes state (it is forbidden while process is running) or the schedule.
It does not eecute process_step_finished nor change the state of the process or process step.
Its meant to enable testing individual steps through the worker without running full workflows or creating minimal ones.
"""
from __future__ import annotations
import traceback
from dataclasses import dataclass
from datetime import datetime
from threading import Thread
import time
from typing import Dict, List, Optional, TYPE_CHECKING, Tuple, TypedDict, Union
from dash import callback_context, dcc, html, no_update
from dash.development.base_component import Component
from dash.dependencies import Input, Output, State
from dash_extensions.enrich import DashProxy
from sila2.framework import CommandExecutionStatus
from laborchestrator.logging_manager import StandardLogger as Logger
from laborchestrator.orchestrator_implementation import Orchestrator
from laborchestrator.orchestrator_interface import ProcessExecutionState
from laborchestrator.structures import MoveStep, ProcessStep, SMProcess, StepStatus
from laborchestrator.workflowgraph import parse_job_shop_config
if TYPE_CHECKING:
from laborchestrator.engine.worker_interface import Observable
from laborchestrator.gui.old_dash_app import SMDashApp
StyleValue = Union[str, int, float, bool, None]
Style = Dict[str, StyleValue]
CallbackReturn = Tuple[object, ...]
[docs]
class _DeviceOption(TypedDict):
label: str
value: str
[docs]
class _ManualExecutionRequest(TypedDict):
step_id: str
device: Optional[str]
requires_device_choice: bool
device_options: List[_DeviceOption]
MANUAL_EXECUTE_BUTTON = "manual_execute"
MANUAL_EXECUTE_MODAL = "manual_execute_modal"
MANUAL_EXECUTE_MESSAGE = "manual_execute_message"
MANUAL_EXECUTE_DEVICE_ROW = "manual_execute_device_row"
MANUAL_EXECUTE_DEVICE = "manual_execute_device"
MANUAL_EXECUTE_CONFIRM = "manual_execute_confirm"
MANUAL_EXECUTE_CANCEL = "manual_execute_cancel"
MANUAL_EXECUTE_PENDING = "manual_execute_pending"
MODAL_STYLE: Style = {
"display": "none",
"position": "fixed",
"zIndex": 1000,
"left": 0,
"top": 0,
"width": "100%",
"height": "100%",
"backgroundColor": "rgba(0, 0, 0, 0.35)",
}
MODAL_CONTENT_STYLE: Style = {
"backgroundColor": "white",
"margin": "12% auto",
"padding": "16px",
"border": "1px solid #888",
"width": "380px",
"boxShadow": "0 4px 12px rgba(0, 0, 0, 0.25)",
}
DEVICE_ROW_STYLE: Style = {"display": "none", "marginTop": "12px"}
[docs]
@dataclass
class _StepState:
status: StepStatus
start: Optional[datetime]
finish: Optional[datetime]
result: object
duration: float
[docs]
@classmethod
def from_step(cls, step: ProcessStep) -> "_StepState":
return cls(
status=step.status,
start=step.start,
finish=step.finish,
result=step.result,
duration=step.duration,
)
[docs]
def restore(self, step: ProcessStep) -> None:
step.status = self.status
step.start = self.start
step.finish = self.finish
step.result = self.result
step.duration = self.duration
[docs]
def create_manual_execution_controls() -> Component:
return html.Div(
children=[
html.Button("Execute selected step", id=MANUAL_EXECUTE_BUTTON, n_clicks=0),
dcc.Store(id=MANUAL_EXECUTE_PENDING),
html.Div(
id=MANUAL_EXECUTE_MODAL,
style=_modal_style(visible=False),
children=[
html.Div(
style=MODAL_CONTENT_STYLE,
children=[
html.H4("Confirm manual execution", style={"marginTop": 0}),
html.Div(id=MANUAL_EXECUTE_MESSAGE),
html.Div(
id=MANUAL_EXECUTE_DEVICE_ROW,
style=DEVICE_ROW_STYLE,
children=[
html.Div("Device:"),
dcc.Dropdown(
id=MANUAL_EXECUTE_DEVICE,
options=[],
value=None,
clearable=False,
style={"width": "100%"},
),
],
),
html.Div(
children=[
html.Button("Cancel", id=MANUAL_EXECUTE_CANCEL, n_clicks=0),
html.Button(
"Execute",
id=MANUAL_EXECUTE_CONFIRM,
n_clicks=0,
style={"marginLeft": "8px"},
),
],
style={"marginTop": "16px", "textAlign": "right"},
),
],
)
],
),
]
)
[docs]
def register_manual_execution_callbacks(app: DashProxy, dash_app: SMDashApp) -> None:
@app.callback(
Output(MANUAL_EXECUTE_MODAL, "style"),
Output(MANUAL_EXECUTE_MESSAGE, "children"),
Output(MANUAL_EXECUTE_DEVICE_ROW, "style"),
Output(MANUAL_EXECUTE_DEVICE, "options"),
Output(MANUAL_EXECUTE_DEVICE, "value"),
Output(MANUAL_EXECUTE_PENDING, "data"),
Output("info_text", "children"),
Input(MANUAL_EXECUTE_BUTTON, "n_clicks"),
Input(MANUAL_EXECUTE_CONFIRM, "n_clicks"),
Input(MANUAL_EXECUTE_CANCEL, "n_clicks"),
State(MANUAL_EXECUTE_PENDING, "data"),
State(MANUAL_EXECUTE_DEVICE, "value"),
State("wfg", "selected_node"),
)
def handle_manual_execution(
_open_clicks: Optional[int],
_confirm_clicks: Optional[int],
_cancel_clicks: Optional[int],
pending: Optional[_ManualExecutionRequest],
selected_device: Optional[str],
selected_node: Optional[str]) -> CallbackReturn:
triggered_id = _triggered_id()
if triggered_id == MANUAL_EXECUTE_BUTTON:
return _open_confirmation(dash_app, selected_node)
if triggered_id == MANUAL_EXECUTE_CANCEL:
return _closed_outputs(info_text="Manual execution cancelled.")
if triggered_id == MANUAL_EXECUTE_CONFIRM:
return _confirm_execution(dash_app, pending, selected_device)
return no_update, no_update, no_update, no_update, no_update, no_update, no_update
[docs]
def _open_confirmation(dash_app: SMDashApp, selected_node: Optional[str]) -> CallbackReturn:
step_id = selected_node or dash_app.selected_operation
request, error = _build_request(dash_app.sm_interface, step_id)
if error:
return _closed_outputs(info_text=error)
if request["requires_device_choice"]:
options = request["device_options"]
return (
_modal_style(visible=True),
f"Execute {request['step_id']} immediately?",
{"display": "block", "marginTop": "12px"},
options,
options[0]["value"],
request,
no_update,
)
return (
_modal_style(visible=True),
f"Execute {request['step_id']} immediately on {request['device']}?",
DEVICE_ROW_STYLE,
[],
None,
request,
no_update,
)
[docs]
def _confirm_execution(
dash_app: SMDashApp,
pending: Optional[_ManualExecutionRequest],
selected_device: Optional[str]) -> CallbackReturn:
if not pending:
return _closed_outputs(info_text="No manual execution is pending.")
step_id = pending["step_id"]
device = pending["device"]
if pending["requires_device_choice"]:
device = _clean_device_name(selected_device)
if not device:
return (
_modal_style(visible=True),
f"Execute {step_id} immediately?",
{"display": "block", "marginTop": "12px"},
pending["device_options"],
selected_device,
pending,
"Choose a device before executing the step.",
)
error = _execute_immediately(dash_app.sm_interface, step_id, device)
if error:
return _closed_outputs(info_text=error)
return _closed_outputs(info_text=f"Manual execution started for {step_id} on {device}.")
[docs]
def _build_request(
orchestrator: Orchestrator,
step_id: Optional[str]) -> Tuple[Optional[_ManualExecutionRequest], Optional[str]]:
if not step_id:
return None, "Select a process step first."
if step_id not in orchestrator.jssp.step_by_id:
return None, f"Selected node {step_id} is not a process step."
step = orchestrator.jssp.step_by_id[step_id]
process = orchestrator.jssp.process_by_name.get(step.process_name)
if process is None:
return None, f"Process {step.process_name} was not found for {step_id}."
if _process_state(process) == ProcessExecutionState.RUNNING:
return None, f"Process {step.process_name} is currently running."
main_device = step.main_device
if main_device is None:
return None, f"Step {step_id} has no main device."
preferred = _clean_device_name(main_device.preferred)
if preferred:
return {
"step_id": step_id,
"device": preferred,
"requires_device_choice": False,
"device_options": [],
}, None
options = _device_options(orchestrator, step)
if not options:
return None, f"No available device was found for {step_id}."
return {
"step_id": step_id,
"device": None,
"requires_device_choice": True,
"device_options": options,
}, None
[docs]
def _wait_for_manual_execution(
orchestrator: Orchestrator,
step_id: str,
observable: Observable,
step_state: _StepState,
process: SMProcess,
process_state: ProcessExecutionState) -> None:
try:
while not observable.done:
time.sleep(1)
Logger.info(f"Manual execution finished for {step_id} with result: {observable.status}")
if observable.status == CommandExecutionStatus.finishedSuccessfully:
orchestrator.worker.process_step_finished(step_id, observable.status)
except Exception as ex:
Logger.error(f"Manual execution follow-up failed for {step_id}: {ex}\n{traceback.format_exc()}")
finally:
if step_id in orchestrator.jssp.step_by_id:
step_state.restore(orchestrator.jssp.step_by_id[step_id])
if process.name in orchestrator.jssp.process_by_name:
process.status = process_state
[docs]
def _device_options(orchestrator: Orchestrator, step: ProcessStep) -> List[_DeviceOption]:
main_device = step.main_device
device_type = _device_type_name(main_device.device_type)
lab_config_file = getattr(orchestrator.schedule_manager, "lab_config_file", None)
options = []
if lab_config_file:
try:
for device in parse_job_shop_config(lab_config_file):
if _same_device_type(device.get("Type"), device_type):
device_name = str(device["Name"])
options.append({"label": device_name, "value": device_name})
except Exception as ex:
Logger.warning(f"Could not read lab config for manual execution: {ex}")
current_name = _clean_device_name(main_device.name)
if current_name and all(option["value"] != current_name for option in options):
options.append({"label": current_name, "value": current_name})
return options
[docs]
def _triggered_id() -> Optional[str]:
if not callback_context.triggered:
return None
return callback_context.triggered[0]["prop_id"].split(".")[0]
[docs]
def _closed_outputs(info_text: object) -> CallbackReturn:
return _modal_style(visible=False), no_update, DEVICE_ROW_STYLE, [], None, None, info_text
[docs]
def _modal_style(visible: bool) -> Style:
style = dict(MODAL_STYLE)
style["display"] = "block" if visible else "none"
return style
[docs]
def _process_state(process: SMProcess) -> ProcessExecutionState:
if hasattr(process, "_status"):
return process._status
return process.status
[docs]
def _device_type_name(device_type: object) -> str:
return getattr(device_type, "__name__", str(device_type))
[docs]
def _same_device_type(config_type: object, step_type: str) -> bool:
config_type = str(config_type)
return config_type == step_type or config_type.endswith("." + step_type)
[docs]
def _clean_device_name(device: object) -> Optional[str]:
if device is None:
return None
device = str(device).strip()
if not device or device == "None":
return None
return device
