import traceback
from typing import Any
from dash import dcc, html, no_update
import time
from laborchestrator.logging_manager import StandardLogger as Logger
from dash.dependencies import State, Input, Output
from dash_extensions.enrich import MultiplexerTransform, DashProxy
from .traffic_light import TrafficLight, register_traffic_light_callbacks
from .side_panel import create_side_panel
from .manual_execution import create_manual_execution_controls, register_manual_execution_callbacks
from .error_state_utilities import create_error_state_controls, register_error_state_callbacks
import dash_interactive_graphviz
from threading import Thread
from laborchestrator.structures import Variable, StepStatus
from laborchestrator.pythonlab_reader import PythonLabReader
from laborchestrator.orchestrator_interface import OrchestratorInterface
from laborchestrator.orchestrator_implementation import Orchestrator
import sys
import importlib.util
from pathlib import Path
tests_dir = Path(__file__).resolve().parent.parent.parent / "tests"
sys.path.append(str(tests_dir))
try:
import test_data
except ImportError:
Logger.warning("Could not import test_datas")
test_data = None
block_style = {
'display': 'flex',
'flexDirection': 'column', # Stack children vertically
'alignItems': 'center', # Center align vertically in the block
'justifyContent': 'center', # Optional: center align content within block
'verticalAlign': 'top', # Align block with rest of the app
}
[docs]
def create_timestamp() -> str:
lct = time.localtime()
y, mo, d, h, mi, s = [str(lct.__getattribute__(v)).rjust(2, '0') for v in
['tm_year', 'tm_mon', 'tm_mday', 'tm_hour', 'tm_min', 'tm_sec']]
timestamp = f"{y}{mo}{d}_{h}{mi}{s}"
return timestamp
[docs]
class SMDashApp:
# sm_interface: OrchestratorInterface
sm_interface: Orchestrator # bit hacky, but makes it easier to program. This GUI will be replaced anyway
app: DashProxy
p: Thread
selected_operation: str
[docs]
def _reload_process_module(self,):
"""
Drop the process package and all of its submodules from the module cache and import it again.
ProcessFinder imports the individual process modules with importlib.import_module, which is a
cache hit for everything already loaded. Purging the cache first makes that import re-execute
the files, so edits to create_resources and init_service_resources reach processes added from
now on. Processes that were added earlier keep the class objects they were built from and are
not affected.
"""
module_name = getattr(self.process_module, "__name__", None)
if not module_name:
return
stale = {
name: module for name, module in sys.modules.items()
if name == module_name or name.startswith(f"{module_name}.")
}
for name in stale:
del sys.modules[name]
try:
self.process_module = importlib.import_module(module_name)
except Exception:
Logger.warning(f"Could not reload process module {module_name}:\n{traceback.format_exc()}")
sys.modules.update(stale)
self.process_module = stale[module_name]
@property
def importable_processes(self) -> list[str]:
# refresh the modules, so changes in the process apply
self._reload_process_module()
self.importable_process_by_name = PythonLabReader.get_processes_from_module(self.process_module)
return list(self.importable_process_by_name.keys())
def __init__(self, sm_interface: OrchestratorInterface, port=8050, process_module=test_data, arm_open: bool = True,
db_open: bool = True, browser_open: bool = True):
self.sm_interface = sm_interface
self.process_module = process_module
self.importable_process_by_name = {}
self.test_view = False
self.process_to_add = None
self.process_type = {}
#self.last_jssp = {idx: job.status for idx, job in sm_interface.jssp.definite_job_by_id.items()}
assets_dir = Path(__file__).resolve().parent / "assets"
self.app = DashProxy("Orchestrator webinterface", prevent_initial_callbacks=True,
suppress_callback_exceptions=True, transforms=[MultiplexerTransform()],
assets_folder=str(assets_dir), update_title=None)
self.p = Thread(target=self.app.run, daemon=True, kwargs=dict(debug=False, port=port), )
self.selected_operation = ""
self.app.layout = html.Div(children=[
html.Div(id='main_panel', children=[
html.H1(children='Status of the scheduled processes'),
html.Div(children='''
A web application to observe, schedule and manipulate processes running with the pythonlaborchestrator.
'''),
dcc.Graph(
id='gantt',
figure=""
),
html.Div(children=[
html.Div(children=[
dcc.Checklist(id='stop_gantt', options=[{'label': 'Update Gantt-Chart', 'value': 'yes'}], value=['yes']),
dcc.Checklist(id='stop_wfg', options=[{'label': 'Update WFG', 'value': 'yes'}], value=['yes']),
html.Div(children=[html.Button('Recover error', id='action', n_clicks=0),
dcc.Checklist(id='repeat', options=[{'label': 'Repeat?', 'value': 'yes'}], value=[],
style={'display': 'inline-block'})]),
html.Button('Run Pre Check', id='pre_check', n_clicks=0),
create_manual_execution_controls(),
html.Button("Export current WFG", id='export', n_clicks=0),
html.Button(children=["Export current", html.Br(), "Scheduling Problem"],
id='export_problem', n_clicks=0),
], style=block_style
),
html.Div(children=[
html.Div(id='delay_label', children="Delay in minutes:"),
dcc.Textarea(id="delay", value='0', style={'width': 100, 'height': 20}),
dcc.Dropdown(options=[{'value': ip, 'label': ip}
for ip in self.importable_processes],
id='process_select', style={"width": 250}),
html.Button("Add process", id='add_process', n_clicks=0),
], style=block_style)
,
html.Div(children=[
dcc.Dropdown(options=[], id='processes', style={"width": 250}),
html.Button('Stop Process', id='stop_process', n_clicks=0),
html.Button('Start Process', id='continue_process', n_clicks=0),
html.Button('Remove Process', id='remove_process', n_clicks=0),
html.Button('Schedule Process', id='schedule_process', n_clicks=0),
html.Button('Reschedule', id='reschedule', n_clicks=0),
], style=block_style
),
html.Div(children=[
html.Button("Add Containers to DB", id='add_to_db', n_clicks=0),
html.Button("Mark Containers removed", id='remove_from_db', n_clicks=0),
], style=block_style),
html.Div(children=[
html.Div(id='schedule_time_label', children=["Schedule ", html.Br(), "computing time ", html.Br(), " in seconds:"]),
dcc.Textarea(value=self.sm_interface.get_parameter("scheduling_time"), id="schedule_time",
style={'width': 30, 'height': 20}),
html.Button("Get", id="schedule_time_get", n_clicks=0),
html.Button("Set", id="schedule_time_set", n_clicks=0),
], style=block_style),
TrafficLight("traffic_light", size=40),
], style={
'display': 'flex', # Place blocks side by side
'alignItems': 'flex-start', # Align blocks at the top
}
),
#html.Button("Emergency STOP", id='emergency_stop', n_clicks=0,
# style={'background-color': 'red', 'font-size': '55px', 'width': '300px', 'height': '150px',
# 'vertical-align': 'top', 'display': 'inline-block'}),
html.Div(
children=dash_interactive_graphviz.DashInteractiveGraphviz(
id="wfg", engine='dot',
dot_source="digraph {\n\thello -> world\n}"
),
style={
'height': '35vh', 'position': 'relative', 'overflow': 'hidden',
},
),
html.Div(children=[
html.Div(id='info_text', children=None),
html.Button('Clear Text', id='clear_info_text', n_clicks=0, style={'display': 'none', 'marginTop': '8px'}),
create_error_state_controls(),
]),
html.Div(id='visu', children=None),
html.Details(
open=arm_open,
children=[
html.Summary("Robo Arm View"),
html.Iframe(
src="http://127.0.0.1:8055/",
style={'width': '100%', 'height': '800px'}
),
],
className="arm-panel"
),
dcc.Interval(
id='interval-component',
interval=500,
n_intervals=0
),
], style={'flex': '1 1 auto', 'minWidth': 0}),
*create_side_panel(db_open=db_open, browser_open=browser_open),
], style={
'display': 'grid',
'gridTemplateColumns': '1fr 6px auto',
'alignItems': 'stretch',
}),
register_traffic_light_callbacks(self.app, "traffic_light")
register_manual_execution_callbacks(self.app, self)
register_error_state_callbacks(self.app, self)
@self.app.callback(
Input(component_id='export_problem', component_property='n_clicks'),
Output(component_id='export_problem', component_property='n_clicks')
)
def export_current_scheduling_problem(n_clicks):
timestamp = create_timestamp()
filename = f"jssp_{timestamp}.json"
self.sm_interface.export_current_scheduling_problem(filename)
@self.app.callback(
Input(component_id='process_select', component_property='value'),
Output(component_id='process_select', component_property='options')
)
def refresh_dropdown_menu(n_clicks):
# refreshes the options according to what's in the file system
return [{'value': ip, 'label': ip} for ip in self.importable_processes]
@self.app.callback(
Input(component_id='remove_from_db', component_property='n_clicks'),
State(component_id='processes', component_property='value'),
Output(component_id='remove_from_db', component_property='n_clicks')
)
def remove_containers_from_db(n_clicks, value):
# todo this is not, how it should be done. use interface methods instead.
labware_to_remove = [c.name for c in self.sm_interface.jssp.process_by_name[value].containers]
self.sm_interface.remove_labware(labware_ids=labware_to_remove)
return no_update
@self.app.callback(
Input(component_id='add_to_db', component_property='n_clicks'),
State(component_id='processes', component_property='value'),
Output(component_id='add_to_db', component_property='n_clicks'),
)
def add_containers_to_db(n_clicks, value):
present, missing = self.sm_interface.check_labware_presence([value])
Logger.info(f"found: {present}\nmissing: {missing}")
self.sm_interface.add_labware(missing)
return no_update
@self.app.callback(
Input(component_id='export', component_property='n_clicks'),
State(component_id='gantt', component_property='figure'),
State(component_id='wfg', component_property='dot_source'),
Output(component_id='export', component_property='n_clicks'),
)
def export_state(n_clicks, gantt, wfg):
timestamp = create_timestamp()
with open(f"WFG_{timestamp}.txt", "w") as writer:
writer.write(wfg)
with open(f"Schedule_{timestamp}.txt", "w") as writer:
writer.write(self.sm_interface.gantt_chart_scheduled_processes.to_json())
return no_update
@self.app.callback(
Input(component_id='add_process', component_property='n_clicks'),
Output(component_id='process_select', component_property='title'),
State(component_id='process_select', component_property='value'),
State(component_id='delay', component_property='value'),
)
def add_process(n_clicks, value, delay):
Logger.info(f"add {value} with {delay} minutes delay ")
try:
delay = int(delay)
except:
Logger.warning(f"could not translate delay {delay} to integer")
delay = 0
print(value)
# re-read the process files here
available = self.importable_processes
pl_process = self.importable_process_by_name[value]
name = self.sm_interface.add_process(process_object=pl_process, delay=delay)
self.process_type[name] = value
return 'chosen'
@self.app.callback(
Input('interval-component', 'n_intervals'),
Output(component_id='processes', component_property='options')
)
def update_active_processes(n_intervals):
options = []
for process in self.sm_interface.processes:
name = process.name
options.append(dict(value=name, label=f"{name} ({self.process_type[name] if name in self.process_type else ''})"))
return options
@self.app.callback(
Input(component_id='continue_process', component_property='n_clicks'),
State(component_id='processes', component_property='value'),
Output(component_id='processes', component_property='value'),
)
def start_process(n_clicks, value):
self.sm_interface.start_processes([value])
return no_update
@self.app.callback(
Input(component_id='stop_process', component_property='n_clicks'),
State(component_id='processes', component_property='value'),
Output(component_id='processes', component_property='value'),
)
def stop_process(n_clicks, value):
self.sm_interface.stop_processes([value])
return no_update
@self.app.callback(
Input(component_id='remove_process', component_property='n_clicks'),
State(component_id='processes', component_property='value'),
Output(component_id='processes', component_property='value'),
)
def remove_process(n_clicks, value):
self.sm_interface.remove_processes([value])
return no_update
@self.app.callback(
Input(component_id='schedule_process', component_property='n_clicks'),
State(component_id='processes', component_property='value'),
Output(component_id='processes', component_property='value'),
)
def schedule_process(n_clicks, value):
try:
self.sm_interface.add_to_schedule(value)
except Exception as ex:
Logger.exception(traceback.format_exc())
return no_update
@self.app.callback(
Input(component_id='pre_check', component_property='n_clicks'),
State(component_id='processes', component_property='value'),
Output(component_id='info_text', component_property='children'),
)
def pre_check_process(n_clicks, value):
try:
process = self.sm_interface.jssp.process_by_name[value]
ready, report = self.sm_interface.worker.check_prerequisites(process)
return f"Ready to start:{ready}\n\n{report}"
except Exception as ex:
print(f"{ex}\n{traceback.format_exc()}")
return no_update
@self.app.callback(
Input(component_id='reschedule', component_property='n_clicks'),
Output(component_id='reschedule', component_property='n_clicks'),
)
def reschedule(n_clicks):
# this is not using a proper interface method
self.sm_interface.schedule_manager.mark_schedule_invalid(enforce=True)
@self.app.callback(
Input(component_id='schedule_time_get', component_property='n_clicks'),
Output(component_id='schedule_time', component_property='value'),
)
def get_schedule_time(n_clicks):
return self.sm_interface.get_parameter("scheduling_time")
@self.app.callback(
Input(component_id='schedule_time_set', component_property='n_clicks'),
State(component_id="schedule_time", component_property="value"),
Output(component_id='schedule_time_get', component_property='n_clicks'),
)
def set_schedule_time(n_clicks, value):
print(f"Setting scheduling time to {value}")
self.sm_interface.set_parameter("scheduling_time", value)
return no_update
@self.app.callback(
Input(component_id="emergency_stop", component_property='n_clicks'),
Output(component_id="emergency_stop", component_property='n_clicks'),
)
def stop(n_clicks):
print("this is just a test")
@self.app.callback(
Output(component_id='info_text', component_property='children'),
Input(component_id='wfg', component_property='selected_node'),
)
def select_operation(selected_node):
if not selected_node:
return no_update
self.selected_operation = selected_node
if selected_node in self.sm_interface.jssp.operable:
op = self.sm_interface.jssp.operable[selected_node]
if isinstance(op, Variable):
if op.result is not None:
print('at this point we could visualize the variables result')
return str(op)
else:
g = self.sm_interface.jssp.combined_wfg
if selected_node in g.nodes:
nx_attr = g.nodes[selected_node]
if nx_attr.get('type') == 'container':
cont_name = nx_attr.get('name', selected_node)
if cont_name in self.sm_interface.jssp.container_info_by_name:
return str(self.sm_interface.jssp.container_info_by_name[cont_name])
return no_update
@self.app.callback(
Output(component_id='visu', component_property='children'),
Input(component_id='wfg', component_property='selected_node'),
)
def select_operation2(selected_node):
var = self.sm_interface.get_operable_node(selected_node)
if isinstance(var, Variable):
if var.result is not None:
if "Data" in var.result.__dir__():
s = var.result.Data
return None
@self.app.callback(
Output(component_id='info_text', component_property='children'),
Input(component_id='action', component_property='n_clicks'),
State(component_id='repeat', component_property='value')
)
def manipulate(n_clicks, value):
job = self.sm_interface.get_operable_node(self.selected_operation)
repeat = 'yes' in value
if not job.status == StepStatus.ERROR:
print("Selected Operation is not in Error-State")
Logger.warning("Selected Operation is not in Error-State")
self.sm_interface.error_resolved(self.selected_operation, repeat_operation=repeat)
return f"Recovered {self.selected_operation}, repeat: {repeat}"
@self.app.callback(
Output(component_id='clear_info_text', component_property='style'),
Input(component_id='info_text', component_property='children'),
)
def toggle_clear_info_text_button(info_text):
if info_text:
return {'display': 'inline-block', 'marginTop': '8px'}
return {'display': 'none', 'marginTop': '8px'}
@self.app.callback(
Output(component_id='info_text', component_property='children'),
Output(component_id='wfg', component_property='selected_node'),
Input(component_id='clear_info_text', component_property='n_clicks'),
)
def clear_info_text(n_clicks):
if n_clicks:
return None, None
return no_update, no_update
@self.app.callback(
[Output(component_id='gantt', component_property='figure'),
Output(component_id='traffic_light-active', component_property="data")],
Input(component_id='interval-component', component_property='n_intervals'),
State(component_id='stop_gantt', component_property='value')
)
def refresh_gantt(n_intervals, value):
if 'yes' in value:
if n_intervals % 2 == 0:
try:
gantt = self.sm_interface.gantt_chart_scheduled_processes
return [gantt, self.sm_interface.schedule_manager.schedule_quality.value]
except Exception as ex:
return [no_update, no_update]
return no_update
@self.app.callback(
Output(component_id='wfg', component_property='dot_source'),
Input('interval-component', 'n_intervals'),
State(component_id='stop_wfg', component_property='value'),
State(component_id='wfg', component_property='dot_source')
)
def refresh_wfg(n_intervals, value, current_dot_source):
try:
if 'yes' not in value:
return no_update
fig_str = str(self.sm_interface.workflow_graph_scheduled_processes)
# only update if there is something to update
if current_dot_source == fig_str:
return no_update
return fig_str
except Exception as ex:
print(f"{ex}\n{traceback.format_exc()}")
print("wfg-update failed")
return no_update
[docs]
def run(self):
import logging
if self.p.is_alive():
Logger.warning('Server is already running. Restarting server')
self.stop()
logging.getLogger('werkzeug').setLevel(logging.ERROR)
self.p.start()
[docs]
def stop(self):
print("Sorry, I don't know, how to stop")
