feat(webcam): add exposure, auto-exposure, gain and brightness control to WebcamViewer

Settings are posted to the camera server's /camera/settings endpoint and read
back on startup in a background thread, so an unreachable server cannot block
device loading. set_exposure_time() disables auto exposure first.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

csaxs_bec/devices/omny/webcam_viewer.py

@@ -5,13 +5,32 @@ import numpy as np
 from ophyd import Device, Component as Cpt, Kind, Signal
 from ophyd_devices import PreviewSignal
 import traceback 
+from urllib.parse import urlsplit
 
 from bec_lib.logger import bec_logger
 
 logger = bec_logger.logger
 
+# The camera server passes "exposure" straight to V4L2 exposure_time_absolute,
+# which is in units of 100 us.
+EXPOSURE_UNIT_MS = 0.1
+# V4L2 auto_exposure menu values used by the camera server.
+AUTO_EXPOSURE_ON = 3  # aperture priority
+AUTO_EXPOSURE_OFF = 1  # manual
+SETTINGS_TIMEOUT_S = 3
+
+
 class WebcamViewer(Device):
-    USER_ACCESS = ["start_live_mode", "stop_live_mode"]
+    USER_ACCESS = [
+        "start_live_mode",
+        "stop_live_mode",
+        "get_exposure_time",
+        "set_exposure_time",
+        "set_auto_exposure_enabled",
+        "set_gain",
+        "set_brightness",
+        "refresh_settings",
+    ]
     preview = Cpt(PreviewSignal, ndim=2, num_rotation_90=0, transpose=False)
     live_mode_enabled = Cpt(
         Signal,
@@ -20,6 +39,24 @@ class WebcamViewer(Device):
         doc="Enable or disable live mode.",
         kind=Kind.config,
     )
+    exposure_time = Cpt(
+        Signal,
+        name="exposure_time",
+        value=0.0,
+        doc="Camera exposure time (ms). Only effective with auto exposure disabled.",
+        kind=Kind.config,
+    )
+    auto_exposure_enabled = Cpt(
+        Signal,
+        name="auto_exposure_enabled",
+        value=True,
+        doc="Enable/disable camera auto exposure.",
+        kind=Kind.config,
+    )
+    gain = Cpt(Signal, name="gain", value=0.0, doc="Camera gain.", kind=Kind.config)
+    brightness = Cpt(
+        Signal, name="brightness", value=0.0, doc="Camera brightness.", kind=Kind.config
+    )
 
     def __init__(self, url:str, name:str, num_rotation_90=0, transpose=False, **kwargs) -> None:
         super().__init__(name=name, **kwargs)
@@ -31,7 +68,17 @@ class WebcamViewer(Device):
         self._live_mode_lock = threading.RLock()
         self.preview.num_rotation_90 = num_rotation_90
         self.preview.transpose = transpose
+        parts = urlsplit(url)
+        self._settings_url = f"{parts.scheme}://{parts.netloc}/camera/settings"
+        self._refreshing = False
         self.live_mode_enabled.subscribe(self._on_live_mode_enabled_changed, run=False)
+        self.exposure_time.subscribe(self._on_exposure_time_changed, run=False)
+        self.auto_exposure_enabled.subscribe(self._on_auto_exposure_enabled_changed, run=False)
+        self.gain.subscribe(self._on_gain_changed, run=False)
+        self.brightness.subscribe(self._on_brightness_changed, run=False)
+        # Load the camera's current settings in the background, so an unreachable
+        # camera server cannot block the device from loading.
+        threading.Thread(target=self._initial_refresh, daemon=True).start()
 
     def start_live_mode(self) -> None:
         self.live_mode_enabled.put(True)
@@ -39,6 +86,93 @@ class WebcamViewer(Device):
     def stop_live_mode(self) -> None:
         self.live_mode_enabled.put(False)
 
+    # ---- camera settings (HTTP /camera/settings on the camera server) ----
+
+    def _post_settings(self, payload: dict) -> None:
+        """POST settings to the camera server; log instead of raising on failure."""
+        try:
+            response = requests.post(
+                self._settings_url, json=payload, timeout=SETTINGS_TIMEOUT_S
+            )
+            response.raise_for_status()
+            applied = response.json().get("applied", {})
+            bad = {k: v for k, v in applied.items() if isinstance(v, str)}
+            if bad:
+                logger.error(f"{self.name}: camera server rejected settings {bad}")
+        except Exception:  # pylint: disable=broad-except
+            logger.error(
+                f"{self.name}: failed to apply {payload} at {self._settings_url}: "
+                f"{traceback.format_exc()}"
+            )
+
+    def _on_exposure_time_changed(self, *args, value, **kwargs) -> None:
+        if self._refreshing:
+            return
+        payload = {"exposure": float(value) / EXPOSURE_UNIT_MS}
+        if self.auto_exposure_enabled.get():
+            logger.warning(
+                f"{self.name}: exposure_time set while auto exposure is enabled; "
+                "it has no effect until set_auto_exposure_enabled(False)."
+            )
+        self._post_settings(payload)
+
+    def _on_auto_exposure_enabled_changed(self, *args, value, **kwargs) -> None:
+        if self._refreshing:
+            return
+        self._post_settings({"auto_exposure": AUTO_EXPOSURE_ON if value else AUTO_EXPOSURE_OFF})
+
+    def _on_gain_changed(self, *args, value, **kwargs) -> None:
+        if not self._refreshing:
+            self._post_settings({"gain": float(value)})
+
+    def _on_brightness_changed(self, *args, value, **kwargs) -> None:
+        if not self._refreshing:
+            self._post_settings({"brightness": float(value)})
+
+    def get_exposure_time(self) -> float:
+        """Get the cached exposure time (ms). Use refresh_settings() to re-read the camera."""
+        return float(self.exposure_time.get())
+
+    def set_exposure_time(self, value: float) -> None:
+        """Set the exposure time (ms). Disables auto exposure first, otherwise the camera ignores it."""
+        if self.auto_exposure_enabled.get():
+            self.auto_exposure_enabled.put(False)
+        self.exposure_time.put(value)
+
+    def set_auto_exposure_enabled(self, enable: bool) -> None:
+        self.auto_exposure_enabled.put(bool(enable))
+
+    def set_gain(self, value: float) -> None:
+        self.gain.put(value)
+
+    def set_brightness(self, value: float) -> None:
+        self.brightness.put(value)
+
+    def _initial_refresh(self) -> None:
+        try:
+            self.refresh_settings()
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.warning(f"{self.name}: could not read camera settings from {self._settings_url}: {exc}")
+
+    def refresh_settings(self) -> dict:
+        """Read the current settings from the camera server into the Signals."""
+        response = requests.get(self._settings_url, timeout=SETTINGS_TIMEOUT_S)
+        response.raise_for_status()
+        data = response.json()
+        self._refreshing = True
+        try:
+            if "exposure" in data:
+                self.exposure_time.put(float(data["exposure"]) * EXPOSURE_UNIT_MS)
+            if "auto_exposure" in data:
+                self.auto_exposure_enabled.put(float(data["auto_exposure"]) >= 2.0)
+            if "gain" in data:
+                self.gain.put(float(data["gain"]))
+            if "brightness" in data:
+                self.brightness.put(float(data["brightness"]))
+        finally:
+            self._refreshing = False
+        return data
+
     def _on_live_mode_enabled_changed(self, *args, value, **kwargs) -> None:
         self._apply_live_mode(bool(value))