GUI API¶

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle(_BASE_TITLE)
        with as_file(files(__package__) / "resources" / "icon.ico") as icon_path:
            self.setWindowIcon(QIcon(str(icon_path)))

        # The window owns only the experiment, which owns its videos (results).
        # It is None until a video is opened or an experiment folder is loaded;
        # the experiment's ``folder`` is the save location (None until saved).
        self.experiment: Experiment | None = None
        self._thread: threading.Thread | None = None
        self._worker: (
            ObjectTrackingWorker | FaceDetectionWorker | BodyPoseWorker | None
        ) = None
        self._in_setup = False
        #: Remaining step types queued by "Run all"; consumed one at a time as
        #: each step finishes, so later steps see earlier steps' results.
        self._pending_steps: list[type] = []

        self._build_menu_bar()

        self.open_button = QPushButton("Open video…")
        self.open_button.clicked.connect(self._choose_video)

        self.file_label = QLabel("No file selected")

        self.progress_bar = QProgressBar()
        self.progress_bar.setVisible(False)

        self.cancel_button = QPushButton("Cancel")
        self.cancel_button.setVisible(False)
        self.cancel_button.clicked.connect(self._cancel_run)

        self.video_viewer = VideoViewer()

        top_bar = QHBoxLayout()
        top_bar.addWidget(self.open_button)
        top_bar.addWidget(self.file_label, stretch=1)

        bottom_bar = QHBoxLayout()
        bottom_bar.addWidget(self.progress_bar, stretch=1)
        bottom_bar.addWidget(self.cancel_button)

        layout = QVBoxLayout()
        layout.addLayout(top_bar)
        layout.addWidget(self.video_viewer, stretch=1)
        layout.addLayout(bottom_bar)

        central = QWidget()
        central.setLayout(layout)
        self.setCentralWidget(central)

        # The pipeline editor is the authority for the experiment's pipeline. It
        # lives in a movable dock so it is easy to relocate later.
        self.pipeline_editor = PipelineEditor()
        self.pipeline_editor.setEnabled(False)
        self.pipeline_editor.changed.connect(self._on_pipeline_edited)
        self.pipeline_editor.run_requested.connect(self._start_step)
        self.pipeline_editor.run_all_requested.connect(self._start_run_all)
        self.pipeline_dock = QDockWidget("Pipeline", self)
        self.pipeline_dock.setWidget(self.pipeline_editor)
        self.addDockWidget(Qt.DockWidgetArea.RightDockWidgetArea, self.pipeline_dock)

        # How to run each step: its worker, readiness check and window plumbing.
        # Built last since it closes over ``self.video_viewer``.
        self._step_runners: dict[type, _StepRunner] = {
            ObjectTrackingStep: _StepRunner(
                worker_cls=ObjectTrackingWorker,
                ready=lambda video: video.video_path is not None,
                begin=lambda video, k: video.begin_object_tracking(k),
                live_frame_slot=self.video_viewer.show_live_frame,
                on_finished=self._on_finished,
            ),
            FaceDetectionStep: _StepRunner(
                worker_cls=FaceDetectionWorker,
                ready=lambda video: video.data is not None,
                begin=lambda video, k: video.begin_face_detection(k),
                live_frame_slot=self.video_viewer.show_live_face_frame,
                on_finished=self._on_face_finished,
            ),
            BodyPoseStep: _StepRunner(
                worker_cls=BodyPoseWorker,
                ready=lambda video: video.data is not None,
                begin=lambda video, k: video.begin_body_pose_detection(k),
                live_frame_slot=self.video_viewer.show_live_pose_frame,
                on_finished=self._on_pose_finished,
            ),
        }
        self._update_step_availability()

    def _video(self) -> Video | None:
        """The current input's :class:`Video` (its results), or ``None``."""
        if self.experiment is None:
            return None
        return self.experiment.video(self.experiment.config.inputs[0])

    def _build_menu_bar(self) -> None:
        file_menu = self.menuBar().addMenu("&File")

        self.new_action = QAction("&New", self)
        self.new_action.setShortcut(QKeySequence.StandardKey.New)
        self.new_action.triggered.connect(self._new_experiment)
        file_menu.addAction(self.new_action)

        self.open_action = QAction("&Open…", self)
        self.open_action.setShortcut(QKeySequence.StandardKey.Open)
        self.open_action.triggered.connect(self._choose_experiment)
        file_menu.addAction(self.open_action)

        self.save_action = QAction("&Save", self)
        self.save_action.setShortcut(QKeySequence.StandardKey.Save)
        self.save_action.triggered.connect(self._save_experiment)
        self.save_action.setEnabled(False)
        file_menu.addAction(self.save_action)

        file_menu.addSeparator()

        self.exit_action = QAction("E&xit", self)
        self.exit_action.setShortcut(QKeySequence.StandardKey.Quit)
        self.exit_action.triggered.connect(self.close)
        file_menu.addAction(self.exit_action)

    def _new_experiment(self) -> None:
        """Reset to an empty state, discarding the current experiment/results."""
        if self._thread is not None:
            return
        # Emptying the current video clears the viewer's overlays (it still holds
        # that instance) before the experiment is dropped.
        video = self._video()
        if video is not None:
            video.clear()
        self.experiment = None
        self._update_title()
        self.file_label.setText("No file selected")
        self.video_viewer.refresh_overlays()
        self.video_viewer.enable_controls(False)
        self._update_step_availability()
        self.save_action.setEnabled(False)
        self._bind_editor_to_experiment()

    def _save_experiment(self) -> None:
        """Write the experiment (config and any computed results) to its folder."""
        if self.experiment is None:
            return
        if self.experiment.folder is None:
            folder = QFileDialog.getExistingDirectory(self, "Save experiment to folder")
            if not folder:
                return
            self.experiment.save(Path(folder))
        else:
            self.experiment.save()
        self._update_title()
        self.statusBar().showMessage(f"Saved experiment to {self.experiment.folder}")

    def _update_title(self) -> None:
        """Show the open experiment folder in the title bar, if any."""
        folder = self.experiment.folder if self.experiment is not None else None
        if folder is not None:
            self.setWindowTitle(f"{_BASE_TITLE} :: [{folder.name}]")
        else:
            self.setWindowTitle(_BASE_TITLE)

    def _bind_editor_to_experiment(self) -> None:
        """Populate the pipeline editor from the experiment (or disable it)."""
        if self.experiment is None:
            self.pipeline_editor.reset()
            self.pipeline_editor.setEnabled(False)
            return
        self.pipeline_editor.setEnabled(True)
        self.pipeline_editor.set_from(self.experiment.config)

    def _on_pipeline_edited(self) -> None:
        """Adopt the editor's pipeline as the experiment's, when it is valid."""
        if self.experiment is None:
            return
        try:
            self.pipeline_editor.apply_to(self.experiment.config)
        except (ValidationError, ValueError):
            self.statusBar().showMessage("Pipeline has invalid settings; not applied")

    def _choose_video(self) -> None:
        path, _ = QFileDialog.getOpenFileName(
            self,
            "Open video",
            "",
            "Video files (*.mp4 *.avi *.mov *.mkv);;All files (*)",
        )
        if path:
            self._load_video(Path(path))

    def _load_video(self, path: Path) -> None:
        # Opening a video starts a fresh, unsaved single-input experiment.
        config = ExperimentConfig(
            name=path.stem or "experiment",
            inputs=[VideoInput(id=path.stem or "video", path=path)],
        )
        experiment = Experiment(config)
        try:
            self.video_viewer.load(experiment.video(config.inputs[0]))
        except OSError as exc:
            QMessageBox.critical(self, "Could not open video", str(exc))
            return
        self.experiment = experiment
        self.file_label.setText(str(path))
        self._update_title()
        self._bind_editor_to_experiment()
        self.save_action.setEnabled(True)
        # A fresh video has no object tracking results yet, so later passes wait.
        self._update_step_availability()

    def load_experiment(self, folder: str | Path) -> None:
        """Open an experiment folder, as the File ▸ Open action does.

        Public entry point for launching the GUI on an experiment directly.
        Invalid folders are reported to the user, not raised.
        """
        self._load_experiment(Path(folder))

    def _choose_experiment(self) -> None:
        folder = QFileDialog.getExistingDirectory(self, "Open experiment folder")
        if folder:
            self._load_experiment(Path(folder))

    def _load_experiment(self, folder: Path) -> None:
        try:
            experiment = Experiment.load(folder)
        except (OSError, ValueError, ValidationError) as exc:
            QMessageBox.critical(self, "Could not open experiment", str(exc))
            return

        # The GUI shows a single video, so open the experiment's first input; the
        # experiment loads that video's cached results (if any) on access.
        spec = experiment.config.inputs[0]
        video_path = experiment.resolved_input_path(spec)
        if not video_path.exists():
            QMessageBox.critical(self, "Video not found", str(video_path))
            return
        video = experiment.video(spec)
        try:
            self.video_viewer.load(video)
        except OSError as exc:
            QMessageBox.critical(self, "Could not open video", str(exc))
            return

        self.experiment = experiment
        self.file_label.setText(str(video_path))
        self._update_title()
        self._bind_editor_to_experiment()
        self.save_action.setEnabled(True)
        self.video_viewer.refresh_overlays()
        self._update_step_availability()
        if video.data is not None:
            self.statusBar().showMessage(
                f"Loaded cached results from {experiment.output_path(spec)}"
            )

    def _update_step_availability(self) -> None:
        """Enable each step's "Run" button (and "Run all") once its inputs are ready.

        Object tracking needs a video; later passes run on tracked boxes, so
        they wait for object tracking's results. Whether a *run* is currently
        in progress is handled separately, by disabling the whole pipeline
        editor (see ``_set_running``).
        """
        video = self._video()
        has_video = video is not None and video.video_path is not None
        has_tracks = video is not None and video.data is not None
        self.pipeline_editor.set_run_enabled(ObjectTrackingStep, has_video)
        self.pipeline_editor.set_run_enabled(FaceDetectionStep, has_tracks)
        self.pipeline_editor.set_run_enabled(BodyPoseStep, has_tracks)
        self.pipeline_editor.set_run_all_enabled(has_video)

    def _step_config(self, step_type):
        """The editor's validated config for a step, or None (with an alert)."""
        try:
            return self.pipeline_editor.config_for(step_type)
        except (ValidationError, ValueError) as exc:
            QMessageBox.critical(self, "Invalid settings", str(exc))
            return None

    @Slot(object)
    def _start_step(self, step_type: type) -> None:
        """Run one pipeline step, using the editor's current arguments for it."""
        if self._thread is not None:
            return
        video = self._video()
        runner = self._step_runners[step_type]
        if video is None or not runner.ready(video):
            self._pending_steps = []
            return
        step = self._step_config(step_type)
        if step is None:
            self._pending_steps = []
            return

        # Discards that step's previous results; keeps everything else (e.g. a
        # face/pose pass keeps the tracked boxes it runs over). The embedding
        # budget comes from the step config, so the GUI and CLI behave identically.
        runner.begin(video, getattr(step, "embeddings_per_track", 0))
        self._begin_run()

        self._worker = runner.worker_cls(video, step)
        self._worker.new_frame.connect(self._on_new_frame)
        self._worker.new_frame.connect(runner.live_frame_slot)
        self._worker.finished.connect(runner.on_finished)
        self._worker.failed.connect(self._on_failed)
        self._worker.cancelled.connect(self._on_cancelled)

        self._thread = threading.Thread(target=self._worker.run, daemon=True)
        self._thread.start()

    def _start_run_all(self) -> None:
        """Run every enabled pipeline step in order, one after another."""
        if self._thread is not None:
            return
        try:
            steps = self.pipeline_editor.enabled_steps()
        except (ValidationError, ValueError) as exc:
            QMessageBox.critical(self, "Invalid settings", str(exc))
            return
        self._pending_steps = [type(step) for step in steps]
        self._continue_run_all()

    def _continue_run_all(self) -> None:
        """Start the next step queued by "Run all", if any are left."""
        if self._pending_steps:
            self._start_step(self._pending_steps.pop(0))

    def _begin_run(self) -> None:
        """Shared start-up for object tracking and later detection runs."""
        self._set_running(True)
        # Weights are built/downloaded before the first frame is processed, so
        # show a busy bar until the first frame arrives.
        self._in_setup = True
        self.progress_bar.setRange(0, 0)
        self.progress_bar.setTextVisible(True)
        self.progress_bar.setFormat("Downloading weights…")

    @Slot(object)
    def _on_new_frame(self, frame) -> None:
        total = self.video_viewer.frame_count
        if self._in_setup:
            # First frame processed: switch from the busy "downloading" bar to a
            # determinate progress bar (a 0..0 range stays busy if total unknown).
            self._in_setup = False
            self.progress_bar.setRange(0, total)
            self.progress_bar.setFormat("%p%" if total else "Object tracking…")
        if total:
            self.progress_bar.setValue(frame.frame_idx)

    def _cancel_run(self) -> None:
        if self._worker is not None:
            self._worker.cancel()
        self.cancel_button.setEnabled(False)
        self.cancel_button.setText("Cancelling…")

    @Slot()
    def _on_finished(self) -> None:
        data = self._video().data
        self.statusBar().showMessage(
            f"Object tracking finished: {data['track_id'].nunique()} tracklets, "
            f"{len(data)} detections"
        )
        self._set_running(False)
        self._continue_run_all()

    @Slot()
    def _on_face_finished(self) -> None:
        data = self._video().data
        n_faces = int(data["face_score"].notna().sum())
        self.statusBar().showMessage(
            f"Face detection finished: {n_faces} faces over {len(data)} detections"
        )
        self._set_running(False)
        self._continue_run_all()

    @Slot()
    def _on_pose_finished(self) -> None:
        data = self._video().data
        n_poses = int(data["pose_score"].notna().sum())
        self.statusBar().showMessage(
            f"Body pose detection finished: {n_poses} poses over {len(data)} detections"
        )
        self._set_running(False)
        self._continue_run_all()

    @Slot(str, str)
    def _on_failed(self, message: str, details: str) -> None:
        # A failure stops a "Run all" chain rather than pressing on regardless.
        self._pending_steps = []
        dialog = QMessageBox(self)
        dialog.setIcon(QMessageBox.Icon.Critical)
        dialog.setWindowTitle(f"{self._worker.operation_name} failed")
        dialog.setText(message)
        dialog.setDetailedText(details)
        dialog.exec()
        self._set_running(False)

    @Slot()
    def _on_cancelled(self) -> None:
        # Cancelling one step cancels the rest of a "Run all" chain too.
        self._pending_steps = []
        self.statusBar().showMessage(f"{self._worker.operation_name} cancelled")
        self._set_running(False)

    def _set_running(self, running: bool) -> None:
        if not running:
            # background thread has reported back; drop our references to it.
            self._thread = None
            self._worker = None
            self.video_viewer.refresh_overlays()
            self._update_step_availability()
        self.open_button.setEnabled(not running)
        self.new_action.setEnabled(not running)
        self.open_action.setEnabled(not running)
        self.save_action.setEnabled(not running and self.experiment is not None)
        self.pipeline_editor.setEnabled(not running and self.experiment is not None)
        self.video_viewer.enable_controls(not running)
        self.progress_bar.setVisible(running)
        self.cancel_button.setVisible(running)
        self.cancel_button.setEnabled(True)
        self.cancel_button.setText("Cancel")

    def closeEvent(self, event) -> None:
        if self._worker is not None:
            self._worker.cancel()
        if self._thread is not None:
            self._thread.join(timeout=5.0)
        super().closeEvent(event)

class PipelineEditor(QWidget):
    """Edit an experiment's pipeline steps and their arguments.

    Populate from an experiment with :meth:`set_from` (or :meth:`reset` to
    defaults), and write the edited values back into one with :meth:`apply_to`.
    ``changed`` fires on any toggle or field edit. Each step has its own "Run"
    button (``run_requested``, with the step's type); ``run_all_requested`` fires
    from the button that runs every enabled step in order. Running is out of
    scope for this widget -- it only reports the requests, and its buttons'
    enabled state is driven from outside via
    :meth:`set_run_enabled`/:meth:`set_run_all_enabled`.
    """

    changed = Signal()
    run_requested = Signal(object)
    run_all_requested = Signal()

    def __init__(self, parent: QWidget | None = None) -> None:
        super().__init__(parent)
        self._sections: list[_StepSection] = []
        layout = QVBoxLayout(self)
        for attr_name, step_type, title, optional in _STEPS:
            section = _StepSection(attr_name, step_type, title, optional)
            section.changed.connect(self.changed)
            section.run_requested.connect(
                lambda step_type=step_type: self.run_requested.emit(step_type)
            )
            self._sections.append(section)
            layout.addWidget(section)

        self.run_all_button = QPushButton("Run all")
        self.run_all_button.setEnabled(False)
        self.run_all_button.clicked.connect(self.run_all_requested)
        layout.addWidget(self.run_all_button)
        layout.addStretch(1)

    def set_from(self, config: ExperimentConfig) -> None:
        """Populate the editor from ``config``'s pipeline (no ``changed``)."""
        for section in self._sections:
            section.blockSignals(True)
            section.set_from(getattr(config, section.attr_name))
            section.blockSignals(False)

    def reset(self) -> None:
        """Reset every step to its defaults, optional steps switched off."""
        for section in self._sections:
            section.blockSignals(True)
            section.reset()
            section.blockSignals(False)

    def apply_to(self, config: ExperimentConfig) -> None:
        """Write the edited steps back onto ``config``'s pipeline fields.

        Disabled optional steps become ``None``. All steps are validated before
        anything is assigned, so an invalid field leaves ``config`` intact.
        Raises :class:`pydantic.ValidationError` / :class:`ValueError` if any
        step's arguments are invalid.
        """
        values = {
            s.attr_name: (s.to_step() if s.is_enabled() else None)
            for s in self._sections
        }
        for name, value in values.items():
            setattr(config, name, value)

    def enabled_steps(self) -> list[StepSpec]:
        """The enabled steps, in order, built and validated from the widgets.

        Raises :class:`pydantic.ValidationError` / :class:`ValueError` if any
        enabled step's arguments are invalid.
        """
        return [s.to_step() for s in self._sections if s.is_enabled()]

    def _section(self, step_type: type) -> _StepSection:
        """The section editing ``step_type``, or ``KeyError`` if unknown."""
        for section in self._sections:
            if section.step_type is step_type:
                return section
        raise KeyError(step_type)

    def config_for(self, step_type: type) -> StepSpec:
        """The validated config for one step, whether or not it is enabled.

        Lets an interactive run of a single pass use the arguments the user has
        set, independent of whether the step is toggled into the saved pipeline.
        Raises :class:`pydantic.ValidationError` / :class:`ValueError` if invalid.
        """
        return self._section(step_type).to_step()

    def set_run_enabled(self, step_type: type, enabled: bool) -> None:
        """Enable/disable one step's "Run" button (e.g. while its inputs aren't ready)."""
        self._section(step_type).set_run_enabled(enabled)

    def set_run_all_enabled(self, enabled: bool) -> None:
        self.run_all_button.setEnabled(enabled)

class VideoViewer(QWidget):
    """Display a video with play/pause, a seek slider and a frame spinbox."""

    frame_changed = Signal(int)

    def __init__(self, parent: QWidget | None = None) -> None:
        super().__init__(parent)

        self._capture: cv2.VideoCapture | None = None
        self._frame_count = 0
        self._fps = 25.0
        self._current = 0

        # the video being displayed; supplies the boxes to draw per frame
        self._video: Video | None = None
        self._overlay_items: list[QGraphicsItem] = []

        # video display
        self._scene = QGraphicsScene(self)
        self._pixmap_item = QGraphicsPixmapItem()
        self._scene.addItem(self._pixmap_item)
        self._view = QGraphicsView(self._scene)
        self._view.setRenderHint(QPainter.RenderHint.SmoothPixmapTransform)
        self._view.setAlignment(Qt.AlignmentFlag.AlignCenter)

        # controls
        self._play_button = QPushButton("Play")
        self._play_button.setCheckable(True)
        self._play_button.toggled.connect(self._on_play_toggled)

        self._slider = QSlider(Qt.Orientation.Horizontal)
        self._slider.setEnabled(False)
        self._slider.valueChanged.connect(self.set_frame)

        self._spinbox = QSpinBox()
        self._spinbox.setEnabled(False)
        self._spinbox.valueChanged.connect(self.set_frame)

        self._total_label = QLabel("/ 0")

        controls = QHBoxLayout()
        controls.addWidget(self._play_button)
        controls.addWidget(self._slider, stretch=1)
        controls.addWidget(self._spinbox)
        controls.addWidget(self._total_label)

        layout = QVBoxLayout(self)
        layout.addWidget(self._view, stretch=1)
        layout.addLayout(controls)

        # playback timer
        self._timer = QTimer(self)
        self._timer.timeout.connect(self._advance)

    def load(self, video: Video) -> None:
        """Display ``video``, showing its first frame and its boxes (if any)."""
        self._stop()
        if self._capture is not None:
            self._capture.release()

        capture = cv2.VideoCapture(str(video.video_path))
        if not capture.isOpened():
            raise OSError(f"Could not open video: {video.video_path}")

        self._video = video
        self._capture = capture
        self._fps = capture.get(cv2.CAP_PROP_FPS) or 25.0
        self._timer.setInterval(int(1000 / self._fps))

        count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT)) or 0
        for control in (self._slider, self._spinbox):
            control.setEnabled(count > 0)
            control.setMinimum(0)
        self._set_frame_count(count)

        self._current = -1
        self.set_frame(0)
        self._fit()

    def set_frame(self, index: int) -> None:
        """Display the frame at ``index`` (0-based), with its tracklet boxes."""
        if self._goto(index):
            self.refresh_overlays()

    @Slot(object)
    def show_live_frame(self, frame) -> None:
        """Display a freshly tracked frame and draw its boxes directly.

        Connected to the object tracking worker's per-frame signal; ``frame`` is
        a BoxMOT per-frame result with 1-based indexing.
        """
        self._goto(frame.frame_idx - 1)
        self._draw_boxes(boxes_from_tracks(frame.tracks))

    @Slot(object)
    def show_live_face_frame(self, result) -> None:
        """Display a freshly face-detected frame, with person boxes and faces.

        Connected to the face-detection worker's per-frame signal; ``result`` is
        a :class:`FaceFrameResult` with 0-based indexing. The person boxes come
        from the already-tracked video, the faces straight from the result.
        """
        self._goto(result.frame_idx)
        self._clear_overlays()
        if self._video is not None:
            for box in self._video.boxes_for_frame(self._current):
                self._add_box(box)
        for face in result.faces:
            self._add_face(face)

    @Slot(object)
    def show_live_pose_frame(self, result) -> None:
        """Display a freshly pose-detected frame, with person boxes and poses.

        Connected to the body-pose worker's per-frame signal; ``result`` is a
        :class:`PoseFrameResult` with 0-based indexing. The person boxes come
        from the already-tracked video, the poses straight from the result.
        """
        self._goto(result.frame_idx)
        self._clear_overlays()
        if self._video is not None:
            for box in self._video.boxes_for_frame(self._current):
                self._add_box(box)
        for pose in result.poses:
            self._add_pose(pose)

    def enable_controls(self, enable: bool) -> None:
        """Enable or disable the play button and seek controls."""
        if not enable:
            self._stop()
        has_frames = self._frame_count > 0
        self._play_button.setEnabled(enable and has_frames)
        self._slider.setEnabled(enable and has_frames)
        self._spinbox.setEnabled(enable and has_frames)

    def refresh_overlays(self) -> None:
        """Redraw the current frame's person boxes and any detected faces."""
        self._clear_overlays()
        if self._video is None or self._current < 0:
            return
        for box in self._video.boxes_for_frame(self._current):
            self._add_box(box)
        for pose in self._video.poses_for_frame(self._current):
            self._add_pose(pose)
        for face in self._video.faces_for_frame(self._current):
            self._add_face(face)

    @property
    def current_frame(self) -> int:
        return self._current

    @property
    def frame_count(self) -> int:
        return self._frame_count

    def _goto(self, index: int) -> bool:
        """Show the video image at ``index`` and sync controls.

        Returns ``True`` if the displayed frame actually changed, so callers can
        decide whether overlays need redrawing.
        """
        if self._capture is None or self._frame_count == 0:
            return False
        index = max(0, min(int(index), self._frame_count - 1))
        if index == self._current:
            return False

        index, frame = self._read(index)
        if frame is None or index == self._current:
            # Nothing decoded, or _read stepped back to the frame already shown.
            return False
        self._current = index
        self._show(frame)

        # Keep slider/spinbox in sync without re-triggering set_frame.
        for control in (self._slider, self._spinbox):
            control.blockSignals(True)
            control.setValue(index)
            control.blockSignals(False)

        self.frame_changed.emit(index)
        return True

    def _draw_boxes(self, boxes: list[BoundingBox]) -> None:
        self._clear_overlays()
        for box in boxes:
            self._add_box(box)

    def _set_frame_count(self, count: int) -> None:
        """Set the frame count and update the slider/spinbox range and label."""
        self._frame_count = max(0, count)
        last = max(0, self._frame_count - 1)
        for control in (self._slider, self._spinbox):
            control.blockSignals(True)
            control.setMaximum(last)
            control.blockSignals(False)
        self._total_label.setText(f"/ {self._frame_count}")

    def _read(self, index: int):
        """Read the frame at ``index``, stepping back to the last decodable one.

        ``CAP_PROP_FRAME_COUNT`` over-estimates for many codecs, so the trailing
        frames it promises may not actually decode. When a read fails we treat
        everything from ``index`` on as non-existent, shrink the frame count to
        match, and retry the frame before it. Returns
        ``(actual_index, frame)``, or ``(-1, None)`` if nothing decodes.
        """
        # The capture cursor sits at _current + 1 after the last read, so only
        # seek (expensive) when the requested frame isn't the next one.
        sequential = index == self._current + 1
        while index >= 0:
            if not sequential:
                self._capture.set(cv2.CAP_PROP_POS_FRAMES, index)
            ok, frame = self._capture.read()
            if ok:
                return index, frame
            self._set_frame_count(index)
            index -= 1
            sequential = False
        return -1, None

    def _show(self, frame) -> None:
        height, width = frame.shape[:2]
        image = QImage(
            frame.data, width, height, frame.strides[0], QImage.Format.Format_BGR888
        )
        self._pixmap_item.setPixmap(QPixmap.fromImage(image))
        self._scene.setSceneRect(0, 0, width, height)

    def _clear_overlays(self) -> None:
        for item in self._overlay_items:
            self._scene.removeItem(item)
        self._overlay_items.clear()

    def _add_rect(self, box: BoundingBox, style: Qt.PenStyle) -> None:
        """Draw ``box`` as a rectangle coloured by its id, in the given pen style."""
        rect = QGraphicsRectItem(box.x1, box.y1, box.x2 - box.x1, box.y2 - box.y1)
        pen = QPen(get_color(box.track_id))
        pen.setStyle(style)
        # constant on-screen pen width regardless of zoom
        pen.setCosmetic(True)
        pen.setWidth(2)
        rect.setPen(pen)
        self._scene.addItem(rect)
        self._overlay_items.append(rect)

    def _add_box(self, box: BoundingBox) -> None:
        self._add_rect(box, Qt.PenStyle.SolidLine)

        label = QGraphicsSimpleTextItem(str(box.track_id))
        label.setBrush(QBrush(get_color(box.track_id)))
        # constant on-screen label size regardless of zoom
        label.setFlag(QGraphicsItem.GraphicsItemFlag.ItemIgnoresTransformations)
        label.setPos(box.x1, box.y1)
        self._scene.addItem(label)
        self._overlay_items.append(label)

    def _add_face(self, face: FaceBox) -> None:
        # dashed, so the face box reads as distinct from its person box
        self._add_rect(face.box, Qt.PenStyle.DashLine)

        color = get_color(face.box.track_id)
        for px, py in face.landmarks:
            # a small constant-size dot regardless of zoom, centred on the point
            dot = QGraphicsEllipseItem(-2.0, -2.0, 4.0, 4.0)
            dot.setBrush(QBrush(color))
            dot.setPen(QPen(Qt.PenStyle.NoPen))
            dot.setFlag(QGraphicsItem.GraphicsItemFlag.ItemIgnoresTransformations)
            dot.setPos(px, py)
            self._scene.addItem(dot)
            self._overlay_items.append(dot)

    def _add_pose(self, pose: BodyPose) -> None:
        color = get_color(pose.box.track_id)
        pen = QPen(color)
        pen.setCosmetic(True)
        pen.setWidth(2)

        visible = [
            score > 0.0 and isfinite(px) and isfinite(py)
            for px, py, score in pose.keypoints
        ]
        for start, end in SKELETON:
            if start >= len(pose.keypoints) or end >= len(pose.keypoints):
                continue
            if not (visible[start] and visible[end]):
                continue
            x1, y1, _ = pose.keypoints[start]
            x2, y2, _ = pose.keypoints[end]
            line = QGraphicsLineItem(x1, y1, x2, y2)
            line.setPen(pen)
            self._scene.addItem(line)
            self._overlay_items.append(line)

        for px, py, score in pose.keypoints:
            if not (score > 0.0 and isfinite(px) and isfinite(py)):
                continue
            dot = QGraphicsEllipseItem(-2.0, -2.0, 4.0, 4.0)
            dot.setBrush(QBrush(color))
            dot.setPen(QPen(Qt.PenStyle.NoPen))
            dot.setFlag(QGraphicsItem.GraphicsItemFlag.ItemIgnoresTransformations)
            dot.setPos(px, py)
            self._scene.addItem(dot)
            self._overlay_items.append(dot)

    def _advance(self) -> None:
        if self._current + 1 >= self._frame_count:
            self._play_button.setChecked(False)
            return
        self.set_frame(self._current + 1)

    def _on_play_toggled(self, playing: bool) -> None:
        self._play_button.setText("Pause" if playing else "Play")
        if playing and self._capture is not None:
            self._timer.start()
        else:
            self._timer.stop()

    def _stop(self) -> None:
        self._timer.stop()
        self._play_button.setChecked(False)

    def _fit(self) -> None:
        if not self._pixmap_item.pixmap().isNull():
            self._view.fitInView(self._pixmap_item, Qt.AspectRatioMode.KeepAspectRatio)

    def resizeEvent(self, event) -> None:
        super().resizeEvent(event)
        self._fit()

    def showEvent(self, event) -> None:
        super().showEvent(event)
        self._fit()