Automation 2.0: WIP added endpoints to backend, frontend connected and panel made, now debugging

This commit is contained in:
appleb_m committed 2026-03-26 17:29:17 +01:00
1 parent 3dd5f46f1e
commit 303173ae82
9 files changed
+415 -190

No files matched your search

@@ -15,6 +15,7 @@ from aare.common.automation_models import (
WorkflowStateKind,
)
CURRENT_QUEUE_LIMIT = 576
def build_default_steps() -> list[WorkflowStepRecord]:
return [
@@ -60,6 +61,10 @@ class WorkflowRedisManager:
if not item.item_id:
item.item_id = self._next_item_id()
current_count = self._client.zcard(self._queue_key())
if current_count >= CURRENT_QUEUE_LIMIT:
raise ValueError(f"Queue size limit reached ({CURRENT_QUEUE_LIMIT} items). Please clear the queue.")
pipe = self._client.pipeline(transaction=True)
pipe.set(self._item_key(item.item_id), item.model_dump_json())
pipe.zadd(self._queue_key(), {item.item_id: float(item.order_index)})
+4 -4
View File
@@ -258,7 +258,7 @@ class SimulatedMountHandler(StateHandler):
def execute(self, context: WorkflowContext) -> StateResult:
self.validate(context)
import time
time.sleep(0.5) # Simulate some work
time.sleep(10) # Simulate some work
context.current_state = WorkflowStateKind.MOUNT
context.last_message = "[SIMULATION] Sample would be mounted"
return StateResult(
@@ -280,7 +280,7 @@ class SimulatedLoopCentreHandler(StateHandler):
def execute(self, context: WorkflowContext) -> StateResult:
self.validate(context)
import time
time.sleep(0.3)
time.sleep(10)
context.current_state = WorkflowStateKind.LOOP_CENTRE
context.last_message = "[SIMULATION] Loop would be centred"
return StateResult(
@@ -302,7 +302,7 @@ class SimulatedRasterHandler(StateHandler):
def execute(self, context: WorkflowContext) -> StateResult:
self.validate(context)
import time
time.sleep(0.4)
time.sleep(10)
context.current_state = WorkflowStateKind.RASTER
context.last_message = "[SIMULATION] Raster scan would be performed"
return StateResult(
@@ -324,7 +324,7 @@ class SimulatedDataCollectionHandler(StateHandler):
def execute(self, context: WorkflowContext) -> StateResult:
self.validate(context)
import time
time.sleep(0.5)
time.sleep(10)
context.current_state = WorkflowStateKind.DATA_COLLECTION
context.last_message = "[SIMULATION] Data collection would be performed"
return StateResult(
+69 -2
View File
@@ -149,7 +149,11 @@ async def create_queue_item(
recipe=request.recipe,
)
item = redis_mgr.create_item(item)
try:
item = redis_mgr.create_item(item)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
return item
@@ -199,6 +203,68 @@ async def delete_queue_item(
redis_mgr.delete_item(item_id)
return {"ok": True, "message": f"Item {item_id} deleted"}
@router.delete("/queue/clear")
async def clear_queue(
token: str = Depends(oauth2_scheme),
redis_mgr: WorkflowRedisManager = Depends(get_redis_manager),
):
"""Clear all non-running items from the queue."""
data = parse_token(token)
check_jwt_rw(get_cfg(), data)
items = redis_mgr.list_items(include_finished=True)
deleted_count = 0
for item in items:
# Skip running items
if item.status == QueueItemStatus.RUNNING:
continue
# Check access
if not data.staff and item.owner_pgroup not in data.pgroups:
continue
redis_mgr.delete_item(item.item_id)
deleted_count += 1
return {"ok": True, "message": f"Cleared {deleted_count} items from queue"}
@router.post("/control/skip_sample", response_model=ControlActionResponse)
async def request_skip_sample(
token: str = Depends(oauth2_scheme),
redis_mgr: WorkflowRedisManager = Depends(get_redis_manager),
runner: PersistentWorkflowRunner = Depends(get_runner),
):
"""Skip the current sample entirely (abort current sample and move to next)."""
data = parse_token(token)
check_jwt_rw(get_cfg(), data)
runtime = redis_mgr.get_runtime()
if not runtime.running or not runtime.current_item_id:
return ControlActionResponse(
ok=False,
control=redis_mgr.get_control(),
message="No sample currently running",
)
# Mark current item as skipped and complete it
runner.complete_item(runtime.current_item_id, QueueItemStatus.SKIPPED)
redis_mgr.append_event(WorkflowEvent(
beamline=runner.beamline,
item_id=runtime.current_item_id,
event_type="sample_skipped",
actor=data.sub,
message="Sample skipped by user",
))
return ControlActionResponse(
ok=True,
control=redis_mgr.get_control(),
message="Sample skipped",
)
@router.post("/queue/{item_id}/move")
async def move_queue_item(
@@ -304,8 +370,9 @@ async def request_abort(
data = parse_token(token)
check_jwt_rw(get_cfg(), data)
# Clear pause_requested when aborting
control = redis_mgr.request_control(
{"abort_requested": True},
{"abort_requested": True, "pause_requested": False, "resume_requested": False},
requested_by=data.sub,
)
+85 -51
View File
@@ -29,12 +29,12 @@ from aare.common.automation_workflow import (
class PersistentWorkflowRunner:
"""
Orchestrates workflow execution with Redis persistence.
Orchestrates workflow execution with Redis persistence.
Combines:
- State handlers (from automation_workflow)
- Redis persistence (from automation_queue_manager)
"""
Combines:
- State handlers (from automation_workflow)
- Redis persistence (from automation_queue_manager)
"""
def __init__(
self,
@@ -183,6 +183,10 @@ class PersistentWorkflowRunner:
message=f"Item finished with status {status.value}",
))
if status in (QueueItemStatus.COMPLETED, QueueItemStatus.ABORTED, QueueItemStatus.SKIPPED):
self._redis.delete_item(item_id)
return item
return item
def check_control(self) -> ControlState:
@@ -203,17 +207,22 @@ class AutomationLoop:
Background task that drives fully automated workflow execution.
Polls control state and processes the queue automatically.
IMPORTANT: This loop does NOT auto-start. It must be explicitly started
via the start() method (triggered by the "Start Automation" button).
"""
def __init__(
self,
runner: PersistentWorkflowRunner,
redis_manager: WorkflowRedisManager,
poll_interval: float = 0.5,
poll_interval: float = 0.3,
step_delay: float = 0.5, # Delay between steps for control checks
):
self._runner = runner
self._redis = redis_manager
self._poll_interval = poll_interval
self._step_delay = step_delay
self._task: asyncio.Task | None = None
self._enabled = False
self._on_step_complete: Callable[[str, str, StateResult], None] | None = None
@@ -231,7 +240,7 @@ class AutomationLoop:
self._on_step_complete = callback
def start(self) -> None:
"""Start the automation loop."""
"""Start the automation loop. Must be explicitly called."""
if self._task is not None and not self._task.done():
return # Already running
@@ -248,6 +257,10 @@ class AutomationLoop:
"""Stop the automation loop gracefully."""
self._enabled = False
# Cancel the task if it exists
if self._task is not None and not self._task.done():
self._task.cancel()
self._redis.append_event(WorkflowEvent(
beamline=self._runner.beamline,
event_type="automation_stopped",
@@ -258,7 +271,57 @@ class AutomationLoop:
"""Main automation loop."""
while self._enabled:
try:
# Check control state FIRST before doing anything
control = self._runner.check_control()
runtime = self._redis.get_runtime()
# Handle abort immediately
if control.abort_requested:
if runtime.current_item_id:
self._runner.complete_item(runtime.current_item_id, QueueItemStatus.ABORTED)
# Clear all control flags including pause
self._redis.clear_control()
self.stop()
await asyncio.sleep(self._poll_interval)
continue
# Handle pause
if control.pause_requested:
if runtime.running and not runtime.paused:
self._redis.patch_runtime({"paused": True})
self._redis.append_event(WorkflowEvent(
beamline=self._runner.beamline,
item_id=runtime.current_item_id,
event_type="workflow_paused",
message="Workflow paused by user request",
))
await asyncio.sleep(self._poll_interval)
continue
# Handle resume
if control.resume_requested and runtime.paused:
self._redis.patch_runtime({"paused": False})
self._redis.request_control(
{"resume_requested": False},
requested_by="automation_loop",
)
self._redis.append_event(WorkflowEvent(
beamline=self._runner.beamline,
item_id=runtime.current_item_id,
event_type="workflow_resumed",
message="Workflow resumed",
))
# Don't process if paused
if runtime.paused:
await asyncio.sleep(self._poll_interval)
continue
await self._tick()
except asyncio.CancelledError:
# Loop was cancelled (stop() was called)
break
except Exception as e:
self._redis.patch_runtime({"last_error": str(e)})
self._redis.append_event(WorkflowEvent(
@@ -266,51 +329,14 @@ class AutomationLoop:
event_type="automation_error",
message=f"Automation error: {e}",
))
# Brief pause on error before retrying
await asyncio.sleep(2.0)
await asyncio.sleep(self._poll_interval)
async def _tick(self) -> None:
"""Single iteration of the automation loop."""
control = self._runner.check_control()
"""Single iteration of the automation loop - process one step."""
runtime = self._redis.get_runtime()
# Handle pause state
if control.pause_requested:
if runtime.running and not runtime.paused:
self._redis.patch_runtime({"paused": True})
self._redis.append_event(WorkflowEvent(
beamline=self._runner.beamline,
item_id=runtime.current_item_id,
event_type="workflow_paused",
message="Workflow paused by user request",
))
return
# Handle resume
if control.resume_requested and runtime.paused:
self._redis.patch_runtime({"paused": False})
self._redis.request_control(
{"resume_requested": False},
requested_by="automation_loop",
)
self._redis.append_event(WorkflowEvent(
beamline=self._runner.beamline,
item_id=runtime.current_item_id,
event_type="workflow_resumed",
message="Workflow resumed",
))
# Don't process if paused
if runtime.paused:
return
# Handle abort
if control.abort_requested and runtime.current_item_id:
self._runner.complete_item(runtime.current_item_id, QueueItemStatus.ABORTED)
self._redis.clear_control()
return
control = self._runner.check_control()
# If nothing running, try to start next item
if not runtime.running or not runtime.current_item_id:
@@ -319,8 +345,15 @@ class AutomationLoop:
return # Queue empty, nothing to do
self._runner.start_item(next_item.item_id)
# Add delay after starting to allow UI to catch up
await asyncio.sleep(self._step_delay)
runtime = self._redis.get_runtime()
# Re-check control after potential start
control = self._runner.check_control()
if control.abort_requested or control.pause_requested:
return # Let main loop handle it
# Process current item
item = self._redis.get_item(runtime.current_item_id)
if item is None:
@@ -330,8 +363,7 @@ class AutomationLoop:
# Check if item is complete
if item.current_step_index >= len(item.steps):
self._runner.complete_item(item.item_id, QueueItemStatus.COMPLETED)
# Check for next sample request
# Clear next_sample_requested if set
if control.next_sample_requested:
self._redis.request_control(
{"next_sample_requested": False},
@@ -339,7 +371,7 @@ class AutomationLoop:
)
return
# Handle skip request
# Handle skip request (skip current step)
if control.skip_requested:
step_kind = item.steps[item.current_step_index].kind
self._redis.update_step(item.item_id, item.current_step_index, status="skipped")
@@ -366,15 +398,17 @@ class AutomationLoop:
)
# Run the step (this is blocking in the async context)
step_name = item.steps[item.current_step_index].kind
result = await asyncio.to_thread(self._runner.run_current_step, context)
# Add delay after step to allow control checks
await asyncio.sleep(self._step_delay)
# Notify callback if set
if self._on_step_complete:
step_name = item.steps[item.current_step_index - 1].kind # -1 because index advanced
self._on_step_complete(item.item_id, step_name, result)
if __name__ == "__main__":
client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
redis_mgr = WorkflowRedisManager(client, beamline="x10sa")
+2 -2
View File
@@ -37,8 +37,8 @@ if __name__ == "__main__":
default_gonio_camera_id = 3
case MXBeamline.X10SA:
default_url = "http://127.0.0.1:5210"
default_zmq_addr = "tcp://x10sa-spark-01:9091" #"tcp://x10sa-pserv-01:9089" #
default_pred_zmq_addr = "tcp://x10sa-spark-01:9091" #"tcp://sls-gpu-003:9089"#""
default_zmq_addr = "tcp://sls-gpu-003:9089"#"tcp://x10sa-spark-01:9091" #"tcp://x10sa-spark-01:9091" #"tcp://x10sa-pserv-01:9089" #
default_pred_zmq_addr = "tcp://sls-gpu-003:9089"#"""tcp://x10sa-spark-01:9091" #
default_beamline_cam_addr = "axis-accc8eb02488.psi.ch"
default_gonio_cam_addr = "axis-accc8ea5e463.psi.ch"
default_gonio_camera_id = 1
+1
View File
@@ -538,6 +538,7 @@ class MainWindow(QMainWindow):
self.workflow_panel.request_skip.connect(self.daq.workflow_skip)
self.workflow_panel.request_start_automation.connect(self.daq.workflow_start_automation)
self.workflow_panel.request_stop_automation.connect(self.daq.workflow_stop_automation)
self.workflow_panel.request_skip_sample.connect(self.daq.workflow_skip_sample)
# DAQ worker -> workflow panel
self.daq.workflow_queue_loaded.connect(self.workflow_panel.update_queue)
+175 -85
View File
@@ -4,7 +4,7 @@ Workflow automation panel for queue management and step control.
Displays:
- Queue items with status (supports drag-drop from sample list)
- Current step progress
- Control buttons (Next/Pause/Resume/Abort/Skip)
- Control buttons (Start Guided/Start Automation/Pause/Resume/Abort/Skip)
- Live status from SSE
"""
@@ -87,6 +87,13 @@ class StepProgressWidget(QWidget):
if 0 <= index < len(self._step_labels):
self._step_labels[index].setStyleSheet(self._style_for_step(index, status))
def clear_steps(self) -> None:
"""Clear all step labels."""
for lbl in self._step_labels:
lbl.deleteLater()
self._step_labels.clear()
self._steps = []
def _style_for_step(self, index: int, status: str) -> str:
"""Get stylesheet for step based on status."""
base = "padding: 4px; border-radius: 3px; "
@@ -118,6 +125,7 @@ class DraggableQueueListWidget(QListWidget):
samples_dropped = Signal(list) # list[SampleShortInfo]
item_reordered = Signal(str, int) # item_id, new_index
delete_requested = Signal(list) # list[item_ids]
start_item_requested = Signal(str) # item_id
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
@@ -145,7 +153,6 @@ class DraggableQueueListWidget(QListWidget):
"""Accept drops from sample panels."""
mime = event.mimeData()
if mime.hasText():
# Check if it's sample data (JSON)
try:
text = mime.text()
if text.startswith("{") or text.startswith("["):
@@ -153,7 +160,6 @@ class DraggableQueueListWidget(QListWidget):
return
except Exception:
pass
# Accept internal moves
if event.source() == self:
event.acceptProposedAction()
return
@@ -173,7 +179,6 @@ class DraggableQueueListWidget(QListWidget):
if mime.hasText():
text = mime.text()
try:
# Try to parse as SampleShortInfoList
sample_list = SampleShortInfoList.model_validate_json(text)
if sample_list.s:
self.samples_dropped.emit(sample_list.s)
@@ -183,7 +188,6 @@ class DraggableQueueListWidget(QListWidget):
pass
try:
# Try single sample
sample = SampleShortInfo.model_validate_json(text)
self.samples_dropped.emit([sample])
event.acceptProposedAction()
@@ -191,14 +195,11 @@ class DraggableQueueListWidget(QListWidget):
except Exception:
pass
# Internal reorder
if event.source() == self:
# Get drop position
drop_row = self.indexAt(event.position().toPoint()).row()
if drop_row < 0:
drop_row = self.count()
# Get selected items
selected = self.selectedItems()
if selected and self._item_ids:
for item in selected:
@@ -251,8 +252,7 @@ class DraggableQueueListWidget(QListWidget):
if action == delete_action:
self.delete_requested.emit([self._item_ids[row]])
elif action == start_action:
# Will need to emit a signal for this
pass
self.start_item_requested.emit(self._item_ids[row])
elif action == move_top_action:
self.item_reordered.emit(self._item_ids[row], 0)
elif action == move_bottom_action:
@@ -266,8 +266,8 @@ class WorkflowPanel(QWidget):
Supports:
- Drag-drop samples from TELL sample panel
- Queue management (delete, reorder, clear, add all)
- Step-by-step guided mode
- Full automation mode
- Step-by-step guided mode (explicit start)
- Full automation mode (explicit start)
"""
# Signals for DAQ worker
@@ -282,17 +282,20 @@ class WorkflowPanel(QWidget):
request_pause = Signal()
request_resume = Signal()
request_abort = Signal()
request_skip = Signal()
request_skip = Signal() # Skip current step
request_skip_sample = Signal() # Skip entire current sample
request_start_automation = Signal()
request_stop_automation = Signal()
request_start_guided = Signal(str) # Start guided mode with item_id
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self._queue_items: list[QueueItem] = []
self._all_samples: list[SampleShortInfo] = [] # Cache for "Add All"
self._all_samples: list[SampleShortInfo] = []
self._runtime: RuntimeState | None = None
self._control: ControlState | None = None
self._automation_enabled = False
self._was_paused_before_automation = False # Track pause state before automation
self._setup_ui()
def _setup_ui(self) -> None:
@@ -316,84 +319,104 @@ class WorkflowPanel(QWidget):
layout.addWidget(status_group)
# === Start Buttons (explicit start required) ===
start_group = QGroupBox("Start Processing")
start_layout = QHBoxLayout(start_group)
self._start_guided_btn = QPushButton("▶ Start Guided Mode")
self._start_guided_btn.setStyleSheet("background-color: #4CAF50; color: white; font-weight: bold; padding: 8px;")
self._start_guided_btn.setToolTip("Start processing the first pending sample in guided (step-by-step) mode")
self._start_guided_btn.clicked.connect(self._on_start_guided)
self._start_auto_btn = QPushButton("▶▶ Start Automation")
self._start_auto_btn.setStyleSheet("background-color: #2196F3; color: white; font-weight: bold; padding: 8px;")
self._start_auto_btn.setToolTip("Start fully automated processing of all samples")
self._start_auto_btn.clicked.connect(self._on_start_automation)
self._stop_btn = QPushButton("⏹ Stop")
self._stop_btn.setStyleSheet("background-color: #9E9E9E; color: white; font-weight: bold; padding: 8px;")
self._stop_btn.setToolTip("Stop automation mode (current step will complete)")
self._stop_btn.clicked.connect(self._on_stop)
self._stop_btn.setVisible(False)
start_layout.addWidget(self._start_guided_btn)
start_layout.addWidget(self._start_auto_btn)
start_layout.addWidget(self._stop_btn)
layout.addWidget(start_group)
# === Control Buttons ===
controls_group = QGroupBox("Controls")
controls_group = QGroupBox("Step Controls")
controls_layout = QVBoxLayout(controls_group)
# Mode toggle
mode_layout = QHBoxLayout()
self._mode_label = QLabel("Mode:")
self._guided_btn = QPushButton("Guided Manual")
self._guided_btn.setCheckable(True)
self._guided_btn.setChecked(True)
self._auto_btn = QPushButton("Automation")
self._auto_btn.setCheckable(True)
self._guided_btn.clicked.connect(self._on_guided_mode)
self._auto_btn.clicked.connect(self._on_automation_mode)
mode_layout.addWidget(self._mode_label)
mode_layout.addWidget(self._guided_btn)
mode_layout.addWidget(self._auto_btn)
mode_layout.addStretch()
controls_layout.addLayout(mode_layout)
# Step controls
# Step controls row 1
step_layout = QHBoxLayout()
self._next_btn = QPushButton("Next Step")
self._next_btn.setStyleSheet("background-color: #4CAF50; color: white;")
self._next_btn.setToolTip("Execute the next step (guided mode only)")
self._next_btn.clicked.connect(self.request_next_step.emit)
self._skip_btn = QPushButton("Skip")
self._skip_btn = QPushButton("Skip Step")
self._skip_btn.setStyleSheet("background-color: #FF9800; color: white;")
self._skip_btn.setToolTip("Skip the current step and move to the next")
self._skip_btn.clicked.connect(self.request_skip.emit)
self._pause_btn = QPushButton("Pause")
self._pause_btn.setStyleSheet("background-color: #2196F3; color: white;")
self._pause_btn.clicked.connect(self._on_pause_resume)
self._abort_btn = QPushButton("Abort")
self._abort_btn.setStyleSheet("background-color: #F44336; color: white;")
self._abort_btn.clicked.connect(self.request_abort.emit)
self._skip_sample_btn = QPushButton("Skip Sample")
self._skip_sample_btn.setStyleSheet("background-color: #FF5722; color: white;")
self._skip_sample_btn.setToolTip("Skip the entire current sample and move to the next")
self._skip_sample_btn.clicked.connect(self.request_skip_sample.emit)
step_layout.addWidget(self._next_btn)
step_layout.addWidget(self._skip_btn)
step_layout.addWidget(self._pause_btn)
step_layout.addWidget(self._abort_btn)
step_layout.addWidget(self._skip_sample_btn)
controls_layout.addLayout(step_layout)
# Step controls row 2
control_layout2 = QHBoxLayout()
self._pause_btn = QPushButton("⏸ Pause")
self._pause_btn.setStyleSheet("background-color: #2196F3; color: white;")
self._pause_btn.clicked.connect(self._on_pause_resume)
self._abort_btn = QPushButton("🛑 Abort")
self._abort_btn.setStyleSheet("background-color: #F44336; color: white;")
self._abort_btn.clicked.connect(self.request_abort.emit)
control_layout2.addWidget(self._pause_btn)
control_layout2.addWidget(self._abort_btn)
controls_layout.addLayout(control_layout2)
layout.addWidget(controls_group)
# === Queue Section ===
queue_group = QGroupBox("Queue (drag samples here)")
queue_layout = QVBoxLayout(queue_group)
# Info label
self._drop_hint = QLabel("💡 Drag samples from the Sample List to add them")
self._drop_hint.setStyleSheet("color: #666; font-style: italic;")
queue_layout.addWidget(self._drop_hint)
# Draggable queue list
self._queue_list = DraggableQueueListWidget()
self._queue_list.setMinimumHeight(150)
self._queue_list.itemDoubleClicked.connect(self._on_item_double_clicked)
self._queue_list.samples_dropped.connect(self._on_samples_dropped)
self._queue_list.delete_requested.connect(self._on_delete_requested)
self._queue_list.item_reordered.connect(self._on_item_reordered)
self._queue_list.start_item_requested.connect(self._on_start_specific_item)
queue_layout.addWidget(self._queue_list)
# Queue management buttons
queue_btn_layout = QHBoxLayout()
self._add_all_btn = QPushButton("➕ Add All Samples")
self._add_all_btn = QPushButton("➕ Add All")
self._add_all_btn.clicked.connect(self._on_add_all_clicked)
self._add_all_btn.setToolTip("Add all samples from the Sample List to the queue")
self._add_all_btn.setToolTip("Add all samples from the Sample List")
self._remove_selected_btn = QPushButton("🗑 Remove Selected")
self._remove_selected_btn = QPushButton("🗑 Remove")
self._remove_selected_btn.clicked.connect(self._on_remove_selected)
self._clear_btn = QPushButton("✖ Clear Queue")
self._clear_btn = QPushButton("✖ Clear")
self._clear_btn.clicked.connect(self._on_clear_queue)
self._refresh_btn = QPushButton("🔄")
@@ -413,20 +436,47 @@ class WorkflowPanel(QWidget):
# Initial state
self._update_button_states()
def _on_guided_mode(self) -> None:
"""Switch to guided manual mode."""
self._guided_btn.setChecked(True)
self._auto_btn.setChecked(False)
if self._automation_enabled:
self.request_stop_automation.emit()
def _on_start_guided(self) -> None:
"""Start guided mode with the first pending sample."""
pending = [i for i in self._queue_items if i.status == QueueItemStatus.PENDING]
if not pending:
QMessageBox.information(self, "No Samples", "No pending samples in the queue.")
return
# Start the first pending item
self.request_start_item.emit(pending[0].item_id)
def _on_start_specific_item(self, item_id: str) -> None:
"""Start guided mode with a specific sample."""
self.request_start_item.emit(item_id)
def _on_start_automation(self) -> None:
"""Start full automation mode."""
pending = [i for i in self._queue_items if i.status == QueueItemStatus.PENDING]
is_running = self._runtime is not None and self._runtime.running
if not pending and not is_running:
QMessageBox.information(self, "No Samples", "No pending samples in the queue.")
return
# Remember if we were paused before starting automation
self._was_paused_before_automation = (
self._control is not None and self._control.pause_requested
)
self._automation_enabled = True
self.request_start_automation.emit()
self._update_button_states()
def _on_automation_mode(self) -> None:
"""Switch to automation mode."""
self._auto_btn.setChecked(True)
self._guided_btn.setChecked(False)
if not self._automation_enabled:
self.request_start_automation.emit()
def _on_stop(self) -> None:
"""Stop automation mode."""
self._automation_enabled = False
self.request_stop_automation.emit()
# Restore pause state if it was paused before automation started
if self._was_paused_before_automation:
self.request_pause.emit()
self._update_button_states()
def _on_pause_resume(self) -> None:
@@ -436,20 +486,11 @@ class WorkflowPanel(QWidget):
else:
self.request_pause.emit()
def _on_item_double_clicked(self, item: QListWidgetItem) -> None:
"""Start processing double-clicked item."""
idx = self._queue_list.row(item)
if 0 <= idx < len(self._queue_items):
queue_item = self._queue_items[idx]
if queue_item.status == QueueItemStatus.PENDING:
self.request_start_item.emit(queue_item.item_id)
def _on_samples_dropped(self, samples: list[SampleShortInfo]) -> None:
"""Handle samples dropped onto the queue."""
logger.info(f"Adding {len(samples)} samples to workflow queue")
for sample in samples:
self.request_add_sample.emit(sample)
# Refresh after a short delay to let server process
QTimer.singleShot(500, self.request_queue_refresh.emit)
def _on_delete_requested(self, item_ids: list[str]) -> None:
@@ -485,14 +526,36 @@ class WorkflowPanel(QWidget):
reply = QMessageBox.question(
self,
"Clear Queue",
f"Remove all {len(self._queue_items)} items from the queue?",
"Remove all non-running items from the queue?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if reply == QMessageBox.StandardButton.Yes:
self.request_clear_queue.emit()
QTimer.singleShot(200, self._clear_local_queue)
QTimer.singleShot(500, self.request_queue_refresh.emit)
def _clear_local_queue(self) -> None:
"""Clear local queue state (called after server clear)."""
# Keep only running items
self._queue_items = [i for i in self._queue_items if i.status == QueueItemStatus.RUNNING]
# Update GUI
self._queue_list.clear()
item_ids = []
for item in self._queue_items:
status_emoji = "▶️"
display_text = f"{status_emoji} {item.sample_name or item.item_id}"
list_item = QListWidgetItem(display_text)
list_item.setBackground(QColor("#E6F3FF"))
self._queue_list.addItem(list_item)
item_ids.append(item.item_id)
self._queue_list.set_item_ids(item_ids)
self._update_button_states()
def _on_add_all_clicked(self) -> None:
"""Add all available samples to the queue."""
if not self._all_samples:
@@ -518,25 +581,34 @@ class WorkflowPanel(QWidget):
"""Update button enabled/disabled states based on current state."""
is_running = self._runtime is not None and self._runtime.running
is_paused = self._runtime is not None and self._runtime.paused
is_automation = self._auto_btn.isChecked()
has_pending = any(i.status == QueueItemStatus.PENDING for i in self._queue_items)
# In automation mode, hide the Next button
self._next_btn.setVisible(not is_automation)
self._start_guided_btn.setVisible(not is_running and not self._automation_enabled)
self._start_auto_btn.setVisible(not self._automation_enabled)
self._stop_btn.setVisible(self._automation_enabled)
self._start_guided_btn.setEnabled(has_pending)
self._start_auto_btn.setEnabled(has_pending or is_running)
self._next_btn.setVisible(not self._automation_enabled)
self._next_btn.setEnabled(is_running and not is_paused)
# Skip buttons - enabled when running
self._skip_btn.setEnabled(is_running)
self._skip_sample_btn.setEnabled(is_running)
self._abort_btn.setEnabled(is_running)
# Pause/Resume button
if is_paused:
self._pause_btn.setText("Resume")
self._pause_btn.setText("▶ Resume")
self._pause_btn.setStyleSheet("background-color: #4CAF50; color: white;")
else:
self._pause_btn.setText("Pause")
self._pause_btn.setText("⏸ Pause")
self._pause_btn.setStyleSheet("background-color: #2196F3; color: white;")
self._pause_btn.setEnabled(is_running)
# Update drop hint visibility
# Drop hint
has_items = len(self._queue_items) > 0
self._drop_hint.setVisible(not has_items)
@@ -587,7 +659,8 @@ class WorkflowPanel(QWidget):
"font-size: 14px; font-weight: bold; color: #FF9800;"
)
else:
self._status_label.setText("▶️ Running")
mode_str = "Automation" if self._automation_enabled else "Guided"
self._status_label.setText(f"▶️ Running ({mode_str})")
self._status_label.setStyleSheet(
"font-size: 14px; font-weight: bold; color: #4CAF50;"
)
@@ -601,6 +674,14 @@ class WorkflowPanel(QWidget):
"font-size: 14px; font-weight: bold; color: #757575;"
)
self._current_item_label.setText("No item running")
self._step_progress.clear_steps()
# Reset automation enabled when nothing is running
if self._automation_enabled and not runtime.running:
# Check if queue has more pending items
pending = [i for i in self._queue_items if i.status == QueueItemStatus.PENDING]
if not pending:
self._automation_enabled = False
self._update_button_states()
@@ -622,8 +703,6 @@ class WorkflowPanel(QWidget):
def update_automation_enabled(self, enabled: bool) -> None:
"""Update automation mode state."""
self._automation_enabled = enabled
self._auto_btn.setChecked(enabled)
self._guided_btn.setChecked(not enabled)
self._update_button_states()
@Slot(list)
@@ -636,5 +715,16 @@ class WorkflowPanel(QWidget):
"""Handle workflow events from SSE stream."""
logger.debug(f"Workflow event: {event_type} - {message}")
if event_type in ("item_started", "item_completed", "step_completed", "step_skipped"):
self.request_queue_refresh.emit()
if event_type in ("item_started", "item_completed", "step_completed", "step_skipped", "sample_skipped"):
self.request_queue_refresh.emit()
# Detect automation stop
if event_type == "automation_stopped":
self._automation_enabled = False
# Restore pause state if it was paused before automation started
if self._was_paused_before_automation:
self.request_pause.emit()
self._was_paused_before_automation = False
self._update_button_states()
+17 -6
View File
@@ -1281,9 +1281,15 @@ class DAQWorker(QObject):
try:
response_data = self.handle_response(reply)
data = json.loads(response_data)
from aare.common.automation_models import QueueItem
from aare.common.automation_models import QueueItem, QueueItemStatus
items = [QueueItem.model_validate(i) for i in data.get("items", [])]
self.workflow_queue_loaded.emit(items)
# Also emit the currently running item for step progress
for item in items:
if item.status == QueueItemStatus.RUNNING:
self.workflow_item_updated.emit(item)
break
except Exception as e:
logger.error(f"Failed to load workflow queue: {e}")
@@ -1334,15 +1340,20 @@ class DAQWorker(QObject):
@Slot()
def workflow_clear_queue(self):
"""Clear all pending items from the queue."""
# Load queue first, then delete all pending items
"""Clear all non-running items from the queue via server."""
if self.__base_url is None:
return
request = QNetworkRequest(QUrl(f"{self.__base_url}/workflow/queue"))
request = QNetworkRequest(QUrl(f"{self.__base_url}/workflow/queue/clear"))
request.setRawHeader(b"Authorization", f"Bearer {self.__token}".encode())
reply = self.__net_manager.get(request)
reply.finished.connect(lambda: self._handle_clear_queue_response(reply))
request.setRawHeader(b"Content-Type", b"application/json")
reply = self.__net_manager.deleteResource(request)
reply.finished.connect(lambda: self.handle_req_response(reply))
@Slot()
def workflow_skip_sample(self):
"""Skip the entire current sample."""
self.generic_post("workflow/control/skip_sample")
def _handle_clear_queue_response(self, reply: QNetworkReply):
try:
+57 -40
View File
@@ -3,7 +3,7 @@ SSE client for workflow events.
Connects to the /workflow/sse endpoint and emits signals for:
- Runtime state changes
- Control state changes
- Control state changes
- Individual workflow events
"""
@@ -24,10 +24,10 @@ logger = setup_logger("aareGUI")
class WorkflowSSEClient(QObject):
"""
SSE client that subscribes to workflow events.
Emits signals when state changes are received.
"""
# Signals
runtime_changed = Signal(object) # RuntimeState
control_changed = Signal(object) # ControlState
@@ -35,7 +35,7 @@ class WorkflowSSEClient(QObject):
connected = Signal()
disconnected = Signal()
error = Signal(str)
def __init__(
self,
base_url: str,
@@ -43,85 +43,92 @@ class WorkflowSSEClient(QObject):
parent: QObject | None = None,
):
super().__init__(parent)
self._base_url = base_url
self._token = token
self._manager = QNetworkAccessManager(self)
self._reply: QNetworkReply | None = None
self._buffer = ""
# Reconnection
self._reconnect_timer = QTimer(self)
self._reconnect_timer.setInterval(5000) # 5 seconds
self._reconnect_timer.timeout.connect(self.connect)
self._should_reconnect = False
def connect(self) -> None:
"""Start SSE connection."""
if self._reply is not None:
return # Already connected
self._should_reconnect = True
url = QUrl(f"{self._base_url}/workflow/sse")
# Add token as query param for SSE (can't use headers easily)
url.setQuery(f"token={self._token}")
request = QNetworkRequest(url)
request.setRawHeader(b"Authorization", f"Bearer {self._token}".encode())
request.setRawHeader(b"Accept", b"text/event-stream")
request.setRawHeader(b"Cache-Control", b"no-cache")
self._reply = self._manager.get(request)
self._reply.readyRead.connect(self._on_data_ready)
self._reply.finished.connect(self._on_finished)
self._reply.errorOccurred.connect(self._on_error)
self._reconnect_timer.stop()
logger.debug("Workflow SSE: connecting...")
def disconnect(self) -> None:
"""Stop SSE connection."""
self._should_reconnect = False
self._reconnect_timer.stop()
if self._reply is not None:
self._reply.abort()
self._reply.deleteLater()
try:
self._reply.abort()
except Exception:
pass
try:
self._reply.deleteLater()
except Exception:
pass
self._reply = None
self.disconnected.emit()
@Slot()
def _on_data_ready(self) -> None:
"""Handle incoming SSE data."""
if self._reply is None:
return
data = self._reply.readAll().data().decode("utf-8")
self._buffer += data
# Process complete events (separated by double newlines)
while "\n\n" in self._buffer:
event_data, self._buffer = self._buffer.split("\n\n", 1)
self._parse_event(event_data)
try:
data = self._reply.readAll().data().decode("utf-8")
self._buffer += data
# Process complete events (separated by double newlines)
while "\n\n" in self._buffer:
event_data, self._buffer = self._buffer.split("\n\n", 1)
self._parse_event(event_data)
except Exception as e:
logger.warning(f"Workflow SSE data read error: {e}")
def _parse_event(self, event_data: str) -> None:
"""Parse a single SSE event."""
event_type = "message"
data_lines = []
for line in event_data.split("\n"):
if line.startswith("event:"):
event_type = line[6:].strip()
elif line.startswith("data:"):
data_lines.append(line[5:].strip())
if not data_lines:
return
data_str = "\n".join(data_lines)
try:
if event_type == "runtime":
runtime = RuntimeState.model_validate_json(data_str)
@@ -134,25 +141,35 @@ class WorkflowSSEClient(QObject):
self.workflow_event.emit(event)
except Exception as e:
logger.warning(f"Workflow SSE: failed to parse {event_type}: {e}")
@Slot()
def _on_finished(self) -> None:
"""Handle connection finished."""
if self._reply is not None:
self._reply.deleteLater()
try:
self._reply.deleteLater()
except Exception:
pass
self._reply = None
self._buffer = ""
self.disconnected.emit()
# Reconnect if desired
if self._should_reconnect:
logger.debug("Workflow SSE: disconnected, will reconnect...")
self._reconnect_timer.start()
@Slot(QNetworkReply.NetworkError)
def _on_error(self, error: QNetworkReply.NetworkError) -> None:
"""Handle connection error."""
error_msg = self._reply.errorString() if self._reply else str(error)
error_msg = ""
if self._reply is not None:
try:
error_msg = self._reply.errorString()
except Exception:
error_msg = str(error)
else:
error_msg = str(error)
logger.warning(f"Workflow SSE error: {error_msg}")
self.error.emit(error_msg)
self.error.emit(error_msg)