diff --git a/src/aare/gui/main_window.py b/src/aare/gui/main_window.py index 218eb05c..89b25d8d 100644 --- a/src/aare/gui/main_window.py +++ b/src/aare/gui/main_window.py @@ -1403,6 +1403,12 @@ class MainWindow(QMainWindow): if hasattr(self, "workflow_sse") and self.workflow_sse is not None: self.workflow_sse.disconnect() + try: + if hasattr(self, "daq") and self.daq is not None: + self.daq.cleanup() + except Exception as e: + logger.warning(f"Failed to clean up DAQ worker: {e}") + for attr_name in ( "camera_thread", "prediction_thread", diff --git a/src/aare/gui/threads/daq_worker.py b/src/aare/gui/threads/daq_worker.py index b5b72f13..c6e72b29 100644 --- a/src/aare/gui/threads/daq_worker.py +++ b/src/aare/gui/threads/daq_worker.py @@ -49,6 +49,7 @@ class DAQWorker(QObject): automation_progress = Signal(object) gui_sessions_loaded = Signal(list) gui_close_requested = Signal(int, int, str) + recovery_action_completed = Signal(str) #dedicated signals for polled device errors and request-time errors polled_devices_status = Signal(str, bool) # (message, is_error) @@ -104,6 +105,8 @@ class DAQWorker(QObject): self.__timer.timeout.connect(self.regular_update) self.__timer.start() self.__counter = 0 + self._automation_progress_buffer = "" + self._cleanup_done = False self._last_auth_error_log_ts = 0.0 self._auth_error_min_interval = 10.0 @@ -1227,40 +1230,60 @@ class DAQWorker(QObject): success=progress_payload.get("success"), ) + def _handle_automation_progress_event(self, payload: str) -> None: + if not payload: + return + + logger.info(f"[automation_progress raw] {payload}") + outer = json.loads(payload) + progress_payload = outer.get("progress") + if progress_payload is None: + logger.info("[automation_progress raw] no progress payload in SSE event") + return + + progress = self._parse_automation_progress(progress_payload) + self.automation_progress.emit(progress) + + current = progress.current_step or "Idle" + if progress.finished: + if progress.success is True: + logger.info(f"Automation progress: {current} - finished successfully") + elif progress.success is False: + logger.info(f"Automation progress: {current} - finished with error") + else: + logger.info(f"Automation progress: {current} - finished") + else: + logger.info(f"Automation progress: {current}") + + def _process_automation_progress_buffer(self) -> None: + while "\n\n" in self._automation_progress_buffer: + event_data, self._automation_progress_buffer = self._automation_progress_buffer.split("\n\n", 1) + + data_lines: list[str] = [] + for line in event_data.splitlines(): + if line.startswith("data:"): + data_lines.append(line[5:].lstrip()) + + if not data_lines: + continue + + payload = "\n".join(data_lines) + self._handle_automation_progress_event(payload) + def _read_automation_progress_stream(self, reply: QNetworkReply): try: chunk = reply.readAll().data().decode("utf-8") - for line in chunk.splitlines(): - if line.startswith("data:"): - payload = line[5:].strip() - if not payload: - continue + if not chunk: + return - logger.info(f"[automation_progress raw] {payload}") - outer = json.loads(payload) - progress_payload = outer.get("progress") - if not progress_payload: - logger.info("[automation_progress raw] no progress payload in SSE event") - continue - - progress = self._parse_automation_progress(progress_payload) - self.automation_progress.emit(progress) - - current = progress.current_step or "Idle" - if progress.finished: - if progress.success is True: - logger.info(f"Automation progress: {current} - finished successfully") - elif progress.success is False: - logger.info(f"Automation progress: {current} - finished with error") - else: - logger.info(f"Automation progress: {current} - finished") - else: - logger.info(f"Automation progress: {current}") + self._automation_progress_buffer += chunk + self._process_automation_progress_buffer() except Exception as e: logger.error(f"Automation progress stream parse error: {e}") def _restart_automation_progress_stream(self): self._automation_progress_stream_reply = None + self._automation_progress_buffer = "" if self.__base_url is not None: QTimer.singleShot(1000, self.start_automation_progress_stream) @@ -1271,6 +1294,8 @@ class DAQWorker(QObject): if self._automation_progress_stream_reply is not None: return + self._automation_progress_buffer = "" + request = QNetworkRequest(QUrl(f"{self.__base_url}/sse/automation_progress")) request.setRawHeader(b"Authorization", f"Bearer {self.__token}".encode("utf-8")) reply = self.__net_manager.get(request) @@ -1919,4 +1944,44 @@ class DAQWorker(QObject): request.setRawHeader(b"Authorization", f"Bearer {self.__token}".encode("utf-8")) request.setRawHeader(b"Content-Type", b"application/json") reply = self.__net_manager.post(request, QByteArray(b"")) - reply.finished.connect(lambda: reply.deleteLater()) \ No newline at end of file + reply.finished.connect(lambda: reply.deleteLater()) + + def cleanup(self) -> None: + if getattr(self, "_cleanup_done", False): + return + self._cleanup_done = True + + try: + if hasattr(self, "_baton_timeout_timer") and self._baton_timeout_timer is not None: + self._baton_timeout_timer.stop() + except Exception as e: + logger.warning(f"Failed to stop _baton_timeout_timer: {e}") + + try: + if hasattr(self, "_DAQWorker__timer") and self.__timer is not None: + self.__timer.stop() + except Exception as e: + logger.warning(f"Failed to stop __timer: {e}") + + for attr_name in ( + "_baton_stream_reply", + "_face_detection_stream_reply", + "_automation_progress_stream_reply", + ): + reply = getattr(self, attr_name, None) + if reply is None: + continue + + try: + reply.abort() + except Exception as e: + logger.warning(f"Failed to abort {attr_name}: {e}") + + try: + reply.deleteLater() + except Exception as e: + logger.warning(f"Failed to delete {attr_name}: {e}") + + setattr(self, attr_name, None) + + self._automation_progress_buffer = "" \ No newline at end of file