"""
GUI helpers around the error/exclusion state of the workflow.
Mirrors the structure of manual_execution.py: a `create_*_controls()` factory for the
layout component and a `register_*_callbacks()` function for the Dash callbacks, so the
main dash app only has to place the control and register the callbacks.
The single button is context sensitive to the selected node:
* a container node -> toggles whether its steps are excluded from the schedule (immediate)
* a process step -> asks for confirmation, then puts that step (and its containers) into
the error state
* anything else -> the button is hidden
"""
from __future__ import annotations
from typing import Optional, TYPE_CHECKING
from dash import dcc, html, no_update
from dash.dependencies import Input, Output, State
from dash_extensions.enrich import DashProxy
from laborchestrator.logging_manager import StandardLogger as Logger
from laborchestrator.structures import SchedulingInstance, SMProcess
if TYPE_CHECKING:
from laborchestrator.gui.old_dash_app import SMDashApp
ERROR_STATE_BUTTON = "error_state_button"
ERROR_STATE_CONFIRM = "error_state_confirm"
ERROR_STATE_CAPTION = "Put into error-state"
[docs]
def create_error_state_controls() -> html.Div:
# context sensitive button (hidden until a container or process-step node is selected)
# plus the confirmation dialog shown before a step is put into error state manually
return html.Div(children=[
html.Button("", id=ERROR_STATE_BUTTON, n_clicks=0, style={"display": "none", "marginTop": "8px"}),
dcc.ConfirmDialog(id=ERROR_STATE_CONFIRM),
])
[docs]
def register_error_state_callbacks(app: DashProxy, dash_app: "SMDashApp") -> None:
@app.callback(
Output(ERROR_STATE_BUTTON, "style"),
Output(ERROR_STATE_BUTTON, "children"),
Input("wfg", "selected_node"),
)
def update_error_state_button(selected_node):
jssp = dash_app.sm_interface.jssp
if selected_node in jssp.container_info_by_name:
process = _process_of_container(jssp, selected_node)
excluded = process is not None and selected_node in process.labware_excluded_from_schedule
return _shown_style(), _exclusion_caption(excluded)
if selected_node in jssp.step_by_id:
return _shown_style(), ERROR_STATE_CAPTION
return {"display": "none", "marginTop": "8px"}, no_update
@app.callback(
Output(ERROR_STATE_CONFIRM, "message"),
Output(ERROR_STATE_CONFIRM, "displayed"),
Output(ERROR_STATE_BUTTON, "children"),
Input(ERROR_STATE_BUTTON, "n_clicks"),
State("wfg", "selected_node"),
)
def on_error_state_button(n_clicks, selected_node):
jssp = dash_app.sm_interface.jssp
if not n_clicks:
return no_update, no_update, no_update
# containers are toggled in/out of the schedule right away, no confirmation needed
if selected_node in jssp.container_info_by_name:
return no_update, no_update, _toggle_exclusion(dash_app, selected_node)
# putting a step into error state is destructive -> ask for confirmation, do nothing yet
if selected_node in jssp.step_by_id:
message = (f"Are you sure you want to set step {selected_node} into error mode manually? "
f"This will trigger an error report and prevent follow-up steps from execution.")
return message, True, no_update
return no_update, no_update, no_update
@app.callback(
Output("info_text", "children"),
Input(ERROR_STATE_CONFIRM, "submit_n_clicks"),
State("wfg", "selected_node"),
)
def confirm_error_state(submit_n_clicks, selected_node):
jssp = dash_app.sm_interface.jssp
if not submit_n_clicks or selected_node not in jssp.step_by_id:
return no_update
step = jssp.step_by_id[selected_node]
dash_app.sm_interface.worker.set_step_error(step, "manually set to error state via the GUI")
return f"Step {selected_node} was set into error mode manually."
[docs]
def _toggle_exclusion(dash_app: "SMDashApp", cont_name: str):
process = _process_of_container(dash_app.sm_interface.jssp, cont_name)
if process is None:
return no_update
excluded = cont_name in process.labware_excluded_from_schedule
if excluded:
process.labware_excluded_from_schedule.remove(cont_name)
else:
process.labware_excluded_from_schedule.append(cont_name)
# trigger a reschedule so the change takes effect
dash_app.sm_interface.schedule_manager.mark_schedule_invalid(enforce=True)
return _exclusion_caption(not excluded)
[docs]
def _process_of_container(jssp: SchedulingInstance, cont_name: str) -> Optional[SMProcess]:
process = next((p for p in jssp.process_by_name.values()
if any(cont.name == cont_name for cont in p.containers)), None)
if process is None:
Logger.warning(f"Container {cont_name} is not linked to a process")
return process
[docs]
def _exclusion_caption(excluded: bool) -> str:
return "Add to schedule" if excluded else "Remove from schedule"
[docs]
def _shown_style() -> dict:
return {"display": "inline-block", "marginTop": "8px"}
