Handle async pose stop and validate POET frames

Makes pose processor shutdown non-throwing by returning a success flag from `reset()`/`shutdown()`, and wires that into the main window so users get a status message when stop is deferred during model initialization. It also refactors the worker loop to honor stop requests around backend init, cleanly close backend state in a single `finally` path, and avoid emitting init errors when a stop was intentional. In POET inference, it now validates input as 3-channel image data and enforces contiguous frame memory before inference to prevent backend/runtime issues.

Author: C-Achard

dlclivegui/gui/main_window.py

@@ -2993,14 +2993,14 @@ def _reset_active_pose_processor(
         self,
         *,
         reset_processor_plugin: bool,
-    ) -> None:
+    ) -> bool:
         if self._backend_name == "dlc":
             self._dlc.reset(
                 reset_processor_plugin=reset_processor_plugin,
             )
-            return
+            return True
 
-        self._active_pose_processor.reset()
+        return self._active_pose_processor.reset()
 
     def _stop_inference(self, show_message: bool = True) -> None:
         if self._rec_manager.is_active and self._backend_name == "dlc":
@@ -3020,7 +3020,9 @@ def _stop_inference(self, show_message: bool = True) -> None:
         self._dlc_active = False
         self._dlc_initialized = False
         # Does NOT invoke the normal rec-stop/save hooks. Persistence is processor-dependent.
-        self._reset_active_pose_processor(reset_processor_plugin=True)
+        stopped = self._reset_active_pose_processor(reset_processor_plugin=True)
+        if not stopped:
+            self.statusBar().showMessage("Stopping pose inference after model init completed...", 5000)
         self._last_pose = None
         self._overlay_renderer.clear_runtime_state()
         self._last_processor_vid_recording = False

dlclivegui/services/inference/models/poet/poet_processor.py

@@ -210,10 +210,15 @@ def get_pose(
     ) -> np.ndarray | None:
         del frame_time
 
+        frame = np.asarray(frame)
+        if frame.ndim != 3 or frame.shape[2] != 3:
+            # FIXME @C-Achard 2026/08/28 - optionally convert here as well
+            raise ValueError(f"Incorrect frame format, expected 3-channel BGR, got {frame.shape!r}")
+
         torch = _require_torch()
 
         with torch.inference_mode():
-            return self._infer(frame, torch)
+            return self._infer(np.ascontiguousarray(frame), torch)
 
     def _infer(
         self,

dlclivegui/services/inference/processor.py

@@ -56,10 +56,12 @@ def configure(self, backend_factory: Callable[[], PoseBackend]) -> None:
     def is_configured(self) -> bool:
         return self._backend_factory is not None
 
-    def reset(self) -> None:
+    def reset(self) -> bool:
         if not self._stop_worker():
-            raise RuntimeError("Failed to stop worker thread")
+            return False
+
         self._backend = None
+
         with self._stats_lock:
             self._frames_enqueued = 0
             self._frames_processed = 0
@@ -71,8 +73,10 @@ def reset(self) -> None:
             self._signal_emit_times.clear()
             self._total_process_times.clear()
 
-    def shutdown(self) -> None:
-        self.reset()
+        return True
+
+    def shutdown(self) -> bool:
+        return self.reset()
 
     def enqueue_frame(self, frame: np.ndarray, timestamp: float) -> None:
         if self._worker_thread is None:
@@ -214,83 +218,96 @@ def _process_frame(
 
         self.frame_processed.emit()
 
-    def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None:
+    def _worker_loop(
+        self,
+        init_frame: np.ndarray,
+        init_timestamp: float,
+    ) -> None:
         try:
             if self._backend_factory is None:
                 raise RuntimeError("No backend configured.")
+
             self._backend = self._backend_factory()
 
+            if self._stop_event.is_set():
+                return
+
             self._backend.init_inference(init_frame)
+
+            if self._stop_event.is_set():
+                return
+
             self.initialized.emit(True)
 
-            self._process_frame(init_frame, init_timestamp, time.perf_counter(), queue_wait_time=0.0)
+            self._process_frame(
+                init_frame,
+                init_timestamp,
+                time.perf_counter(),
+                queue_wait_time=0.0,
+            )
+
             with self._stats_lock:
                 self._frames_enqueued += 1
 
+            while not self._stop_event.is_set():
+                queue_ref = self._queue
+                if queue_ref is None:
+                    break
+
+                try:
+                    frame, timestamp, enqueued_at = queue_ref.get(timeout=0.05)
+                except queue.Empty:
+                    continue
+
+                queue_wait_time = max(
+                    0.0,
+                    time.perf_counter() - enqueued_at,
+                )
+
+                try:
+                    self._process_frame(
+                        frame,
+                        timestamp,
+                        enqueued_at,
+                        queue_wait_time=queue_wait_time,
+                    )
+                except Exception as exc:
+                    logger.exception(
+                        "Pose inference failed",
+                        exc_info=exc,
+                    )
+                    self.error.emit(str(exc))
+                finally:
+                    try:
+                        queue_ref.task_done()
+                    except ValueError:
+                        pass
+
         except Exception as exc:
-            logger.exception("Failed to initialize pose backend", exc_info=exc)
-            self.error.emit(str(exc))
-            self.initialized.emit(False)
+            if not self._stop_event.is_set():
+                logger.exception(
+                    "Failed to initialize pose backend",
+                    exc_info=exc,
+                )
+                self.error.emit(str(exc))
+                self.initialized.emit(False)
 
+        finally:
             backend = self._backend
             self._backend = None
 
             if backend is not None:
                 try:
                     backend.close()
-                except Exception as exc:
-                    logger.exception("Failed to close backend", exc_info=exc)
+                except Exception:
+                    logger.exception("Failed to close pose backend.")
 
             self._queue = None
-            self._worker_thread = None
-            return
 
-        while True:
-            if self._stop_event.is_set():
-                if self._queue is not None:
-                    try:
-                        frame, ts, enq = self._queue.get_nowait()
-                    except queue.Empty:
-                        break
-                    else:
-                        try:
-                            self._process_frame(frame, ts, enq, queue_wait_time=0.0)
-                        except Exception as exc:
-                            logger.exception("Pose inference failed", exc_info=exc)
-                            self.error.emit(str(exc))
-                        finally:
-                            try:
-                                self._queue.task_done()
-                            except ValueError:
-                                pass
-                        continue
-
-            try:
-                frame, ts, enq = self._queue.get(
-                    timeout=0.05,
-                )
-            except queue.Empty:
-                break
-            else:
-                queue_wait_time = max(0.0, time.perf_counter() - enq)
-
-            try:
-                self._process_frame(
-                    frame,
-                    ts,
-                    enq,
-                    queue_wait_time=queue_wait_time,
-                )
-            except Exception as exc:
-                logger.exception("Pose inference failed", exc_info=exc)
-                self.error.emit(str(exc))
-            finally:
-                try:
-                    self._queue.task_done()
-                except ValueError:
-                    pass
+            if self._worker_thread is threading.current_thread():
+                self._worker_thread = None
 
-        logger.info("Pose worker thread exiting")
+            logger.info("Pose worker thread exiting")
 
 
 def create_pose_processor(