From 303173ae82d0c356b12712a7093f4f8ad2b10a40 Mon Sep 17 00:00:00 2001 From: appleb_m Date: Thu, 26 Mar 2026 17:29:17 +0100 Subject: [PATCH] Automation 2.0: WIP added endpoints to backend, frontend connected and panel made, now debugging --- src/aare/common/automation_queue_manager.py | 5 + src/aare/common/automation_workflow.py | 8 +- src/aare/daq/automation_api_router.py | 71 +++++- src/aare/daq/automation_runner.py | 136 ++++++---- src/aare/gui/gui.py | 4 +- src/aare/gui/main_window.py | 1 + src/aare/gui/panels/automation_panel.py | 260 +++++++++++++------- src/aare/gui/threads/daq_worker.py | 23 +- src/aare/gui/threads/workflow_sse_client.py | 97 +++++--- 9 files changed, 415 insertions(+), 190 deletions(-) diff --git a/src/aare/common/automation_queue_manager.py b/src/aare/common/automation_queue_manager.py index 54e4d2e2..4b63acdd 100644 --- a/src/aare/common/automation_queue_manager.py +++ b/src/aare/common/automation_queue_manager.py @@ -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)}) diff --git a/src/aare/common/automation_workflow.py b/src/aare/common/automation_workflow.py index 98f26ac1..dacd48d0 100644 --- a/src/aare/common/automation_workflow.py +++ b/src/aare/common/automation_workflow.py @@ -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( diff --git a/src/aare/daq/automation_api_router.py b/src/aare/daq/automation_api_router.py index 1236ffc0..bc4f67eb 100644 --- a/src/aare/daq/automation_api_router.py +++ b/src/aare/daq/automation_api_router.py @@ -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, ) diff --git a/src/aare/daq/automation_runner.py b/src/aare/daq/automation_runner.py index 4626e601..8b672235 100644 --- a/src/aare/daq/automation_runner.py +++ b/src/aare/daq/automation_runner.py @@ -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") diff --git a/src/aare/gui/gui.py b/src/aare/gui/gui.py index 284ec734..d5957486 100644 --- a/src/aare/gui/gui.py +++ b/src/aare/gui/gui.py @@ -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 diff --git a/src/aare/gui/main_window.py b/src/aare/gui/main_window.py index 67b75a75..be28070b 100644 --- a/src/aare/gui/main_window.py +++ b/src/aare/gui/main_window.py @@ -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) diff --git a/src/aare/gui/panels/automation_panel.py b/src/aare/gui/panels/automation_panel.py index e53ea753..3fd82e5c 100644 --- a/src/aare/gui/panels/automation_panel.py +++ b/src/aare/gui/panels/automation_panel.py @@ -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() \ No newline at end of file + 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() \ No newline at end of file diff --git a/src/aare/gui/threads/daq_worker.py b/src/aare/gui/threads/daq_worker.py index df49f5f7..dffd18e0 100644 --- a/src/aare/gui/threads/daq_worker.py +++ b/src/aare/gui/threads/daq_worker.py @@ -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: diff --git a/src/aare/gui/threads/workflow_sse_client.py b/src/aare/gui/threads/workflow_sse_client.py index f1f4d857..ec264106 100644 --- a/src/aare/gui/threads/workflow_sse_client.py +++ b/src/aare/gui/threads/workflow_sse_client.py @@ -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) \ No newline at end of file