feat: optional camera view outside the workarea

rayforge/camera/controller.py

@@ -10,6 +10,18 @@
 from ..image.util.srgb import resize_linear_nd
 from ..shared.util.glib import idle_add
 from .models.camera import Camera
+from .outside_view import (
+    LensModel,
+    OutsideView,
+    expand_area,
+    frame_coverage_extent,
+    lens_coverage_extent,
+    lens_remap_maps,
+    native_output_size,
+    remap_transparent,
+    to_opaque_bgra,
+    warp_transparent,
+)
 from .source import (
     create_camera_source,
     list_local_device_ids,
@@ -70,6 +82,10 @@ def __init__(self, config: Camera):
         self._active_source: CameraSource | None = None
         self._last_source_key = self._source_key()
         self._disposed: bool = False
+        # Cached outside view coverage and remap tables, which only
+        # change with calibration, alignment, and display geometry.
+        # Accessed from the GLib main loop only, like its callers.
+        self._outside_cache: dict[str, tuple] = {}
 
         # Stream lifecycle: _lifecycle_lock serializes start/stop so that
         # at most one capture thread can ever own the device at a time
@@ -103,7 +119,7 @@ def __init__(self, config: Camera):
         self.resolutions_probed = Signal()
 
         self.config.changed.connect(self._on_config_changed)
-        self.config.settings_changed.connect(self._on_config_changed)
+        self.config.settings_changed.connect(self._on_capture_settings_changed)
 
     def _source_key(self) -> tuple:
         """A tuple identifying which physical source is being captured.
@@ -119,9 +135,8 @@ def _on_config_changed(self, sender):
         """Reacts to changes in the data model."""
         if self._disposed:
             return
-        self._settings_dirty = True
-        with self._frame_lock:
-            self._accumulator = None  # Reset smoothing if settings change
+        if not self.config.outside_view_enabled:
+            self._outside_cache.clear()
         new_source_key = self._source_key()
         source_changed = new_source_key != self._last_source_key
         self._last_source_key = new_source_key
@@ -136,6 +151,14 @@ def _on_config_changed(self, sender):
             # Also stop if it's disabled, regardless of subscribers
             self._stop_capture_stream()
 
+    def _on_capture_settings_changed(self, sender):
+        """Reacts to changes that affect capturing or frame processing."""
+        if self._disposed:
+            return
+        self._settings_dirty = True
+        with self._frame_lock:
+            self._accumulator = None  # Reset smoothing if settings change
+
     @staticmethod
     def list_available_devices() -> list[str]:
         """
@@ -346,6 +369,66 @@ def get_work_surface_image(
             logger.warning("No image data available.")
             return None
 
+        return self._work_surface_from_image(image, output_size, physical_area)
+
+    def get_work_surface_images(
+        self,
+        output_size: tuple[int, int],
+        physical_area: tuple[Pos, Pos],
+        outside_margin_mm: float,
+        max_dimension: int,
+    ) -> tuple[np.ndarray | None, OutsideView | None]:
+        """
+        Get the workspace image and the surrounding outside view from a
+        single frame.
+
+        The workspace image is identical to what
+        :meth:`get_work_surface_image` returns for the same arguments. The
+        outside view covers ``physical_area`` expanded by the effective
+        margin on all four sides, which is ``outside_margin_mm`` clamped to
+        the area the camera actually sees. It is rendered at the camera's
+        pixel density, capped to ``max_dimension``, independent of
+        ``output_size``.
+
+        With lens correction, the outside view is sampled from the raw
+        frame captured together with the processed one, since lens
+        correction crops the edges of the processed frame.
+
+        Returns:
+            A tuple (workspace image, outside view). The outside view is
+            None without alignment, without coverage beyond the workspace,
+            or on failure; the workspace image is None on failure.
+        """
+        with self._frame_lock:
+            image = self._image_data
+            raw = self._raw_image_data
+
+        if image is None:
+            logger.warning("No image data available.")
+            return None, None
+
+        workspace = self._work_surface_from_image(
+            image, output_size, physical_area
+        )
+        try:
+            outside = self._outside_view_from_image(
+                image,
+                raw,
+                physical_area,
+                outside_margin_mm,
+                max_dimension,
+            )
+        except (cv2.error, ValueError, np.linalg.LinAlgError) as e:
+            logger.error(f"Failed to render outside camera view: {e}")
+            outside = None
+        return workspace, outside
+
+    def _work_surface_from_image(
+        self,
+        image: np.ndarray,
+        output_size: tuple[int, int],
+        physical_area: tuple[Pos, Pos],
+    ) -> np.ndarray | None:
         if self.config.image_to_world is not None:
             return self._transform_with_homography(
                 image, output_size, physical_area
@@ -418,6 +501,91 @@ def _transform_with_homography(
             logger.error(f"Failed to apply perspective warp: {e}")
             return None
 
+    def _outside_view_from_image(
+        self,
+        image: np.ndarray,
+        raw: np.ndarray | None,
+        physical_area: tuple[Pos, Pos],
+        margin_mm: float,
+        max_dimension: int,
+    ) -> OutsideView | None:
+        if self.config.image_to_world is None or margin_mm <= 0:
+            return None
+        H = self._compute_homography(image.shape[0])
+        lens = self._lens_model(image)
+        if lens is not None and raw is not None:
+            return self._outside_view_from_raw(
+                raw,
+                H,
+                lens,
+                physical_area,
+                margin_mm,
+                max_dimension,
+            )
+
+        margin = min(
+            margin_mm,
+            frame_coverage_extent(H, image.shape, physical_area),
+        )
+        if margin <= 0:
+            return None
+        area = expand_area(physical_area, margin)
+        size = native_output_size(H, physical_area, margin, max_dimension)
+        warped = warp_transparent(to_opaque_bgra(image), H, size, area)
+        return OutsideView(image=warped, area=area, margin_mm=margin)
+
+    def _outside_view_from_raw(
+        self,
+        raw: np.ndarray,
+        H: np.ndarray,
+        lens: LensModel,
+        physical_area: tuple[Pos, Pos],
+        margin_mm: float,
+        max_dimension: int,
+    ) -> OutsideView | None:
+        """
+        Render the outside view from the raw frame, so that content lens
+        correction pushed out of the corrected frame is still shown.
+        """
+        base_key = (H.tobytes(), lens.key(), raw.shape[:2], physical_area)
+
+        extent_key = (base_key, margin_mm)
+        cached = self._outside_cache.get("extent")
+        if cached is not None and cached[0] == extent_key:
+            margin = cached[1]
+        else:
+            margin = lens_coverage_extent(
+                H, lens, raw.shape, physical_area, margin_mm
+            )
+            self._outside_cache["extent"] = (extent_key, margin)
+        if margin <= 0:
+            return None
+
+        area = expand_area(physical_area, margin)
+        size = native_output_size(H, physical_area, margin, max_dimension)
+        maps_key = (base_key, area, size)
+        cached = self._outside_cache.get("maps")
+        if cached is not None and cached[0] == maps_key:
+            maps = cached[1]
+        else:
+            maps = lens_remap_maps(H, lens, size, area)
+            self._outside_cache["maps"] = (maps_key, maps)
+
+        remapped = remap_transparent(to_opaque_bgra(raw), maps)
+        return OutsideView(image=remapped, area=area, margin_mm=margin)
+
+    def _lens_model(self, image: np.ndarray) -> LensModel | None:
+        """The lens correction applied to frames of this image's size."""
+        cam_mat, dist = self._get_effective_calibration(
+            image.shape[0], image.shape[1]
+        )
+        if cam_mat is None or dist is None:
+            return None
+        return LensModel(
+            camera_matrix=np.asarray(cam_mat, dtype=np.float64),
+            distortion=np.asarray(dist, dtype=np.float64),
+        )
+
     def _apply_settings(self, source) -> None:
         """Applies the current settings to the VideoCapture object."""
         try:
@@ -545,6 +713,7 @@ def _handle_frame_failure(self):
         self._consecutive_failures += 1
         with self._frame_lock:
             self._image_data = None
+            self._raw_image_data = None
         return self._consecutive_failures >= self.MAX_CONSECUTIVE_FAILURES
 
     def _log_frame_failure(self, source) -> None:
@@ -754,9 +923,10 @@ def _stop_locked(self):
         self._capture_thread = None
 
     def _clear_image_data(self) -> None:
-        """Drops the processed frame under the frame lock."""
+        """Drops the captured frames under the frame lock."""
         with self._frame_lock:
             self._image_data = None
+            self._raw_image_data = None
 
     def capture_image(self, *, apply_settings: bool = True):
         """

rayforge/camera/models/camera.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import math
 import uuid
 from collections.abc import Sequence
 from datetime import datetime, timezone
@@ -15,6 +16,8 @@
 Pos = tuple[float, float]
 PointList = Sequence[Pos]
 
+OUTSIDE_VIEW_MAX_MARGIN_MM = 100.0
+
 
 def _as_utc(dt: datetime | None) -> datetime | None:
     """Interpret a possibly naive datetime as UTC.
@@ -60,6 +63,12 @@ def __init__(
 
         self._prefer_yuyv: bool = False
 
+        # Optional camera view around the workspace, drawn on the work
+        # surface only outside the workspace rectangle.
+        self._outside_view_enabled: bool = False
+        self._outside_view_margin_mm: float = 30.0
+        self._outside_view_transparency: float = 0.65
+
         self._resolution: tuple[int, int] | None = None
 
         # Lens calibration parameters
@@ -319,6 +328,79 @@ def prefer_yuyv(self, value: bool):
         self.changed.send(self)
         self.settings_changed.send(self)
 
+    @property
+    def outside_view_enabled(self) -> bool:
+        """Whether to show the camera image around the workspace."""
+        return self._outside_view_enabled
+
+    @outside_view_enabled.setter
+    def outside_view_enabled(self, value: bool):
+        if not isinstance(value, bool):
+            raise TypeError("outside_view_enabled must be a boolean.")
+        if self._outside_view_enabled == value:
+            return
+        logger.debug(
+            f"Camera outside_view_enabled changed from "
+            f"{self._outside_view_enabled} to {value}"
+        )
+        self._outside_view_enabled = value
+        self.changed.send(self)
+
+    @property
+    def outside_view_margin_mm(self) -> float:
+        """Margin shown on each side of the workspace, in mm."""
+        return self._outside_view_margin_mm
+
+    @outside_view_margin_mm.setter
+    def outside_view_margin_mm(self, value: float):
+        if not isinstance(value, (int, float)) or isinstance(value, bool):
+            raise TypeError("Outside view margin must be a number.")
+        if not math.isfinite(value):
+            raise ValueError("Outside view margin must be finite.")
+        if not (0.0 <= value <= OUTSIDE_VIEW_MAX_MARGIN_MM):
+            logger.warning(
+                f"Outside view margin {value} is outside range "
+                f"(0.0-{OUTSIDE_VIEW_MAX_MARGIN_MM}). "
+                "Clamping to nearest bound."
+            )
+            value = max(0.0, min(value, OUTSIDE_VIEW_MAX_MARGIN_MM))
+        value = float(value)
+        if self._outside_view_margin_mm == value:
+            return
+        logger.debug(
+            f"Camera outside_view_margin_mm changed from "
+            f"{self._outside_view_margin_mm} to {value}"
+        )
+        self._outside_view_margin_mm = value
+        self.changed.send(self)
+
+    @property
+    def outside_view_transparency(self) -> float:
+        """Visibility of the outside view; higher is more visible."""
+        return self._outside_view_transparency
+
+    @outside_view_transparency.setter
+    def outside_view_transparency(self, value: float):
+        if not isinstance(value, (int, float)) or isinstance(value, bool):
+            raise TypeError("Outside view transparency must be a number.")
+        if not math.isfinite(value):
+            raise ValueError("Outside view transparency must be finite.")
+        if not (0.0 <= value <= 1.0):
+            logger.warning(
+                f"Outside view transparency {value} is outside range "
+                "(0.0-1.0). Clamping to nearest bound."
+            )
+            value = max(0.0, min(value, 1.0))
+        value = float(value)
+        if self._outside_view_transparency == value:
+            return
+        logger.debug(
+            f"Camera outside_view_transparency changed from "
+            f"{self._outside_view_transparency} to {value}"
+        )
+        self._outside_view_transparency = value
+        self.changed.send(self)
+
     @property
     def resolution(self) -> tuple[int, int] | None:
         return self._resolution
@@ -656,6 +738,9 @@ def to_dict(self) -> dict[str, Any]:
             "transparency": self.transparency,
             "denoise": self.denoise,
             "prefer_yuyv": self.prefer_yuyv,
+            "outside_view_enabled": self.outside_view_enabled,
+            "outside_view_margin_mm": self.outside_view_margin_mm,
+            "outside_view_transparency": self.outside_view_transparency,
             "distortion_k1": self.distortion_k1,
             "distortion_k2": self.distortion_k2,
             "distortion_p1": self.distortion_p1,
@@ -725,6 +810,9 @@ def from_dict(cls, data: dict[str, Any]) -> "Camera":
             "transparency",
             "denoise",
             "prefer_yuyv",
+            "outside_view_enabled",
+            "outside_view_margin_mm",
+            "outside_view_transparency",
             "image_to_world",
             "distortion_k1",
             "distortion_k2",
@@ -768,6 +856,15 @@ def from_dict(cls, data: dict[str, Any]) -> "Camera":
         camera.transparency = data.get("transparency", camera.transparency)
         camera.denoise = data.get("denoise", 0.0)
         camera.prefer_yuyv = data.get("prefer_yuyv", False)
+        camera.outside_view_enabled = data.get(
+            "outside_view_enabled", camera.outside_view_enabled
+        )
+        camera.outside_view_margin_mm = data.get(
+            "outside_view_margin_mm", camera.outside_view_margin_mm
+        )
+        camera.outside_view_transparency = data.get(
+            "outside_view_transparency", camera.outside_view_transparency
+        )
 
         camera.distortion_k1 = data.get("distortion_k1", 0.0)
         camera.distortion_k2 = data.get("distortion_k2", 0.0)