silver_pose/v1/app.py

"""Assemble configuration, the fall pipeline worker, and the PyQt window.

The GUI never computes events. A background worker owns the video source and the
``FallPipeline`` and emits already-decided ``FrameAnalysis`` objects; the window
only renders them. Settings edits become a new immutable runtime snapshot at the
next start, so screenshots and logs stay traceable to the actual config version.
"""

import json
import os
import sys
from dataclasses import replace
from pathlib import Path
from typing import Dict, Optional

from PyQt5 import QtCore, QtWidgets

from v1.alerts import AlertDispatcher, EventArtifactWriter
from v1.config import AppConfig, EventConfig, load_config
from v1.pipeline import FallPipeline
from v1.pose import PoseAdapter
from v1.video_source import SourceMode, SourceStatus, VideoSource
from v1.view_model import SettingsDraft, build_monitor_view


class FrameWorker(QtCore.QThread):
    """Read one source, run the pipeline, and emit render-only analyses."""

    frame_ready = QtCore.pyqtSignal(object)

    def __init__(self, config: AppConfig, pose_adapter: PoseAdapter) -> None:
        super().__init__()
        self._config = config
        self._pose_adapter = pose_adapter
        self._stop = False

    def run(self) -> None:
        # OpenCV reads this env var when it opens the stream, so set it first.
        os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = self._config.ffmpeg_capture_options
        mode = SourceMode.STREAM if self._config.source_mode == "stream" else SourceMode.REPLAY
        source = VideoSource(self._config.source_url, mode=mode)
        pipeline = FallPipeline.from_config(self._config, self._pose_adapter)
        try:
            while not self._stop:
                packet = source.read()
                analysis = pipeline.process(packet)
                self.frame_ready.emit(analysis)
                if packet.status in (SourceStatus.CLOSED, SourceStatus.EOF):
                    break
                self.msleep(5)
        finally:
            source.close()

    def stop(self) -> None:
        self._stop = True


def _running_config(config: AppConfig, draft: SettingsDraft) -> AppConfig:
    values = draft.running_values
    event = EventConfig(
        keypoint_confidence_threshold=values["keypoint_confidence_threshold"],
        suspect_window_seconds=values["suspect_window_seconds"],
        confirm_window_seconds=values["confirm_window_seconds"],
        recovery_window_seconds=values["recovery_window_seconds"],
        cooldown_seconds=values["cooldown_seconds"],
    )
    return replace(
        config, event=event, confidence_threshold=values["model_confidence_threshold"]
    )


def _draft_from_config(config: AppConfig) -> SettingsDraft:
    return SettingsDraft(
        {
            "keypoint_confidence_threshold": config.event.keypoint_confidence_threshold,
            "suspect_window_seconds": config.event.suspect_window_seconds,
            "confirm_window_seconds": config.event.confirm_window_seconds,
            "recovery_window_seconds": config.event.recovery_window_seconds,
            "cooldown_seconds": config.event.cooldown_seconds,
            "model_confidence_threshold": config.confidence_threshold,
        }
    )


class ApplicationController:
    """Own the window, draft and worker lifecycle without doing event logic."""

    def __init__(
        self,
        config: AppConfig,
        pose_adapter: PoseAdapter,
        config_path: Optional[str] = None,
        camera: Optional[Dict] = None,
    ) -> None:
        from v1.gui import MainWindow

        self._config = config
        self._pose_adapter = pose_adapter
        self._draft = _draft_from_config(config)
        env_ready = bool(config.source_url)
        model_summary = "{0} · {1}…".format(config.model_path.name, config.model_sha256[:12])
        self.window = MainWindow(
            self._draft, env_ready, model_summary, config.source_id, config_path, camera
        )
        self.window.monitor.start_requested.connect(self.start)
        self.window.monitor.stop_requested.connect(self.stop)
        self._worker: Optional[FrameWorker] = None
        self._dispatcher: Optional[AlertDispatcher] = None

    def start(self) -> None:
        if self._worker is not None:
            return
        from v1.gui import QtAlertSink

        self._draft.start_monitoring()
        running = _running_config(self._config, self._draft)
        self._pose_adapter.set_confidence_threshold(running.confidence_threshold)
        writer = EventArtifactWriter(running.event_dir, running.source_id)
        self._dispatcher = AlertDispatcher(writer, QtAlertSink(self.window))
        worker = FrameWorker(running, self._pose_adapter)
        worker.frame_ready.connect(self._on_frame)
        worker.finished.connect(self._on_finished)
        self._worker = worker
        self.window.monitor.set_running(True)
        worker.start()

    def stop(self) -> None:
        if self._worker is not None:
            self._worker.stop()

    def _on_frame(self, analysis) -> None:
        view = build_monitor_view(
            analysis, self._draft.running_values["keypoint_confidence_threshold"]
        )
        self.window.monitor.render_view(analysis.packet.image, view)
        if self._dispatcher is not None:
            self._dispatcher.dispatch(analysis.events, analysis.packet.image, view)

    def _on_finished(self) -> None:
        self._worker = None
        self._dispatcher = None
        self.window.monitor.set_running(False)


def main(config_path: Optional[str] = None) -> int:
    path = Path(config_path or os.environ.get("SILVER_POSE_CONFIG", "v1/config.local.json"))
    config = load_config(path)
    pose_adapter = PoseAdapter(
        config.model_path, config.model_sha256, config.confidence_threshold
    )
    camera = _camera_fields(path)
    app = QtWidgets.QApplication(sys.argv)
    controller = ApplicationController(config, pose_adapter, str(path), camera)
    controller.window.show()
    return app.exec_()


def _camera_fields(path: Path) -> Dict:
    """Pre-fill the settings camera form from a structured local source, if any."""

    try:
        source = json.loads(Path(path).read_text(encoding="utf-8")).get("source", {})
    except (OSError, ValueError):
        return {}
    if "host" not in source:
        return {}
    fields = ("host", "port", "channel", "username", "password", "transport", "timeout_seconds", "low_latency")
    return {k: source.get(k) for k in fields if k in source}


if __name__ == "__main__":
    raise SystemExit(main())