[WIP] Sqm enhancements

python/PiFinder/auto_exposure.py

@@ -9,11 +9,9 @@
 
 The controller targets 17 matched stars (acceptable range: 12-22) which provides
 reliable plate solving while avoiding over-saturation and maintaining good performance.
-Rate limiting on downward adjustments reduces CPU usage.
 """
 
 import logging
-import time
 from abc import ABC, abstractmethod
 from typing import List, Optional
 
@@ -121,10 +119,10 @@ def __init__(
         self._repeat_count = 0
 
         # Sweep pattern: exposure values in microseconds
-        # Uses doubling pattern (2x each step)
+        # Start at 400ms (reasonable middle), sweep up, then try shorter exposures
         # Note: This is intentionally NOT using generate_exposure_sweep() because
-        # it uses a specific doubling pattern
-        self._exposures = [25000, 50000, 100000, 200000, 400000, 800000, 1000000]
+        # it uses a specific pattern optimized for recovery
+        self._exposures = [400000, 800000, 1000000, 200000, 100000, 50000, 25000]
         self._repeats_per_exposure = 2  # Try each exposure 2 times
 
         logger.info(
@@ -619,6 +617,183 @@ def reset(self) -> None:
         logger.debug("HistogramZeroStarHandler reset")
 
 
+class ExposureSNRController:
+    """
+    SNR-based auto exposure for SQM measurements.
+
+    Targets a minimum background SNR and exposure time instead of star count.
+    This provides more stable, longer exposures that are better for accurate
+    SQM measurements compared to the histogram-based approach.
+
+    Strategy:
+    - Target specific background level above noise floor
+    - Derive thresholds from camera profile (bit depth, bias offset)
+    - Slower adjustments for stability
+    """
+
+    def __init__(
+        self,
+        min_exposure: int = 10000,  # 10ms minimum
+        max_exposure: int = 1000000,  # 1.0s maximum
+        target_background: int = 30,  # Target background level in ADU
+        min_background: int = 15,  # Minimum acceptable background
+        max_background: int = 100,  # Maximum before saturating
+        adjustment_factor: float = 1.3,  # Gentle adjustments (30% steps)
+    ):
+        """
+        Initialize SNR-based auto exposure.
+
+        Args:
+            min_exposure: Minimum exposure in microseconds (default 10ms)
+            max_exposure: Maximum exposure in microseconds (default 1000ms)
+            target_background: Target median background level in ADU
+            min_background: Minimum acceptable background (increase if below)
+            max_background: Maximum acceptable background (decrease if above)
+            adjustment_factor: Multiplicative adjustment step (default 1.3 = 30%)
+        """
+        self.min_exposure = min_exposure
+        self.max_exposure = max_exposure
+        self.target_background = target_background
+        self.min_background = min_background
+        self.max_background = max_background
+        self.adjustment_factor = adjustment_factor
+
+        logger.info(
+            f"AutoExposure SNR: target_bg={target_background}, "
+            f"range=[{min_background}, {max_background}] ADU, "
+            f"exp_range=[{min_exposure/1000:.0f}, {max_exposure/1000:.0f}]ms, "
+            f"adjustment={adjustment_factor}x"
+        )
+
+    @classmethod
+    def from_camera_profile(
+        cls,
+        camera_type: str,
+        min_exposure: int = 10000,
+        max_exposure: int = 1000000,
+        adjustment_factor: float = 1.3,
+    ) -> "ExposureSNRController":
+        """
+        Create controller with thresholds derived from camera profile.
+
+        Calculates min/target/max background based on bit depth and bias offset.
+
+        Args:
+            camera_type: Camera type (e.g., "imx296_processed", "imx462_processed")
+            min_exposure: Minimum exposure in microseconds
+            max_exposure: Maximum exposure in microseconds
+            adjustment_factor: Multiplicative adjustment step
+
+        Returns:
+            ExposureSNRController configured for the camera
+        """
+        from PiFinder.sqm.camera_profiles import get_camera_profile
+
+        profile = get_camera_profile(camera_type)
+
+        # Derive thresholds from camera specs
+        max_adu = (2 ** profile.bit_depth) - 1
+        bias = profile.bias_offset
+
+        # min_background: bias + margin (2x bias or bias + 8, whichever larger)
+        min_background = int(max(bias * 2, bias + 8))
+
+        # max_background: ~40% of full range (avoid saturation/nonlinearity)
+        max_background = int(max_adu * 0.4)
+
+        # target_background: just above min (lower = shorter exposure = more linear response)
+        target_background = min_background + 2
+
+        logger.info(
+            f"SNR controller from {camera_type}: "
+            f"bit_depth={profile.bit_depth}, bias={bias:.0f}, "
+            f"thresholds=[{min_background}, {target_background}, {max_background}]"
+        )
+
+        return cls(
+            min_exposure=min_exposure,
+            max_exposure=max_exposure,
+            target_background=target_background,
+            min_background=min_background,
+            max_background=max_background,
+            adjustment_factor=adjustment_factor,
+        )
+
+    def update(
+        self,
+        current_exposure: int,
+        image: Image.Image,
+        noise_floor: Optional[float] = None,
+        **kwargs  # Ignore other params (matched_stars, etc.)
+    ) -> Optional[int]:
+        """
+        Update exposure based on background level.
+
+        Args:
+            current_exposure: Current exposure in microseconds
+            image: Current image for analysis
+            noise_floor: Adaptive noise floor from SQM calculator (if available)
+            **kwargs: Ignored (for compatibility with PID interface)
+
+        Returns:
+            New exposure in microseconds, or None if no change needed
+        """
+        # Use adaptive noise floor if available, otherwise fall back to static config
+        # Need margin above noise floor so background_corrected isn't near zero
+        if noise_floor is not None:
+            min_bg = noise_floor + 2
+        else:
+            min_bg = self.min_background
+
+        # Analyze image
+        if image.mode != "L":
+            image = image.convert("L")
+        img_array = np.asarray(image, dtype=np.float32)
+
+        # Use 10th percentile as background estimate (dark pixels)
+        background = float(np.percentile(img_array, 10))
+
+        logger.debug(
+            f"SNR AE: bg={background:.1f}, min={min_bg:.1f} ADU, exp={current_exposure/1000:.0f}ms"
+        )
+
+        # Determine adjustment
+        new_exposure = None
+
+        if background < min_bg:
+            # Too dark - increase exposure
+            new_exposure = int(current_exposure * self.adjustment_factor)
+            logger.info(
+                f"SNR AE: Background too low ({background:.1f} < {min_bg:.1f}), "
+                f"increasing exposure {current_exposure/1000:.0f}ms → {new_exposure/1000:.0f}ms"
+            )
+        elif background > self.max_background:
+            # Too bright - decrease exposure
+            new_exposure = int(current_exposure / self.adjustment_factor)
+            logger.info(
+                f"SNR AE: Background too high ({background:.1f} > {self.max_background}), "
+                f"decreasing exposure {current_exposure/1000:.0f}ms → {new_exposure/1000:.0f}ms"
+            )
+        else:
+            # Background is in acceptable range
+            logger.debug(f"SNR AE: OK (bg={background:.1f} ADU)")
+            return None
+
+        # Clamp to limits
+        new_exposure = max(self.min_exposure, min(self.max_exposure, new_exposure))
+        return new_exposure
+
+    def get_status(self) -> dict:
+        return {
+            "mode": "SNR",
+            "target_background": self.target_background,
+            "min_background": self.min_background,
+            "max_background": self.max_background,
+            "min_exposure": self.min_exposure,
+            "max_exposure": self.max_exposure,
+        }
+
+
 class ExposurePIDController:
     """
     PID controller for automatic camera exposure adjustment.
@@ -631,56 +806,37 @@ class ExposurePIDController:
     def __init__(
         self,
         target_stars: int = 17,
-        gains_decrease: tuple = (
-            500.0,  # Kp: Conservative (was 2000, reduced 75% to prevent crash)
-            5.0,  # Ki: Minimal (was 10, reduced to prevent drift)
-            250.0,  # Kd: Proportional (was 750, reduced 67%)
-        ),  # Kp, Ki, Kd for too many stars (conservative descent)
-        gains_increase: tuple = (
-            4000.0,  # Kp: Moderate aggression (was 8000, reduced 50%)
-            250.0,  # Ki: Moderate (was 500, reduced 50%)
-            1500.0,  # Kd: Moderate (was 3000, reduced 50%)
-        ),  # Kp, Ki, Kd for too few stars (faster ascent)
+        gains: tuple = (2000.0, 100.0, 750.0),  # Kp, Ki, Kd
         min_exposure: int = 25000,
         max_exposure: int = 1000000,
         deadband: int = 5,
-        update_interval: float = 0.5,  # Minimum seconds between decreasing adjustments
         zero_star_handler: Optional[ZeroStarHandler] = None,
     ):
-        """
-        Initialize PID controller with asymmetric gains.
-
-        Uses conservative gains when decreasing exposure (gentle descent),
-        aggressive gains when increasing (fast recovery).
-        """
+        """Initialize PID controller."""
         self.target_stars = target_stars
-        self.gains_decrease = gains_decrease
-        self.gains_increase = gains_increase
+        self.gains = gains
         self.min_exposure = min_exposure
         self.max_exposure = max_exposure
         self.deadband = deadband
-        self.update_interval = update_interval
 
         self._integral = 0.0
         self._last_error: Optional[float] = None
         self._zero_star_count = 0
-        self._last_adjustment_time = 0.0
+        self._nonzero_star_count = 0  # Hysteresis: consecutive non-zero solves
         self._zero_star_handler = zero_star_handler or SweepZeroStarHandler(
             min_exposure=min_exposure, max_exposure=max_exposure
         )
 
         logger.info(
             f"AutoExposure PID: target={target_stars}, deadband={deadband}, "
-            f"update_interval={update_interval}s, "
-            f"gains_dec={gains_decrease}, gains_inc={gains_increase}, "
-            f"range=[{min_exposure}, {max_exposure}]µs"
+            f"gains={gains}, range=[{min_exposure}, {max_exposure}]µs"
         )
 
     def reset(self) -> None:
         self._integral = 0.0
         self._last_error = None
         self._zero_star_count = 0
-        self._last_adjustment_time = 0.0
+        self._nonzero_star_count = 0
         self._zero_star_handler.reset()
         logger.debug("PID controller reset")
 
@@ -706,24 +862,23 @@ def _handle_zero_stars(
         )
 
     def _update_pid(self, matched_stars: int, current_exposure: int) -> Optional[int]:
-        """Core PID algorithm with asymmetric gains."""
+        """Core PID algorithm."""
         error = self.target_stars - matched_stars
 
         if abs(error) <= self.deadband:
             return None
 
-        # Rate limiting: only when decreasing (too many stars)
-        # When increasing (too few stars), respond immediately for faster recovery
-        if error < 0:  # Too many stars, going down
-            current_time = time.time()
-            time_since_last = current_time - self._last_adjustment_time
-            if time_since_last < self.update_interval:
-                return None  # Skip debug log for performance
-        else:
-            current_time = time.time()  # Only get time when needed
+        kp, ki, kd = self.gains
 
-        # Select gains: conservative when decreasing, aggressive when increasing
-        kp, ki, kd = self.gains_decrease if error < 0 else self.gains_increase
+        # Reset integral when error changes sign to prevent accumulated integral
+        # from crashing exposure when conditions change suddenly
+        # (e.g., going from too many stars to too few stars)
+        if self._last_error is not None:
+            if (error > 0 and self._last_error < 0) or (error < 0 and self._last_error > 0):
+                logger.debug(
+                    f"PID: Error sign changed ({self._last_error:.0f} → {error:.0f}), resetting integral"
+                )
+                self._integral = 0.0
 
         # PID calculation
         p_term = kp * error
@@ -746,7 +901,6 @@ def _update_pid(self, matched_stars: int, current_exposure: int) -> Optional[int
                 self._integral -= overshoot / ki
 
         self._last_error = error
-        self._last_adjustment_time = current_time
 
         return clamped_exposure
 
@@ -796,27 +950,18 @@ def set_target(self, target_stars: int) -> None:
         self.target_stars = target_stars
         logger.debug(f"Target stars changed: {old_target} → {target_stars}")
 
-    def set_gains(
-        self,
-        gains_decrease: Optional[tuple] = None,
-        gains_increase: Optional[tuple] = None,
-    ) -> None:
-        """Update PID gains. Each tuple is (Kp, Ki, Kd)."""
-        if gains_decrease is not None:
-            self.gains_decrease = gains_decrease
-        if gains_increase is not None:
-            self.gains_increase = gains_increase
-        logger.debug(f"PID gains: dec={self.gains_decrease}, inc={self.gains_increase}")
+    def set_gains(self, gains: tuple) -> None:
+        """Update PID gains. Tuple is (Kp, Ki, Kd)."""
+        self.gains = gains
+        logger.debug(f"PID gains: {self.gains}")
 
     def get_status(self) -> dict:
         return {
             "target_stars": self.target_stars,
-            "gains_decrease": self.gains_decrease,
-            "gains_increase": self.gains_increase,
+            "gains": self.gains,
             "integral": self._integral,
             "last_error": self._last_error,
             "min_exposure": self.min_exposure,
             "max_exposure": self.max_exposure,
             "deadband": self.deadband,
-            "update_interval": self.update_interval,
         }