diff --git a/dlclivegui/config.py b/dlclivegui/config.py index 11bcf8049..10c7a5a20 100644 --- a/dlclivegui/config.py +++ b/dlclivegui/config.py @@ -549,6 +549,21 @@ def writegear_overrides(self) -> WriteGearOptions | None: return {} +class WebStreamSettings(BaseModel): + """On-demand JPEG web view of the raw tiled preview. + + On/off and network exposure (localhost vs all interfaces) are local QSettings + preferences, not part of the shareable config, so loading someone else's + config can neither start the stream nor expose it on the network. + """ + + port: int = Field(default=8080, ge=1, le=65535) + fps: float = Field(default=5.0, gt=0.0, le=30.0) + max_width: int = Field(default=640, ge=16) + max_height: int = Field(default=480, ge=16) + jpeg_quality: int = Field(default=70, ge=1, le=100) + + class ApplicationSettings(BaseModel): # optional: add a semantic version for migrations version: int = 1 @@ -558,6 +573,7 @@ class ApplicationSettings(BaseModel): recording: RecordingSettings = Field(default_factory=RecordingSettings) bbox: BoundingBoxSettings = Field(default_factory=BoundingBoxSettings) visualization: VisualizationSettings = Field(default_factory=VisualizationSettings) + web_stream: WebStreamSettings = Field(default_factory=WebStreamSettings) @classmethod def from_dict(cls, data: dict[str, Any]) -> ApplicationSettings: @@ -574,6 +590,7 @@ def from_dict(cls, data: dict[str, Any]) -> ApplicationSettings: recording = RecordingSettings(**recording_data) bbox = BoundingBoxSettings(**bbox_data) visualization = VisualizationSettings(**visualization_data) + web_stream = WebStreamSettings(**data.get("web_stream", {})) return cls( camera=camera, @@ -582,6 +599,7 @@ def from_dict(cls, data: dict[str, Any]) -> ApplicationSettings: recording=recording, bbox=bbox, visualization=visualization, + web_stream=web_stream, ) def to_dict(self) -> dict[str, Any]: @@ -596,6 +614,7 @@ def to_dict(self) -> dict[str, Any]: "recording": self.recording.model_dump(), "bbox": self.bbox.model_dump(), "visualization": self.visualization.model_dump(), + "web_stream": self.web_stream.model_dump(), } @classmethod diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py index d049863e7..be19295b2 100644 --- a/dlclivegui/gui/main_window.py +++ b/dlclivegui/gui/main_window.py @@ -87,6 +87,7 @@ from .misc.drag_spinbox import ScrubSpinBox from .misc.eliding_label import ElidingPathLabel from .theme import LOGO, LOGO_ALPHA, AppStyle, apply_theme +from .web_stream import WebStreamController logger = logging.getLogger("DLCLiveGUI") @@ -152,6 +153,7 @@ def __init__(self, config: ApplicationSettings | None = None): self._active_inference_camera_id: str | None = None self._running_cams_ids: set[str] = set() self._current_frame: np.ndarray | None = None + self._shown_frame: np.ndarray | None = None # _current_frame + on-screen overlays self._raw_frame: np.ndarray | None = None self._last_pose: PoseResult | None = None self._dlc_active: bool = False @@ -442,6 +444,16 @@ def _build_menus(self) -> None: self.action_show_controls.toggled.connect(self.controls_dock.setVisible) self.controls_dock.visibilityChanged.connect(self.action_show_controls.setChecked) view_menu.addAction(self.action_show_controls) + ## Web stream (raw tiled frames, pulled on demand by the browser) + self._web_stream = WebStreamController( + self, + self._settings_store, + frame_source=lambda: self._current_frame, + overlay_frame_source=lambda: self._shown_frame, + settings=self._config.web_stream, + show_status=self.statusBar().showMessage, + ) + self._web_stream.add_to_menu(view_menu) ## -------------------- view_menu.addSeparator() ## Style actions @@ -942,6 +954,8 @@ def _apply_config(self, config: ApplicationSettings, *, restore_local_prefs: boo if hasattr(self, "bbox_color_combo"): color_ui.set_bbox_combo_from_bgr(self.bbox_color_combo, self._bbox_color) + self._web_stream.apply_settings(config.web_stream, restore_enabled=restore_local_prefs) + # Processor ## Use custom processor checkbox state if hasattr(self, "use_custom_proc_checkbox"): @@ -973,6 +987,7 @@ def _current_config(self, *, allow_empty_model_path=False) -> ApplicationSetting recording=self._recording_settings_from_ui(), bbox=self._bbox_settings_from_ui(), visualization=self._visualization_settings_from_ui(), + web_stream=self._web_stream.settings, ) def _parse_json(self, value: str) -> dict: @@ -1784,6 +1799,7 @@ def _on_multi_camera_stopped(self) -> None: self.stop_preview_button.setEnabled(False) self._current_frame = None + self._shown_frame = None self._multi_camera_frames.clear() self._multi_camera_display_ids.clear() self._running_cams_ids.clear() @@ -2133,6 +2149,7 @@ def _start_preview(self) -> None: self.preview_button.setEnabled(False) self.stop_preview_button.setEnabled(True) self._current_frame = None + self._shown_frame = None self._raw_frame = None self._last_pose = None self._multi_camera_frames.clear() @@ -2330,6 +2347,7 @@ def _update_display_from_pending(self) -> None: # Create tiled frame on demand (moved from camera thread for performance) tiled = create_tiled_frame(self._multi_camera_frames, labels=self._multi_camera_display_ids) if tiled is not None: + # Read by the web stream's server threads: replace, never draw on in place (overlays copy first). self._current_frame = tiled self._update_video_display(tiled) @@ -2610,6 +2628,8 @@ def _update_video_display(self, frame: np.ndarray) -> None: scale=self._dlc_tile_scale, ) + # Read by the web stream's server threads: replace, never draw on in place. + self._shown_frame = display_frame rgb = cv2.cvtColor(display_frame, cv2.COLOR_BGR2RGB) h, w, ch = rgb.shape @@ -2700,7 +2720,9 @@ def _set_combo_from_color(self, bgr: tuple[int, int, int]) -> None: # ------------------------------------------------------------------ # Qt overrides + # ------------------------------------------------------------------ def closeEvent(self, event: QCloseEvent) -> None: # pragma: no cover - GUI behaviour + self._web_stream.shutdown() if self.multi_camera_controller.is_running(): self.multi_camera_controller.stop(wait=True) diff --git a/dlclivegui/gui/web_stream.py b/dlclivegui/gui/web_stream.py new file mode 100644 index 000000000..d0373b442 --- /dev/null +++ b/dlclivegui/gui/web_stream.py @@ -0,0 +1,226 @@ +"""View-menu actions and lifecycle for the web stream. + +On/off, network exposure and overlays are local QSettings preferences, written only +by the user's menu toggles. Port, fps, size and quality come from the config JSON. +""" + +from __future__ import annotations + +import logging +import threading +from collections.abc import Callable + +import numpy as np +from PySide6.QtCore import QSignalBlocker, QUrl +from PySide6.QtGui import QAction, QDesktopServices, QGuiApplication +from PySide6.QtWidgets import QMenu, QMessageBox, QWidget + +from ..config import WebStreamSettings +from ..services.web_streamer import ( + LOOPBACK_HOST, + NETWORK_HOST, + SHUTDOWN_TIMEOUT, + WebStreamer, + local_url, + network_url, +) +from ..utils.settings_store import DLCLiveGUISettingsStore + +logger = logging.getLogger(__name__) + +_NETWORK_WARNING = ( + "Anyone on your network who can reach this computer will be able to watch the camera " + "stream, without a password.\n\n" + "The stream is view-only: it cannot control the app or read files.\n\n" + "Allow other devices?" +) +_COPY_LINK_TEXT = "Copy link for other devices" + + +def _set_checked_silently(action: QAction, checked: bool) -> None: + with QSignalBlocker(action): + action.setChecked(checked) + + +class WebStreamController: + """Owns the web stream's menu actions, preferences and server lifecycle. + + Menu: three switches (stream on/off, other devices allowed, overlays), then the URLs as actions. + """ + + def __init__( + self, + parent: QWidget, + settings_store: DLCLiveGUISettingsStore, + frame_source: Callable[[], np.ndarray | None], + overlay_frame_source: Callable[[], np.ndarray | None], + settings: WebStreamSettings, + show_status: Callable[[str, int], None], + ): + self._parent = parent + self._store = settings_store + self._raw_source = frame_source + self._overlay_source = overlay_frame_source + # Plain bool, not action.isChecked(): read from the server threads, where Qt objects are off-limits. + self._overlays = self._store.get_web_stream_overlays() + self._settings = settings.model_copy() + self._show_status = show_status + self._streamer: WebStreamer | None = None + self._pending_shutdown: threading.Thread | None = None + + self.action_enable = QAction("Stream to browser", parent, checkable=True) + self.action_enable.toggled.connect(self._on_enable_toggled) + self.action_network = QAction("Allow other devices (no password)", parent, checkable=True) + self.action_network.setChecked(self._store.get_web_stream_allow_network()) # before connect: no prompt + self.action_network.toggled.connect(self._on_network_toggled) + self.action_overlays = QAction("Include on-screen overlays (pose, box)", parent, checkable=True) + self.action_overlays.setChecked(self._overlays) + self.action_overlays.toggled.connect(self._on_overlays_toggled) + self.action_open = QAction(parent) + self.action_open.triggered.connect(lambda: QDesktopServices.openUrl(QUrl(self._local_url()))) + self.action_copy = QAction(_COPY_LINK_TEXT, parent) + self.action_copy.triggered.connect(self._copy_network_url) + self._refresh_menu() + + # ------------------------------------------------------------------ + # Public API (main window) + def add_to_menu(self, menu: QMenu) -> QMenu: + """Add a "Web stream" submenu holding the stream's actions to ``menu``.""" + submenu = menu.addMenu("Web stream") + submenu.addActions([self.action_enable, self.action_network, self.action_overlays]) + submenu.addSeparator() + submenu.addActions([self.action_open, self.action_copy]) + submenu.aboutToShow.connect(self._refresh_menu) # LAN address can change while running + return submenu + + @property + def settings(self) -> WebStreamSettings: + return self._settings.model_copy() + + @property + def streamer(self) -> WebStreamer | None: + return self._streamer + + def apply_settings(self, settings: WebStreamSettings, *, restore_enabled: bool = False) -> None: + """Adopt config parameters; restart a running stream only if they changed. + + ``restore_enabled`` starts the stream if the persisted preference says so (startup only). + """ + changed = settings != self._settings + self._settings = settings.model_copy() + if self._streamer is not None and changed: + self._restart() + elif restore_enabled and self._streamer is None and self._store.get_web_stream_enabled(): + self._start() + self._refresh_menu() + + def shutdown(self) -> None: + """Non-blocking stop for app exit; server threads are daemons and the OS releases the port.""" + self._stop() + + # ------------------------------------------------------------------ + # Menu toggles: the only writers of the persisted preferences + def _on_enable_toggled(self, enabled: bool) -> None: + if enabled: + self._store.set_web_stream_enabled(self._start()) + else: + self._store.set_web_stream_enabled(False) + if self._streamer is not None: + self._stop() + self._show_status("Web stream stopped", 3000) + self._refresh_menu() + + def _on_network_toggled(self, allow: bool) -> None: + if allow and not self._confirm_network(): + _set_checked_silently(self.action_network, False) + return + self._store.set_web_stream_allow_network(allow) + if self._streamer is not None: + self._restart() + self._refresh_menu() + + def _on_overlays_toggled(self, enabled: bool) -> None: + # No restart: the source is chosen per request. + self._overlays = enabled + self._store.set_web_stream_overlays(enabled) + + def _frame(self) -> np.ndarray | None: + """Frame source for the server threads.""" + return self._overlay_source() if self._overlays else self._raw_source() + + def _confirm_network(self) -> bool: + reply = QMessageBox.question( + self._parent, + "Allow other devices?", + _NETWORK_WARNING, + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.No, + ) + return reply == QMessageBox.StandardButton.Yes + + # ------------------------------------------------------------------ + # URLs + def _host(self) -> str: + return NETWORK_HOST if self.action_network.isChecked() else LOOPBACK_HOST + + def _port(self) -> int: + return self._streamer.port if self._streamer is not None else self._settings.port + + def _local_url(self) -> str: + return local_url(self._host(), self._port()) + + def _network_url(self) -> str | None: + return network_url(self._port()) if self._host() == NETWORK_HOST else None + + def _refresh_menu(self) -> None: + """Sync URL actions with state; they are only clickable while the stream runs.""" + running = self._streamer is not None + self.action_open.setText(f"Open {self._local_url()}") + self.action_open.setEnabled(running) + url = self._network_url() if running else None + self.action_copy.setText(f"{_COPY_LINK_TEXT}: {url}" if url else _COPY_LINK_TEXT) + self.action_copy.setEnabled(url is not None) + + def _copy_network_url(self) -> None: + url = self._network_url() + if url is None: + return + QGuiApplication.clipboard().setText(url) + self._show_status(f"Copied {url}", 3000) + + # ------------------------------------------------------------------ + # Lifecycle + def _start(self) -> bool: + """Start the stream and sync the menu check state. Returns whether it is running.""" + # A quick off/on toggle can race the previous server's background shutdown; wait for the port. + pending, self._pending_shutdown = self._pending_shutdown, None + if pending is not None: + pending.join(timeout=SHUTDOWN_TIMEOUT) + + host = self._host() + streamer = WebStreamer.from_settings(self._frame, self._settings, host=host) + try: + streamer.start() + except OSError as exc: + logger.warning("Failed to start web stream on %s:%s: %s", host, self._settings.port, exc) + self._show_status(f"Web stream failed to start on port {self._settings.port}: {exc}", 5000) + _set_checked_silently(self.action_enable, False) + return False + + self._streamer = streamer + _set_checked_silently(self.action_enable, True) + if host == NETWORK_HOST: + logger.warning("Web stream is reachable from the network without authentication") + self._show_status(f"Web stream at {self._local_url()} (links in View > Web stream)", 8000) + return True + + def _stop(self, wait: bool = False) -> None: + streamer, self._streamer = self._streamer, None + if streamer is not None: + self._pending_shutdown = streamer.stop(wait=wait) + _set_checked_silently(self.action_enable, False) + + def _restart(self) -> None: + # Blocking stop so the new server can rebind the port; user/config actions only, not a hot path. + self._stop(wait=True) + self._start() diff --git a/dlclivegui/services/web_streamer.py b/dlclivegui/services/web_streamer.py new file mode 100644 index 000000000..7ec05577b --- /dev/null +++ b/dlclivegui/services/web_streamer.py @@ -0,0 +1,251 @@ +"""Low-priority, on-demand JPEG web view of the GUI's raw tiled frame. + +Not encoding unless ``/frame.jpg`` is requested, and +encoding is capped at ``fps`` regardless of client count. +""" + +from __future__ import annotations + +import logging +import socket +import sys +import threading +import time +from collections.abc import Callable +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import cv2 +import numpy as np + +from dlclivegui.config import WebStreamSettings + +logger = logging.getLogger(__name__) + +# User-facing defaults (port, fps, size, quality) live in ``WebStreamSettings``; +# the values below are server internals. +LOOPBACK_HOST = "127.0.0.1" +NETWORK_HOST = "0.0.0.0" # all interfaces; no auth +SHUTDOWN_TIMEOUT = 2.0 # max wait for a server to release its port, wherever a caller waits +_POLL_INTERVAL = 0.1 # serve_forever shutdown latency +_CLIENT_TIMEOUT = 5.0 # drop idle or stalled connections +_WILDCARD_HOSTS = {"", "0.0.0.0"} + + +def lan_ip() -> str | None: + """Best-effort IPv4 address other machines can reach, or ``None`` (offline, no route). + + A UDP ``connect`` sends nothing; it only makes the OS pick the outgoing interface. + ``gethostbyname(gethostname())`` is avoided: it returns 127.0.1.1 on Debian/Ubuntu. + """ + try: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: + s.connect(("192.0.2.1", 9)) # TEST-NET-1 (RFC 5737), never routed to a real host + ip = s.getsockname()[0] + except OSError: + return None + return None if ip.startswith("127.") or ip in _WILDCARD_HOSTS else ip + + +def local_url(host: str, port: int) -> str: + """URL to open on this machine for a server bound to ``host`` (a wildcard address is not browsable).""" + return f"http://{'localhost' if host in _WILDCARD_HOSTS else host}:{port}/" + + +def network_url(port: int) -> str | None: + """URL for other machines to reach a server bound to all interfaces, or ``None`` if no LAN address.""" + ip = lan_ip() + return f"http://{ip}:{port}/" if ip else None + + +class _StreamServer(ThreadingHTTPServer): + """``ThreadingHTTPServer`` with bind fail when port is already taken, cross-platform. + + POSIX: keep ``SO_REUSEADDR``; it only permits rebinding over ``TIME_WAIT`` + and still refuses a second listener (macOS/BSD may allow a more specific + address next to a wildcard bind). + Windows: ``SO_REUSEADDR`` lets a second listener bind the same port, so + use a plain bind. ``SO_EXCLUSIVEADDRUSE`` is not used: it can block + rebinding while closed connections sit in ``TIME_WAIT``. + """ + + allow_reuse_address = sys.platform != "win32" + daemon_threads = True + + +_INDEX_HTML = """ +DLC Live stream + +waiting for frames + +""" + + +class WebStreamer: + """Serve the latest frame from ``frame_source`` as JPEG over HTTP.""" + + def __init__( + self, + frame_source: Callable[[], np.ndarray | None], + *, + host: str, + port: int, + fps: float, + max_size: tuple[int, int], + jpeg_quality: int, + ): + self._frame_source = frame_source + self._host = host + self._port = port + self._min_interval = 1.0 / fps if fps > 0 else 0.0 + self._max_size = max_size + self._jpeg_quality = int(jpeg_quality) + self._index = _INDEX_HTML.replace("__PERIOD_MS__", str(int(self._min_interval * 1000))).encode() + + self._server: _StreamServer | None = None + self._thread: threading.Thread | None = None + + self._cache_lock = threading.Lock() + self._cached_src: np.ndarray | None = None + self._cached_jpeg: bytes | None = None + self._last_encode = 0.0 + + @classmethod + def from_settings( + cls, frame_source: Callable[[], np.ndarray | None], settings: WebStreamSettings, *, host: str + ) -> WebStreamer: + return cls( + frame_source, + host=host, + port=settings.port, + fps=settings.fps, + max_size=(settings.max_width, settings.max_height), + jpeg_quality=settings.jpeg_quality, + ) + + @property + def port(self) -> int: + return self._server.server_address[1] if self._server else self._port + + @property + def url(self) -> str: + return local_url(self._host, self.port) + + def is_running(self) -> bool: + return self._server is not None + + def start(self) -> None: + """Bind and serve in a daemon thread. Raises ``OSError`` if the port is unavailable.""" + if self._server is not None: + return + server = _StreamServer((self._host, self._port), self._make_handler()) + self._server = server + self._thread = threading.Thread( + target=server.serve_forever, kwargs={"poll_interval": _POLL_INTERVAL}, name="WebStreamer", daemon=True + ) + self._thread.start() + logger.info("Web stream serving at %s", self.url) + + def stop(self, wait: bool = False) -> threading.Thread | None: + """Shut down on a short-lived daemon thread. + + ``wait=False`` returns immediately (GUI toggle). ``wait=True`` blocks + until the port is released, for an immediate restart on the same port. + Returns the shutdown thread (``None`` if not running) so a caller can + wait for the port later. + """ + server, self._server = self._server, None + self._thread = None + if server is None: + return None + + def _shutdown() -> None: + try: + server.shutdown() + finally: + server.server_close() + + shutdown_thread = threading.Thread(target=_shutdown, name="WebStreamerShutdown", daemon=True) + shutdown_thread.start() + with self._cache_lock: + self._cached_src = None + self._cached_jpeg = None + if wait: + shutdown_thread.join(timeout=SHUTDOWN_TIMEOUT) + return shutdown_thread + + def latest_jpeg(self) -> bytes | None: + """Return JPEG bytes for the current frame, encoding at most once per ``1/fps``.""" + frame = self._frame_source() + if frame is None: + return None + with self._cache_lock: + now = time.monotonic() + if self._cached_jpeg is not None and ( + frame is self._cached_src or now - self._last_encode < self._min_interval + ): + return self._cached_jpeg + jpeg = self._encode(frame) + if jpeg is not None: + self._cached_src = frame + self._cached_jpeg = jpeg + self._last_encode = now + return self._cached_jpeg + + def _encode(self, frame: np.ndarray) -> bytes | None: + h, w = frame.shape[:2] + max_w, max_h = self._max_size + scale = min(max_w / w, max_h / h, 1.0) + if scale < 1.0: + frame = cv2.resize(frame, (max(1, int(w * scale)), max(1, int(h * scale))), interpolation=cv2.INTER_AREA) + ok, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, self._jpeg_quality]) + return buf.tobytes() if ok else None + + def _make_handler(self) -> type[BaseHTTPRequestHandler]: + streamer = self + + class _Handler(BaseHTTPRequestHandler): + timeout = _CLIENT_TIMEOUT + + def do_GET(self) -> None: # noqa: N802 - http.server API + path = self.path.split("?", 1)[0] + try: + if path == "/": + self._send(200, "text/html; charset=utf-8", streamer._index) + elif path == "/frame.jpg": + try: + jpeg = streamer.latest_jpeg() + except Exception: + logger.exception("Web stream frame encoding failed") + jpeg = None + if jpeg is None: + self._send(503, "text/plain", b"no frame") + else: + self._send(200, "image/jpeg", jpeg) + else: + self._send(404, "text/plain", b"not found") + except (ConnectionError, TimeoutError): + pass # client went away or stalled past ``timeout`` + + def _send(self, status: int, content_type: str, body: bytes) -> None: + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args) -> None: # noqa: A002 - http.server API + pass + + return _Handler diff --git a/dlclivegui/utils/settings_store.py b/dlclivegui/utils/settings_store.py index c54265177..8103b0d00 100644 --- a/dlclivegui/utils/settings_store.py +++ b/dlclivegui/utils/settings_store.py @@ -38,6 +38,9 @@ class DLCLiveGUISettingsStore: KEY_USE_TIMESTAMP = "recording/use_timestamp" KEY_FAST_ENCODING = "recording/fast_encoding" KEY_REC_FILENAME = "recording/rec_filename" + KEY_WEB_STREAM_ENABLED = "web_stream/enabled" + KEY_WEB_STREAM_ALLOW_NETWORK = "web_stream/allow_network" + KEY_WEB_STREAM_OVERLAYS = "web_stream/overlays" def __init__(self, qsettings: QSettings | None = None): self._s = qsettings or QSettings("DeepLabCut", "DLCLiveGUI") @@ -110,6 +113,24 @@ def get_fast_encoding(self, default: bool = False) -> bool: def set_fast_encoding(self, enabled: bool) -> None: self._s.setValue(self.KEY_FAST_ENCODING, bool(enabled)) + def get_web_stream_enabled(self, default: bool = False) -> bool: + return self._get_bool(self.KEY_WEB_STREAM_ENABLED, default=default) + + def set_web_stream_enabled(self, enabled: bool) -> None: + self._s.setValue(self.KEY_WEB_STREAM_ENABLED, bool(enabled)) + + def get_web_stream_allow_network(self, default: bool = False) -> bool: + return self._get_bool(self.KEY_WEB_STREAM_ALLOW_NETWORK, default=default) + + def set_web_stream_allow_network(self, allow: bool) -> None: + self._s.setValue(self.KEY_WEB_STREAM_ALLOW_NETWORK, bool(allow)) + + def get_web_stream_overlays(self, default: bool = False) -> bool: + return self._get_bool(self.KEY_WEB_STREAM_OVERLAYS, default=default) + + def set_web_stream_overlays(self, enabled: bool) -> None: + self._s.setValue(self.KEY_WEB_STREAM_OVERLAYS, bool(enabled)) + def get_rec_filename(self) -> str: return self._get_optional_str(self.KEY_REC_FILENAME) or "" diff --git a/tests/gui/main_window/test_web_stream.py b/tests/gui/main_window/test_web_stream.py new file mode 100644 index 000000000..5a8b9d408 --- /dev/null +++ b/tests/gui/main_window/test_web_stream.py @@ -0,0 +1,303 @@ +# tests/gui/main_window/test_web_stream.py +from __future__ import annotations + +import numpy as np +import pytest +from PySide6.QtCore import QSettings, Qt +from PySide6.QtGui import QGuiApplication + +from dlclivegui.config import ApplicationSettings, WebStreamSettings +from dlclivegui.gui import web_stream as web_stream_mod +from dlclivegui.gui.main_window import DLCLiveMainWindow +from dlclivegui.services.web_streamer import LOOPBACK_HOST, NETWORK_HOST, WebStreamer +from dlclivegui.utils.settings_store import DLCLiveGUISettingsStore + + +@pytest.fixture(autouse=True) +def requested_hosts(monkeypatch): + """Bind loopback on port 0 whatever was requested, and record the requested host. + + Port 0: the OS assigns a free port atomically (no race with other processes). + Loopback: never bind all interfaces in tests (firewall prompts, exposure). + Same-port rebinding is covered by tests/services/test_web_streamer.py. + """ + hosts: list[str] = [] + + class _TestStreamer(WebStreamer): + def __init__(self, *args, **kwargs): + hosts.append(kwargs["host"]) + kwargs.update(host=LOOPBACK_HOST, port=0) + super().__init__(*args, **kwargs) + + monkeypatch.setattr(web_stream_mod, "WebStreamer", _TestStreamer) + return hosts + + +@pytest.fixture +def ctl(window): + """The window's web stream controller.""" + yield window._web_stream + window._web_stream._stop(wait=True) + + +def _fail_bind(monkeypatch): + def _fail(self): + raise OSError("address in use") + + monkeypatch.setattr(WebStreamer, "start", _fail) + + +def _answer_network_prompt(monkeypatch, ctl, answer: bool) -> list[bool]: + asked: list[bool] = [] + monkeypatch.setattr(ctl, "_confirm_network", lambda: asked.append(True) or answer) + return asked + + +@pytest.mark.gui +class TestWebStream: + def test_actions_grouped_under_view_web_stream_submenu(self, window, ctl): + view_menu = next(a.menu() for a in window.menuBar().actions() if a.text() == "&View") + submenu = next(a.menu() for a in view_menu.actions() if a.menu() and a.text() == "Web stream") + items = submenu.actions() + assert [a.isSeparator() for a in items] == [False, False, False, True, False, False] + assert [a for a in items if not a.isSeparator()] == [ + ctl.action_enable, + ctl.action_network, + ctl.action_overlays, + ctl.action_open, + ctl.action_copy, + ] + + def test_off_and_local_by_default(self, ctl): + assert not ctl.action_enable.isChecked() + assert not ctl.action_network.isChecked() + assert ctl.streamer is None + assert ctl._store.get_web_stream_enabled() is False + assert ctl._store.get_web_stream_allow_network() is False + + def test_user_toggle_starts_stops_and_persists(self, ctl, requested_hosts): + ctl.action_enable.setChecked(True) + assert ctl.streamer is not None and ctl.streamer.is_running() + assert ctl.streamer._port == 0 # requested_hosts override is in effect + assert requested_hosts == [LOOPBACK_HOST] + assert ctl._store.get_web_stream_enabled() is True + + ctl.action_enable.setChecked(False) + assert ctl.streamer is None + assert ctl._store.get_web_stream_enabled() is False + + def test_quick_retoggle_restarts(self, ctl): + for _ in range(3): + ctl.action_enable.setChecked(True) + assert ctl.streamer is not None and ctl.streamer.is_running() + ctl.action_enable.setChecked(False) + assert ctl.streamer is None + + def test_open_item_shows_url_and_is_clickable_only_while_running(self, ctl, monkeypatch): + opened = [] + monkeypatch.setattr(web_stream_mod.QDesktopServices, "openUrl", lambda url: opened.append(url.toString())) + + ctl.apply_settings(WebStreamSettings(port=9123)) + assert ctl.action_open.text() == "Open http://127.0.0.1:9123/" # configured port while stopped + assert not ctl.action_open.isEnabled() + + ctl.action_enable.setChecked(True) + url = f"http://127.0.0.1:{ctl.streamer.port}/" # actual bound port + assert ctl.action_open.text() == f"Open {url}" + assert ctl.action_open.isEnabled() + ctl.action_open.trigger() + assert opened == [url] + + ctl.action_enable.setChecked(False) + assert not ctl.action_open.isEnabled() + + def test_copy_link_only_for_other_devices_while_running(self, ctl, monkeypatch): + monkeypatch.setattr("dlclivegui.services.web_streamer.lan_ip", lambda: "192.168.1.20") + ctl.action_enable.setChecked(True) + assert not ctl.action_copy.isEnabled() # localhost only + assert ctl.action_copy.text() == "Copy link for other devices" + + _answer_network_prompt(monkeypatch, ctl, True) + ctl.action_network.setChecked(True) + url = f"http://192.168.1.20:{ctl.streamer.port}/" + assert ctl.action_copy.isEnabled() + assert ctl.action_copy.text() == f"Copy link for other devices: {url}" + assert ctl.action_open.text() == f"Open http://localhost:{ctl.streamer.port}/" + + ctl.action_copy.trigger() + assert QGuiApplication.clipboard().text() == url + + ctl.action_enable.setChecked(False) + assert not ctl.action_copy.isEnabled() + + def test_copy_link_disabled_without_lan_address(self, ctl, monkeypatch): + monkeypatch.setattr("dlclivegui.services.web_streamer.lan_ip", lambda: None) + _answer_network_prompt(monkeypatch, ctl, True) + ctl.action_network.setChecked(True) + ctl.action_enable.setChecked(True) + assert not ctl.action_copy.isEnabled() + assert ctl.action_copy.text() == "Copy link for other devices" + + def test_param_change_restarts_only_when_running_and_changed(self, ctl): + ctl.action_enable.setChecked(True) + first = ctl.streamer + + ctl.apply_settings(ctl.settings) + assert ctl.streamer is first # unchanged params: no restart + + ctl.apply_settings(ctl.settings.model_copy(update={"fps": 1.0})) + assert ctl.streamer is not first and ctl.streamer.is_running() + assert not first.is_running() + assert ctl.action_enable.isChecked() + + def test_config_load_never_starts_stream(self, window, ctl): + ctl._store.set_web_stream_enabled(True) + window._apply_config(window._config) # File > Load path: restore_local_prefs=False + assert ctl.streamer is None + + def test_config_load_cannot_expose_stream_on_network(self, window, ctl, requested_hosts): + ctl.action_enable.setChecked(True) + data = window._config.to_dict() + data["web_stream"] = {"host": NETWORK_HOST, "enabled": True, "port": 9000} + window._apply_config(ApplicationSettings.from_dict(data)) + + assert ctl.streamer is not None # restarted for the new port... + assert requested_hosts == [LOOPBACK_HOST, LOOPBACK_HOST] # ...but never on the network + assert ctl._store.get_web_stream_allow_network() is False + + def test_startup_restores_enabled_preference(self, ctl): + ctl._store.set_web_stream_enabled(True) + ctl.apply_settings(ctl.settings, restore_enabled=True) + assert ctl.streamer is not None + assert ctl.action_enable.isChecked() + + def test_current_config_carries_params_only(self, window, ctl): + ctl.apply_settings(ctl.settings.model_copy(update={"jpeg_quality": 40})) + cfg = window._current_config(allow_empty_model_path=True).web_stream + assert cfg.jpeg_quality == 40 + assert set(cfg.model_dump()) == {"port", "fps", "max_width", "max_height", "jpeg_quality"} + + def test_user_toggle_bind_failure_unchecks_and_persists_off(self, ctl, monkeypatch): + _fail_bind(monkeypatch) + ctl.action_enable.setChecked(True) + + assert not ctl.action_enable.isChecked() + assert ctl.streamer is None + assert ctl._store.get_web_stream_enabled() is False + + def test_startup_bind_failure_keeps_preference(self, ctl, monkeypatch): + # e.g. a second instance: the port is taken now, the user's intent should survive. + ctl._store.set_web_stream_enabled(True) + _fail_bind(monkeypatch) + ctl.apply_settings(ctl.settings, restore_enabled=True) + + assert not ctl.action_enable.isChecked() + assert ctl._store.get_web_stream_enabled() is True + + +@pytest.mark.gui +class TestWebStreamNetwork: + def test_declined_prompt_keeps_localhost(self, ctl, monkeypatch, requested_hosts): + ctl.action_enable.setChecked(True) + asked = _answer_network_prompt(monkeypatch, ctl, False) + ctl.action_network.setChecked(True) + + assert asked == [True] + assert not ctl.action_network.isChecked() + assert ctl._store.get_web_stream_allow_network() is False + assert requested_hosts == [LOOPBACK_HOST] # no restart + + def test_accepted_prompt_restarts_on_network_and_persists(self, ctl, monkeypatch, requested_hosts): + ctl.action_enable.setChecked(True) + _answer_network_prompt(monkeypatch, ctl, True) + ctl.action_network.setChecked(True) + + assert ctl._store.get_web_stream_allow_network() is True + assert ctl.streamer is not None + assert requested_hosts == [LOOPBACK_HOST, NETWORK_HOST] + + def test_disabling_network_needs_no_prompt_and_restarts_local(self, ctl, monkeypatch, requested_hosts): + _answer_network_prompt(monkeypatch, ctl, True) + ctl.action_network.setChecked(True) + ctl.action_enable.setChecked(True) + asked = _answer_network_prompt(monkeypatch, ctl, False) + ctl.action_network.setChecked(False) + + assert asked == [] + assert ctl._store.get_web_stream_allow_network() is False + assert requested_hosts == [NETWORK_HOST, LOOPBACK_HOST] + + def test_toggle_while_stopped_does_not_start(self, ctl, monkeypatch): + _answer_network_prompt(monkeypatch, ctl, True) + ctl.action_network.setChecked(True) + assert ctl.streamer is None + + def test_startup_restores_network_preference_without_prompt(self, qtbot, app_config_two_cams, requested_hosts): + store = DLCLiveGUISettingsStore(QSettings("DeepLabCut", "DLCLiveGUI")) + store.set_web_stream_enabled(True) + store.set_web_stream_allow_network(True) + + def _no_prompt(self): + raise AssertionError("network prompt shown at startup") + + w = DLCLiveMainWindow(config=app_config_two_cams) + qtbot.addWidget(w) + w.setAttribute(Qt.WA_DontShowOnScreen, True) + try: + ctl = w._web_stream + ctl._confirm_network = _no_prompt.__get__(ctl) + assert ctl.action_network.isChecked() + assert ctl.streamer is not None + assert requested_hosts == [NETWORK_HOST] + finally: + w._web_stream._stop(wait=True) + w.close() + + +@pytest.mark.gui +def test_close_event_stops_stream_without_blocking(window, monkeypatch): + ctl = window._web_stream + ctl.action_enable.setChecked(True) + streamer = ctl.streamer + calls = [] + real_stop = streamer.stop + monkeypatch.setattr(streamer, "stop", lambda wait=False: calls.append(wait) or real_stop(wait=wait)) + + window.close() + + assert calls == [False] + assert ctl.streamer is None + + +@pytest.mark.gui +class TestWebStreamOverlays: + def test_raw_frames_by_default(self, window, ctl): + assert not ctl.action_overlays.isChecked() + window._current_frame, window._shown_frame = object(), object() + assert ctl._frame() is window._current_frame + + def test_toggle_switches_source_without_restart_and_persists(self, window, ctl): + ctl.action_enable.setChecked(True) + streamer = ctl.streamer + window._current_frame, window._shown_frame = object(), object() + + ctl.action_overlays.setChecked(True) + assert ctl._frame() is window._shown_frame + assert ctl.streamer is streamer # chosen per request: no restart + assert ctl._store.get_web_stream_overlays() is True + + ctl.action_overlays.setChecked(False) + assert ctl._frame() is window._current_frame + assert ctl._store.get_web_stream_overlays() is False + + def test_shown_frame_is_what_the_screen_shows(self, window): + frame = np.zeros((120, 160, 3), dtype=np.uint8) + window._bbox_enabled = False + window._update_video_display(frame) + assert window._shown_frame is frame # no overlay: same array, no copy + + window._bbox_enabled = True + window._bbox_x0, window._bbox_y0, window._bbox_x1, window._bbox_y1 = 10, 10, 60, 60 + window._update_video_display(frame) + assert window._shown_frame is not frame # overlays are drawn on a copy... + assert window._shown_frame.any() and not frame.any() # ...which carries the box; the source stays raw diff --git a/tests/services/test_web_streamer.py b/tests/services/test_web_streamer.py new file mode 100644 index 000000000..83435da37 --- /dev/null +++ b/tests/services/test_web_streamer.py @@ -0,0 +1,260 @@ +from __future__ import annotations + +import socket +import threading +import urllib.error +import urllib.request +from types import SimpleNamespace + +import cv2 +import numpy as np +import pytest + +from dlclivegui.config import WebStreamSettings +from dlclivegui.services import web_streamer as ws +from dlclivegui.services.web_streamer import LOOPBACK_HOST, NETWORK_HOST, WebStreamer + +# Upper bound for joins that only matters if the code under test is broken +# (passing runs return as soon as the event happens); not a timing assertion. +_FAILURE_CAP = 30.0 + + +def _make(frame_box, *, port: int = 0, host: str = LOOPBACK_HOST, **overrides) -> WebStreamer: + """Streamer with config defaults; port 0 lets the OS pick a free port atomically.""" + d = WebStreamSettings() + params = {"fps": d.fps, "max_size": (d.max_width, d.max_height), "jpeg_quality": d.jpeg_quality, **overrides} + return WebStreamer(lambda: frame_box["frame"], host=host, port=port, **params) + + +def _get(url: str) -> tuple[int, bytes, str]: + try: + with urllib.request.urlopen(url, timeout=5) as resp: + return resp.status, resp.read(), resp.headers.get("Content-Type", "") + except urllib.error.HTTPError as exc: + return exc.code, exc.read(), exc.headers.get("Content-Type", "") + + +@pytest.fixture +def frame_box(): + return {"frame": np.full((960, 1280, 3), 127, dtype=np.uint8)} + + +@pytest.fixture +def streamer(frame_box): + s = _make(frame_box, port=0, fps=5.0, max_size=(640, 480)) + s.start() + yield s + s.stop() + + +@pytest.mark.unit +def test_index_serves_html(streamer): + status, body, ctype = _get(streamer.url) + assert status == 200 + assert ctype.startswith("text/html") + assert b"/frame.jpg" in body + + +@pytest.mark.unit +def test_frame_is_downscaled_jpeg(streamer): + status, body, ctype = _get(streamer.url + "frame.jpg") + assert status == 200 + assert ctype == "image/jpeg" + assert body[:2] == b"\xff\xd8" + img = cv2.imdecode(np.frombuffer(body, np.uint8), cv2.IMREAD_COLOR) + assert img.shape[:2] == (480, 640) + + +@pytest.mark.unit +def test_no_frame_returns_503(frame_box, streamer): + frame_box["frame"] = None + status, _, _ = _get(streamer.url + "frame.jpg") + assert status == 503 + + +@pytest.mark.unit +def test_unknown_path_returns_404(streamer): + status, _, _ = _get(streamer.url + "nope") + assert status == 404 + + +@pytest.mark.unit +def test_encoding_is_cached_and_rate_limited(monkeypatch, frame_box): + calls = {"n": 0} + real_imencode = cv2.imencode + + def counting_imencode(*args, **kwargs): + calls["n"] += 1 + return real_imencode(*args, **kwargs) + + monkeypatch.setattr(ws.cv2, "imencode", counting_imencode) + clock = {"now": 100.0} + monkeypatch.setattr(ws, "time", SimpleNamespace(monotonic=lambda: clock["now"])) + s = _make(frame_box, port=0, fps=5.0) # min interval 0.2 s + + s.latest_jpeg() + s.latest_jpeg() # same source object -> cached + assert calls["n"] == 1 + + frame_box["frame"] = frame_box["frame"].copy() + clock["now"] += 0.1 + s.latest_jpeg() # new object but within 1/fps -> cached + assert calls["n"] == 1 + + clock["now"] += 0.2 + s.latest_jpeg() # new object, interval elapsed -> re-encode + assert calls["n"] == 2 + + clock["now"] += 10.0 + s.latest_jpeg() # interval elapsed but same object -> cached + assert calls["n"] == 2 + + +@pytest.mark.unit +def test_stop_returns_before_shutdown_completes(frame_box): + s = _make(frame_box, port=0) + s.start() + server = s._server + release = threading.Event() + real_shutdown = server.shutdown + + def blocking_shutdown(): + release.wait() + real_shutdown() + + server.shutdown = blocking_shutdown + + t = s.stop() # must not wait for shutdown, which cannot finish until released + try: + assert t is not None and t.is_alive() + assert not s.is_running() + finally: + release.set() + t.join(timeout=_FAILURE_CAP) + assert not t.is_alive() + + +@pytest.mark.unit +def test_shutdown_frees_port(frame_box): + s = _make(frame_box, port=0) + s.start() + server = s._server + port = server.server_address[1] + + t = s.stop() + t.join(timeout=_FAILURE_CAP) + assert not t.is_alive() + assert server.socket.fileno() == -1 # listening socket closed + + again = _make(frame_box, port=port) + again.start() # raises OSError if the port were still held + again.stop(wait=True) + + +@pytest.mark.unit +def test_second_instance_on_same_port_is_refused(frame_box): + # Windows SO_REUSEADDR would let this bind succeed silently; POSIX refuses either way. + first = _make(frame_box, port=0) + first.start() + try: + second = _make(frame_box, port=first._server.server_address[1]) + with pytest.raises(OSError): + second.start() + assert not second.is_running() + finally: + first.stop(wait=True) + + +@pytest.mark.unit +def test_restart_on_same_port_after_serving_requests(frame_box): + # Served requests leave server-side TIME_WAIT connections; rebinding must still work on every OS. + s = _make(frame_box, port=0) + s.start() + port = s._server.server_address[1] + for _ in range(3): + assert _get(s.url + "frame.jpg")[0] == 200 + s.stop(wait=True) + + again = _make(frame_box, port=port) + again.start() + try: + assert _get(again.url + "frame.jpg")[0] == 200 + finally: + again.stop(wait=True) + + +@pytest.mark.unit +def test_stop_returns_shutdown_thread(frame_box): + s = _make(frame_box, port=0) + assert s.stop() is None # not running + s.start() + t = s.stop() + assert t is not None + t.join(timeout=_FAILURE_CAP) + assert not t.is_alive() + + +@pytest.mark.unit +def test_stalled_client_is_dropped(monkeypatch, frame_box): + monkeypatch.setattr(ws, "_CLIENT_TIMEOUT", 0.2) + s = _make(frame_box, port=0) + s.start() + try: + with socket.create_connection(("127.0.0.1", s._server.server_address[1]), timeout=3.0) as sock: + # Send nothing: the server must close the connection after its timeout. + try: + assert sock.recv(1024) == b"" + except ConnectionResetError: + pass # some platforms reset instead of closing cleanly + finally: + s.stop(wait=True) + + +@pytest.mark.unit +def test_from_settings_maps_every_field(): + settings = WebStreamSettings(port=9001, fps=2.0, max_width=320, max_height=200, jpeg_quality=55) + s = WebStreamer.from_settings(lambda: None, settings, host=LOOPBACK_HOST) + assert (s.port, s._min_interval, s._max_size, s._jpeg_quality) == (9001, 0.5, (320, 200), 55) + + +@pytest.mark.unit +def test_local_url(frame_box): + assert ws.local_url(LOOPBACK_HOST, 8123) == "http://127.0.0.1:8123/" + # A wildcard address is not browsable: show localhost. + assert ws.local_url(NETWORK_HOST, 8123) == "http://localhost:8123/" + assert _make(frame_box, port=8123, host=NETWORK_HOST).url == "http://localhost:8123/" # no bind + + +@pytest.mark.unit +def test_network_url(monkeypatch): + monkeypatch.setattr(ws, "lan_ip", lambda: "192.168.1.20") + assert ws.network_url(8123) == "http://192.168.1.20:8123/" + monkeypatch.setattr(ws, "lan_ip", lambda: None) + assert ws.network_url(8123) is None + + +@pytest.mark.unit +def test_lan_ip_is_none_without_route(monkeypatch): + class _NoRoute: + def __init__(self, *a): + pass + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def connect(self, addr): + raise OSError("Network is unreachable") + + monkeypatch.setattr( + ws, "socket", SimpleNamespace(socket=_NoRoute, AF_INET=socket.AF_INET, SOCK_DGRAM=socket.SOCK_DGRAM) + ) + assert ws.lan_ip() is None + + +@pytest.mark.unit +def test_lan_ip_is_never_loopback(): + ip = ws.lan_ip() # environment-dependent value, invariant only + assert ip is None or not ip.startswith("127.") diff --git a/tests/test_config.py b/tests/test_config.py index 6e1cc76cb..081a2136e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -7,6 +7,7 @@ CameraTriggerSettings, MultiCameraSettings, RecordingSettings, + WebStreamSettings, ) from dlclivegui.services.video_recorder import build_writegear_options @@ -198,3 +199,41 @@ def test_build_writegear_options_invalid_fps_uses_default(): ) assert opts["-input_framerate"] == DEFAULT_RECORDING_FPS + + +@pytest.mark.unit +def test_web_stream_defaults_when_missing_from_config(): + settings = ApplicationSettings.from_dict({}) + assert settings.web_stream == WebStreamSettings() + + +@pytest.mark.unit +def test_web_stream_roundtrips_through_json(tmp_path): + ws = WebStreamSettings(port=9000, fps=2.0, max_width=320, max_height=240, jpeg_quality=50) + path = tmp_path / "cfg.json" + ApplicationSettings(web_stream=ws).save(path) + assert ApplicationSettings.load(path).web_stream == ws + + +@pytest.mark.unit +def test_web_stream_exposure_is_not_in_shareable_config(): + # On/off and network exposure live in local QSettings; a config file must never carry or apply them. + assert set(ApplicationSettings().to_dict()["web_stream"]) == { + "port", + "fps", + "max_width", + "max_height", + "jpeg_quality", + } + loaded = ApplicationSettings.from_dict({"web_stream": {"enabled": True, "host": "0.0.0.0", "port": 9000}}) + assert loaded.web_stream == WebStreamSettings(port=9000) + + +@pytest.mark.unit +@pytest.mark.parametrize( + "field, value", + [("port", 0), ("port", 70000), ("fps", 0.0), ("fps", 60.0), ("jpeg_quality", 0), ("max_width", 0)], +) +def test_web_stream_rejects_invalid_values(field, value): + with pytest.raises(ValueError): + WebStreamSettings(**{field: value}) diff --git a/tests/utils/test_settings_store.py b/tests/utils/test_settings_store.py index 702382429..66737697e 100644 --- a/tests/utils/test_settings_store.py +++ b/tests/utils/test_settings_store.py @@ -113,6 +113,18 @@ def test_qt_settings_store_fast_encoding_roundtrip(): assert settstore.get_fast_encoding(default=True) is False +@pytest.mark.parametrize("pref", ["web_stream_enabled", "web_stream_allow_network", "web_stream_overlays"]) +def test_qt_settings_store_web_stream_bool_roundtrip(pref): + settstore = store.DLCLiveGUISettingsStore(qsettings=InMemoryQSettings()) + get, set_ = getattr(settstore, f"get_{pref}"), getattr(settstore, f"set_{pref}") + + assert get() is False # off unless the user turned it on + set_(True) + assert get(default=False) is True + set_(False) + assert get(default=True) is False + + # ----------------------------- # ModelPathStore helpers # -----------------------------