diff --git a/src/aare/gui/sam_cam_test.py b/src/aare/gui/sam_cam_test.py new file mode 100644 index 00000000..cb04d716 --- /dev/null +++ b/src/aare/gui/sam_cam_test.py @@ -0,0 +1,929 @@ +from __future__ import annotations + +import time +from dataclasses import dataclass +from time import sleep, perf_counter +from typing import Callable, Optional + +import numpy as np + +from aare.common.beamline import mx_beamline +from aare.devices.area_detector import epicsAD +from aare.daq.devices import BeamlineDevices +# profile_X = PV('X10SA-SAMCAM:Stats5:ProfileAverageX_RBV') +# profile_Y = PV('X10SA-SAMCAM:Stats5:ProfileAverageY_RBV') +# hist = PV('X10SA-SAMCAM:Stats5:Histogram_RBV') +# +# print(sample_cam.expo_rbv.get()) +# print(np.array(profile_X.get()).mean()) +# print(np.array(profile_Y.get()).mean()) +# print(np.array((hist.get())).mean()) +# roi_mean = PV('X10SA-SAMCAM:Stats5:MeanValue_RBV') +# cursorY = PV('X10SA-SAMCAM:Stats5:CursorY') +# cursorY_RBV = PV('X10SA-SAMCAM:Stats5:CursorY_RBV') +# cursorX = PV('X10SA-SAMCAM:Stats5:CursorX') +# cursorX_RBV = PV('X10SA-SAMCAM:Stats5:CursorX_RBV') + +# Python +@dataclass(frozen=True) +class MeteringConfig: + """ + Metering and robustness parameters for dark-background + bright sample scenes. + """ + # Use central ROI to reduce chance of metering random bright junk near edges. + # 1.0 = full frame, 0.7 = central 70% in width/height. + center_roi: float = 0.7 + + # Dynamic threshold: thr = percentile(gray, bg_percentile) + delta_dn + bg_percentile: float = 10.0 + delta_dn: int = 15 + + # If mask is too small, fall back to a larger ROI or whole frame. + min_mask_fraction: float = 0.002 # 0.2% of ROI pixels + + # Winsorization for brightness metric (NOT for clipping metric): + # clamp values above winsor_high before computing percentiles. + winsor_high: int = 245 + + +@dataclass(frozen=True) +class TargetConfig: + """ + Control objectives. + """ + # Hard constraint: keep saturated fraction small. + clip_limit: float = 0.003 # 0.3% of metered pixels >= clip_level + + # Clip threshold (8-bit): treat >=254 as "near-saturated". + clip_level: int = 254 + + # Percentile target of metered (masked) pixels. For metallic base variability, + # p80 is usually more stable than p90. + percentile: float = 80.0 + percentile_target: float = 160.0 + + # Optional: if the metered pixels are too sparse, you can skip updates. + min_metered_pixels: int = 5000 + + +@dataclass(frozen=True) +class LimitsConfig: + """ + Exposure/gain bounds and stepping. + """ + exposure_min_s: float = 0.0001 + exposure_max_s: float = 1.0 + + exposure_effective_min_s: float = 0.002 + exposure_quantum_s: float = 1e-6 + + gain_min: float = 0.0 + gain_max: float = 36.0 + + # Update aggressiveness + exposure_k: float = 0.35 # proportional factor for percentile error + exposure_clip_drop: float = 0.7 # multiply exposure by this when clipping too high + + gain_step: float = 1.0 # gain adjustment step when exposure hits bounds + + # Safety: limit how fast exposure can change to avoid oscillations + max_exposure_scale_up: float = 1.25 + max_exposure_scale_down: float = 0.75 + + +@dataclass +class Metrics: + bg: float + thr: int + n_total: int + n_metered: int + mask_fraction: float + clip_frac: float + p50: float + p70: float + p80: float + p90: float + mean: float + + +def _center_crop(gray: np.ndarray, frac: float) -> np.ndarray: + if frac >= 1.0: + return gray + if frac <= 0.0: + raise ValueError("center_roi must be in (0, 1].") + h, w = gray.shape[:2] + rh = max(1, int(h * frac)) + rw = max(1, int(w * frac)) + y0 = (h - rh) // 2 + x0 = (w - rw) // 2 + return gray[y0:y0 + rh, x0:x0 + rw] + +def _percentile_from_hist(hist: np.ndarray, percentile: float) -> int: + """ + hist: counts per DN bin [0..255] + percentile: 0..100 + returns: DN value (0..255) + """ + total = int(hist.sum()) + if total <= 0: + return 0 + k = int(np.ceil((percentile / 100.0) * total)) + c = np.cumsum(hist) + return int(np.searchsorted(c, k, side="left")) + +def _hist_u8(a: np.ndarray) -> np.ndarray: + """ + Fast histogram for uint8 array -> length 256. + """ + return np.bincount(a.ravel(), minlength=256) + +# def compute_metrics(gray: np.ndarray, met: MeteringConfig, tgt: TargetConfig) -> Metrics: +# """ +# Compute robust metering metrics for 8-bit dark-background scenes. +# """ +# if gray.ndim != 2: +# raise ValueError("compute_metrics expects a 2D grayscale image.") +# g = gray +# if g.dtype != np.uint8: +# # If gray is float from RGB conversion, this handles it gracefully. +# g = np.clip(g, 0, 255).astype(np.uint8) +# +# roi = _center_crop(g, met.center_roi) +# n_total = int(roi.size) +# +# bg = float(np.percentile(roi, met.bg_percentile)) +# thr = int(min(255, max(0, round(bg + met.delta_dn)))) +# +# mask = roi > thr +# n_metered = int(mask.sum()) +# mask_fraction = float(n_metered / max(1, n_total)) +# +# # If mask is too small, broaden the ROI (fallback to full frame). +# if mask_fraction < met.min_mask_fraction: +# roi = g +# n_total = int(roi.size) +# bg = float(np.percentile(roi, met.bg_percentile)) +# thr = int(min(255, max(0, round(bg + met.delta_dn)))) +# mask = roi > thr +# n_metered = int(mask.sum()) +# mask_fraction = float(n_metered / max(1, n_total)) +# +# if n_metered == 0: +# return Metrics( +# bg=bg, thr=thr, +# n_total=n_total, n_metered=0, mask_fraction=0.0, +# clip_frac=0.0, p50=0.0, p70=0.0, p80=0.0, p90=0.0, mean=float(np.mean(roi)), +# ) +# +# vals = roi[mask] +# +# # Clipping fraction on raw values +# clip_frac = float(np.mean(roi >= tgt.clip_level)) +# +# # Winsorize only for percentile/mean (prevents glints from dominating brightness target) +# vals_w = np.minimum(vals, np.uint8(met.winsor_high)) +# +# return Metrics( +# bg=bg, +# thr=thr, +# n_total=n_total, +# n_metered=n_metered, +# mask_fraction=mask_fraction, +# clip_frac=clip_frac, +# p50=float(np.percentile(vals_w, 50)), +# p70=float(np.percentile(vals_w, 70)), +# p80=float(np.percentile(vals_w, 80)), +# p90=float(np.percentile(vals_w, 90)), +# mean=float(np.mean(vals_w)), +# ) + +def compute_metrics(gray: np.ndarray, met: MeteringConfig, tgt: TargetConfig) -> Metrics: + """ + Compute robust metering metrics for 8-bit dark-background scenes. + + Optimized: + - avoids np.percentile on million-pixel arrays (uses 256-bin histograms) + - optional subsampling + """ + if gray.ndim != 2: + raise ValueError("compute_metrics expects a 2D grayscale image.") + + g = gray + if g.dtype != np.uint8: + g = np.clip(g, 0, 255).astype(np.uint8) + + roi_full = _center_crop(g, met.center_roi) + + # Optional subsampling for speed + s = int(getattr(met, "subsample", 1)) + if s > 1: + roi = roi_full[::s, ::s] + else: + roi = roi_full + + n_total = int(roi.size) + + # Background percentile from histogram + hist_roi = _hist_u8(roi) + bg_dn = _percentile_from_hist(hist_roi, met.bg_percentile) + thr = int(min(255, max(0, bg_dn + int(met.delta_dn)))) + + mask = roi > thr + n_metered = int(mask.sum()) + mask_fraction = float(n_metered / max(1, n_total)) + + # If mask too small, fall back to full frame (still subsampled) + if mask_fraction < met.min_mask_fraction: + roi_full = g + if s > 1: + roi = roi_full[::s, ::s] + else: + roi = roi_full + n_total = int(roi.size) + + hist_roi = _hist_u8(roi) + bg_dn = _percentile_from_hist(hist_roi, met.bg_percentile) + thr = int(min(255, max(0, bg_dn + int(met.delta_dn)))) + + mask = roi > thr + n_metered = int(mask.sum()) + mask_fraction = float(n_metered / max(1, n_total)) + + # Clip fraction on ROI (not just masked) + # Uses histogram so it is cheap. + hist_roi = _hist_u8(roi) + clip_bins = hist_roi[int(tgt.clip_level):].sum() + clip_frac = float(clip_bins / max(1, n_total)) + + if n_metered == 0: + # Approx mean from hist (avoids np.mean) + mean_roi = float(np.dot(np.arange(256, dtype=np.float64), hist_roi) / max(1, n_total)) + return Metrics( + bg=float(bg_dn), + thr=thr, + n_total=n_total, + n_metered=0, + mask_fraction=0.0, + clip_frac=clip_frac, + p50=0.0, + p70=0.0, + p80=0.0, + p90=0.0, + mean=mean_roi, + ) + + # Histogram of masked pixels with winsorization applied: + # - compute histogram of masked values + # - fold bins above winsor_high into winsor_high + vals = roi[mask] + hist_vals = _hist_u8(vals) + + wh = int(met.winsor_high) + if wh < 255: + hist_vals[wh] += hist_vals[wh + 1:].sum() + hist_vals[wh + 1:] = 0 + + p50 = float(_percentile_from_hist(hist_vals, 50.0)) + p70 = float(_percentile_from_hist(hist_vals, 70.0)) + p80 = float(_percentile_from_hist(hist_vals, 80.0)) + p90 = float(_percentile_from_hist(hist_vals, 90.0)) + + mean_vals = float(np.dot(np.arange(256, dtype=np.float64), hist_vals) / max(1, hist_vals.sum())) + + return Metrics( + bg=float(bg_dn), + thr=thr, + n_total=n_total, + n_metered=n_metered, + mask_fraction=mask_fraction, + clip_frac=clip_frac, + p50=p50, + p70=p70, + p80=p80, + p90=p90, + mean=mean_vals, + ) + +def _get_percentile_value(m: Metrics, percentile: float) -> float: + if abs(percentile - 50.0) < 1e-6: + return m.p50 + if abs(percentile - 70.0) < 1e-6: + return m.p70 + if abs(percentile - 80.0) < 1e-6: + return m.p80 + if abs(percentile - 90.0) < 1e-6: + return m.p90 + # If you want arbitrary percentiles, compute them directly in compute_metrics. + raise ValueError("This implementation supports percentile ∈ {50, 80, 90} for speed/stability.") + + +class AutoExposureController: + """ + Camera-agnostic controller: you inject how to read image and how to set/get exposure/gain. + + This keeps the logic testable and usable both in DAQ and GUI contexts. + """ + def __init__( + self, + get_gray_image: Callable[[], np.ndarray], + get_exposure_s: Callable[[], float], + set_exposure_s: Callable[[float], None], + get_gain: Callable[[], float], + set_gain: Callable[[float], None], + metering: MeteringConfig | None = None, + target: TargetConfig | None = None, + limits: LimitsConfig | None = None, + ): + self.get_gray_image = get_gray_image + self.get_exposure_s = get_exposure_s + self.set_exposure_s = set_exposure_s + self.get_gain = get_gain + self.set_gain = set_gain + + self.metering = metering or MeteringConfig() + self.target = target or TargetConfig() + self.limits = limits or LimitsConfig() + + # simple exponential smoothing for metrics + self._ema_clip: Optional[float] = None + self._ema_p: Optional[float] = None + + def _clamp(self, x: float, lo: float, hi: float) -> float: + # Enforce "effective" minimum to avoid PV rounding to 0. + lo_eff = max(lo, self.limits.exposure_effective_min_s) + + x = max(lo_eff, min(hi, x)) + + q = float(self.limits.exposure_quantum_s) + if q > 0: + x = round(x / q) * q + + # Re-enforce bounds after rounding + x = max(lo_eff, min(hi, x)) + return x + + def _settle_after_change(self, new_exp_s: float, base_settle_s: float, fps:float = 25.0) -> None: + """ + Wait long enough that the next acquired frame reflects the new exposure. + + For your typical 2–100 ms exposure range, this keeps the loop fast but stable. + """ + t = max(base_settle_s, 1/fps, new_exp_s) + print(t) + t = min(t, 0.35) + sleep(t) + + def _set_exposure_if_changed(self, new_exp: float, current_exp: float) -> bool: + q = float(self.limits.exposure_quantum_s) if self.limits.exposure_quantum_s > 0 else 0.0 + eps = max(1e-9, 0.5 * q) + if abs(new_exp - current_exp) <= eps: + return False + self.set_exposure_s(new_exp) + return True + + def _set_gain_if_changed(self, new_gain: float, current_gain: float) -> bool: + if abs(new_gain - current_gain) < 1e-6: + return False + self.set_gain(new_gain) + return True + + def pre_bracket(self, + base_settle_s: float = 0.01, + max_steps: int = 8, + max_up: float = 3.0, + max_down: float = 0.33, + deadband_dn: float = 25.0) -> None: + """ + Coarse stage that avoids oscillation: + - uses exposure *= target/measured instead of fixed multipliers + - clamps jump size (max_up/max_down) + - reduces jump size if error sign flips (damping) + """ + last_err: float | None = None + up_lim = float(max_up) + dn_lim = float(max_down) + + for _ in range(max_steps): + gray = self.get_gray_image() + m = compute_metrics(gray, self.metering, self.target) + + exp = float(self.get_exposure_s()) + gain = float(self.get_gain()) + + # If we can't meter yet, increase exposure but not insanely. + if m.n_metered < self.target.min_metered_pixels: + new_exp = self._clamp(exp * min(up_lim, 2.0), self.limits.exposure_min_s, self.limits.exposure_max_s) + if new_exp > exp + 1e-12: + self.set_exposure_s(new_exp) + self._settle_after_change(new_exp, base_settle_s) + last_err = None + continue + + new_gain = self._clamp(gain + self.limits.gain_step, self.limits.gain_min, self.limits.gain_max) + self.set_gain(new_gain) + self._settle_after_change(exp, base_settle_s) + last_err = None + continue + + pv = float(_get_percentile_value(m, self.target.percentile)) + err = float(self.target.percentile_target - pv) + + # If we're already close in the coarse sense, stop bracketing. + if abs(err) <= float(deadband_dn) and (m.clip_frac <= self.target.clip_limit): + break + + # If clipping is high, reduce exposure using a bounded drop (avoid slamming to ~0). + if m.clip_frac > self.target.clip_limit: + # Drop factor based on severity but bounded + r = min(5.0, m.clip_frac / max(1e-12, self.target.clip_limit)) # 1..5 + drop = 0.85 - (r - 1.0) * (0.85 - 0.55) / (5.0 - 1.0) # 0.85..0.55 + drop = max(0.50, min(0.90, drop)) # hard bounds + new_exp = self._clamp(exp * drop, self.limits.exposure_min_s, self.limits.exposure_max_s) + if abs(new_exp - exp) > 1e-12: + self.set_exposure_s(new_exp) + self._settle_after_change(new_exp, base_settle_s) + last_err = err + continue + + # Anti-oscillation: if error sign flips, reduce allowed jump size + if last_err is not None and (err == 0.0 or (err > 0) != (last_err > 0)): + up_lim = max(1.3, up_lim * 0.5) + dn_lim = min(0.8, dn_lim + (1.0 - dn_lim) * 0.5) # move dn_lim toward 1.0 (less aggressive down) + + # Main coarse jump: exp *= target / measured + # If pv is tiny, protect division. + denom = max(1.0, pv) + ratio = float(self.target.percentile_target / denom) + + # Clamp jump size so we don't overreact + ratio = max(dn_lim, min(up_lim, ratio)) + + new_exp = self._clamp(exp * ratio, self.limits.exposure_min_s, self.limits.exposure_max_s) + if abs(new_exp - exp) > 1e-12: + self.set_exposure_s(new_exp) + self._settle_after_change(new_exp, base_settle_s) + + last_err = err + + def time_test(self): + t0 = perf_counter() + gray = self.get_gray_image() + t1 = perf_counter() + m = compute_metrics(gray, self.metering, self.target) + t2 = perf_counter() + exp = float(self.get_exposure_s()) + gain = float(self.get_gain()) + t3 = perf_counter() + print("gray:", t1 - t0, "metrics:", t2 - t1, "pvs:", t3 - t2) + + def run_once(self, settle_s: float = 0.15, max_iters: int = 12, verbose: bool = True) -> tuple[ + float, float, Metrics]: + """ + For stationary / after centering: iterate until within constraints (or max_iters). + + Returns: (final_exposure, final_gain, last_metrics) + """ + base_settle_s = float(settle_s) # keep caller's base settle; do NOT overwrite it in-loop + last_m = None + + for i in range(max_iters): + t0 = perf_counter() + gray = self.get_gray_image() + t1 = perf_counter() + m = compute_metrics(gray, self.metering, self.target) + t2 = perf_counter() + + exp = float(self.get_exposure_s()) + gain = float(self.get_gain()) + pv = _get_percentile_value(m, self.target.percentile) + t3 = perf_counter() + + if verbose: + print( + f"[AE once {i + 1:02d}/{max_iters}] exp={exp:.6f}s gain={gain:.2f} " + f"clip={m.clip_frac:.4f} p{int(self.target.percentile)}={pv:.1f} " + f"mask={m.mask_fraction * 100:.2f}% thr={m.thr} n={m.n_metered} " + f"timing(gray={t1 - t0:.3f}s metrics={t2 - t1:.3f}s pvs={t3 - t2:.3f}s)" + ) + + if (m.clip_frac <= self.target.clip_limit) and (abs(self.target.percentile_target - pv) <= 5.0): + break + + # Example: measure set+settle times for exposure changes + if m.clip_frac > self.target.clip_limit: + r = min(5.0, m.clip_frac / max(1e-12, self.target.clip_limit)) + adaptive_drop = 0.85 - (r - 1.0) * (0.85 - 0.40) / (5.0 - 1.0) + adaptive_drop = self._clamp(adaptive_drop, 0.35, 0.90) + + new_exp = self._clamp(exp * adaptive_drop, self.limits.exposure_min_s, self.limits.exposure_max_s) + + ts = perf_counter() + self.set_exposure_s(new_exp) + te = perf_counter() + + tw0 = perf_counter() + self._settle_after_change(new_exp, base_settle_s) + tw1 = perf_counter() + + if verbose: + print(f" set_exp={te - ts:.3f}s settle={tw1 - tw0:.3f}s") + continue + + # Protect highlights + if m.clip_frac > self.target.clip_limit: + tr = perf_counter() + r = min(5.0, m.clip_frac / max(1e-12, self.target.clip_limit)) + adaptive_drop = 0.85 - (r - 1.0) * (0.85 - 0.40) / (5.0 - 1.0) + adaptive_drop = self._clamp(adaptive_drop, 0.35, 0.90) + ta = perf_counter() + + tne = perf_counter() + new_exp = self._clamp(exp * adaptive_drop, self.limits.exposure_min_s, self.limits.exposure_max_s) + ts = perf_counter() + self.set_exposure_s(new_exp) + te = perf_counter() + tw1 = perf_counter() + self._settle_after_change(new_exp, base_settle_s) + tw0 = perf_counter() + if verbose: + print(f"new_exp {tne-ts}s set_exp={te - ts:.3f}s settle={tw1 - tw0:.3f}s") + continue + + # Percentile control + err = float(self.target.percentile_target - pv) + + err_fullscale = 50.0 + err_norm = min(1.0, abs(err) / err_fullscale) + scale_up_limit = 1.12 + err_norm * (2.20 - 1.12) + scale_dn_limit = 0.88 - err_norm * (0.88 - 0.45) + + scale = 1.0 + self.limits.exposure_k * (err / 255.0) + scale = self._clamp(scale, scale_dn_limit, scale_up_limit) + tne = perf_counter() + new_exp = self._clamp(exp * scale, self.limits.exposure_min_s, self.limits.exposure_max_s) + ts = perf_counter() + if abs(new_exp - exp) < 1e-12: + # only adjust gain at limits + if err > 5 and exp >= self.limits.exposure_max_s - 1e-12: + new_gain = self._clamp(gain + self.limits.gain_step, self.limits.gain_min, self.limits.gain_max) + self.set_gain(new_gain) + self._settle_after_change(exp, base_settle_s) + elif err < -5 and exp <= self.limits.exposure_min_s + 1e-12: + new_gain = self._clamp(gain - self.limits.gain_step, self.limits.gain_min, self.limits.gain_max) + self.set_gain(new_gain) + self._settle_after_change(exp, base_settle_s) + else: + break + else: + ts2 = perf_counter() + self.set_exposure_s(new_exp) + te = perf_counter() + tw1 = perf_counter() + self._settle_after_change(new_exp, base_settle_s) + tw0 = perf_counter() + if verbose: + print(f"new_exp {ts-tne}s set_exp={te - ts2:.3f}s settle={tw0 - tw1:.3f}s") + + exp = float(self.get_exposure_s()) + gain = float(self.get_gain()) + return exp, gain, last_m if last_m is not None else compute_metrics(self.get_gray_image(), self.metering, + self.target) + + def run_continuous( + self, + duration_s: float = 10.0, + update_hz: float = 3.0, + ema_alpha: float = 0.4, + verbose: bool = True, + ) -> None: + """ + For moving scenes: low-rate updates + EMA smoothing to avoid oscillation. + + It will update exposure/gain at `update_hz` for `duration_s`. + """ + if update_hz <= 0: + raise ValueError("update_hz must be > 0") + dt = 1.0 / update_hz + + t0 = perf_counter() + k = 0 + while (perf_counter() - t0) < duration_s: + gray = self.get_gray_image() + m = compute_metrics(gray, self.metering, self.target) + + # Skip if metering unreliable + if m.n_metered < self.target.min_metered_pixels: + if verbose: + print(f"[AE cont] skipped (n_metered={m.n_metered})") + sleep(dt) + continue + + pv = _get_percentile_value(m, self.target.percentile) + + # EMA smoothing + if self._ema_clip is None: + self._ema_clip = m.clip_frac + self._ema_p = pv + else: + self._ema_clip = ema_alpha * m.clip_frac + (1 - ema_alpha) * self._ema_clip + self._ema_p = ema_alpha * pv + (1 - ema_alpha) * self._ema_p + + exp = float(self.get_exposure_s()) + gain = float(self.get_gain()) + + if verbose: + print( + f"[AE cont {k:04d}] exp={exp:.6f}s gain={gain:.2f} " + f"clip={self._ema_clip:.4f} p{int(self.target.percentile)}={self._ema_p:.1f} " + f"mask={m.mask_fraction*100:.2f}%" + ) + + # Control using smoothed metrics + if self._ema_clip > self.target.clip_limit: + new_exp = exp * self.limits.exposure_clip_drop + new_exp = max(new_exp, exp * self.limits.max_exposure_scale_down) + new_exp = self._clamp(new_exp, self.limits.exposure_min_s, self.limits.exposure_max_s) + self.set_exposure_s(new_exp) + else: + err = float(self.target.percentile_target - self._ema_p) + scale = 1.0 + self.limits.exposure_k * (err / 255.0) + scale = self._clamp(scale, self.limits.max_exposure_scale_down, self.limits.max_exposure_scale_up) + new_exp = self._clamp(exp * scale, self.limits.exposure_min_s, self.limits.exposure_max_s) + + if abs(new_exp - exp) > 1e-12: + self.set_exposure_s(new_exp) + else: + # Use gain only if pinned at exposure limit + if err > 8 and exp >= self.limits.exposure_max_s - 1e-12: + self.set_gain(self._clamp(gain + self.limits.gain_step, self.limits.gain_min, self.limits.gain_max)) + elif err < -8 and exp <= self.limits.exposure_min_s + 1e-12: + self.set_gain(self._clamp(gain - self.limits.gain_step, self.limits.gain_min, self.limits.gain_max)) + + k += 1 + sleep(dt) + +# Python +def build_controller(sample_cam, lim: LimitsConfig, tgt: TargetConfig, met:MeteringConfig) -> AutoExposureController: + # sample_cam should be whatever your EPICS/AD wrapper object is. + # Important: ensure camera is in manual exposure/gain mode before control. + + def get_gray() -> np.ndarray: + img = sample_cam.get_image(gray=True) + return img # may be float; controller converts/clamps to uint8 + + def get_exp() -> float: + return float(sample_cam.expo_rbv.get()) + + def set_exp(v: float) -> None: + sample_cam.expo.put(float(v), wait=False) + + def get_gain() -> float: + return float(sample_cam.gain_rbv.get()) + + def set_gain(v: float) -> None: + sample_cam.gain.put(float(v), wait=False) + + + + return AutoExposureController(get_gray, get_exp, set_exp, get_gain, set_gain, metering=met, target=tgt, limits=lim) + +def _now_s() -> float: + return perf_counter() + +def _fmt_ms(s: float) -> str: + return f"{s * 1000.0:.1f} ms" + +def _condition_name(reflector_up: bool, back_light: float) -> str: + if reflector_up and back_light > 0.91: + return "reflector_up + backlight_max" + if reflector_up and back_light <= 0.91: + return "reflector_up + backlight_off" + return "reflector_down" + +def _select_profiles(devs) -> tuple[MeteringConfig, TargetConfig, LimitsConfig]: + """ + Choose metering/targets/limits based on current lighting/reflector state. + + IMPORTANT: + - Use a fine exposure_quantum_s for backlight if you want sub-ms exposures. + - exposure_effective_min_s can be as low as 50 us in backlight mode (per your tests). + """ + # You can keep your existing metering/targets here, or define distinct profiles. + # These are conservative defaults; tweak as needed. + metering_front = MeteringConfig(center_roi=0.7, bg_percentile=10.0, delta_dn=15, winsor_high=230) + targets_front = TargetConfig(clip_limit=0.01, percentile=70.0, percentile_target=160.0, min_metered_pixels=5000) + + metering_back = MeteringConfig(center_roi=0.6, bg_percentile=10.0, delta_dn=8, winsor_high=245) + targets_back = TargetConfig(clip_limit=0.02, percentile=70.0, percentile_target=170.0, min_metered_pixels=2000) + + if devs.reflector_up and devs.back_light > 0.91: + limits = LimitsConfig( + exposure_min_s=0.00005, + exposure_max_s=0.1, + exposure_effective_min_s=0.00005, + exposure_quantum_s=1e-6, # NOT 0.001: you want sub-ms capability here + gain_min=0.0, + gain_max=36.0, + exposure_k=0.70, + max_exposure_scale_up=2.2, + max_exposure_scale_down=0.45, + ) + return metering_back, targets_back, limits + + if devs.reflector_up: + limits = LimitsConfig( + exposure_min_s=0.0005, + exposure_max_s=0.2, + exposure_effective_min_s=0.001, # you said 1 ms is safe in backlight mode + exposure_quantum_s=1e-4, + gain_min=0.0, + gain_max=36.0, + exposure_k=0.80, + max_exposure_scale_up=2.2, + max_exposure_scale_down=0.45, + ) + return metering_back, targets_back, limits + + limits = LimitsConfig( + exposure_min_s=0.002, + exposure_max_s=0.5, + exposure_effective_min_s=0.002, + exposure_quantum_s=1e-4, + gain_min=0.0, + gain_max=36.0, + exposure_k=0.80, + max_exposure_scale_up=2.2, + max_exposure_scale_down=0.45, + ) + return metering_front, targets_front, limits + +def _bench_run_one(ctrl: AutoExposureController, + start_exp_s: float, + start_gain: float, + settle_s: float, + max_iters: int, + label: str, + tol_dn: float) -> dict: + """ + Sets starting exposure/gain, then times run_once convergence. + """ + # Prime starting point + ctrl.set_gain(float(start_gain)) + ctrl.set_exposure_s(float(start_exp_s)) + ctrl._settle_after_change(float(start_exp_s), settle_s) + + t0 = _now_s() + end_exp, end_gain, m = ctrl.run_once(settle_s=settle_s, max_iters=max_iters, verbose=False) + dt = _now_s() - t0 + + pv = _get_percentile_value(m, ctrl.target.percentile) + ok = (m.clip_frac <= ctrl.target.clip_limit) and (abs(ctrl.target.percentile_target - pv) <= float(tol_dn)) + + return { + "label": label, + "t_s": dt, + "start_exp_s": start_exp_s, + "start_gain": start_gain, + "end_exp_s": float(end_exp), + "end_gain": float(end_gain), + "clip": float(m.clip_frac), + "p": float(pv), + "mask_frac": float(m.mask_fraction), + "ok": bool(ok), + "n_metered": int(m.n_metered), + "tol_dn": float(tol_dn), + } + +def benchmark_controller(devs, + sample_cam, + expected_exp_s: dict[str, float], + tolerance_dn: dict[str, float] | None = None, + trials: int = 5, + settle_s: float = 0.01, + max_iters: int = 20, + start_gain: float = 0.0) -> None: + """ + Benchmarks controller convergence speed for three lighting conditions. + + tolerance_dn: optional per-condition tolerance on the chosen percentile metric. + """ + cond = _condition_name(devs.reflector_up, float(devs.back_light)) + met, tgt, lim = _select_profiles(devs) + ctrl = build_controller(sample_cam, lim=lim, tgt=tgt, met=met) + + exp_expected = float(expected_exp_s.get(cond, 0.01)) + exp_min = float(lim.exposure_min_s) + exp_max = float(lim.exposure_max_s) + + tol_dn = 5.0 if tolerance_dn is None else float(tolerance_dn.get(cond, 5.0)) + + starts = [ + ("start=min", exp_min), + ("start=expected", exp_expected), + ("start=max", exp_max), + ] + + print(f"\n=== Benchmark: {cond} ===") + print(f"limits: exp[{lim.exposure_min_s} .. {lim.exposure_max_s}] effective_min={lim.exposure_effective_min_s} quantum={lim.exposure_quantum_s}") + print(f"target: p{int(tgt.percentile)}={tgt.percentile_target} clip_limit={tgt.clip_limit} tol=±{tol_dn:.1f}DN settle_s(base)={settle_s} max_iters={max_iters} trials={trials}") + + results: list[dict] = [] + for label, start_exp in starts: + for k in range(trials): + r = _bench_run_one( + ctrl=ctrl, + start_exp_s=start_exp, + start_gain=start_gain, + settle_s=settle_s, + max_iters=max_iters, + label=label, + tol_dn=tol_dn, + ) + r["trial"] = k + 1 + results.append(r) + + # Print summary + for label, _ in starts: + rr = [r for r in results if r["label"] == label] + times = np.array([r["t_s"] for r in rr], dtype=float) + ok_rate = sum(1 for r in rr if r["ok"]) / max(1, len(rr)) + print( + f"{label:14s} mean={_fmt_ms(times.mean())} p95={_fmt_ms(np.percentile(times, 95))} " + f"min={_fmt_ms(times.min())} max={_fmt_ms(times.max())} ok={ok_rate*100:.0f}%" + ) + + print("\nlabel,trial,t_ms,start_exp_ms,end_exp_ms,start_gain,end_gain,clip,p,mask_frac,tol_dn,ok,n_metered") + for r in results: + print( + f"{r['label']},{r['trial']},{r['t_s']*1000.0:.3f}," + f"{r['start_exp_s']*1000.0:.6f},{r['end_exp_s']*1000.0:.6f}," + f"{r['start_gain']:.2f},{r['end_gain']:.2f}," + f"{r['clip']:.6f},{r['p']:.3f},{r['mask_frac']:.6f},{r['tol_dn']:.1f},{int(r['ok'])},{r['n_metered']}" + ) + + +sample_cam = epicsAD(f"{mx_beamline().name.upper()}-SAMCAM:") +devs = BeamlineDevices(mx_beamline()) +metering = MeteringConfig(center_roi=0.7, bg_percentile=10.0, delta_dn=15, winsor_high=230) +targets = TargetConfig(clip_limit=0.01, percentile=70.0, percentile_target=160.0) + + +# +if devs.reflector_up and devs.back_light > 0.91: + limits = LimitsConfig(exposure_min_s=0.00005, exposure_max_s=0.1, exposure_effective_min_s=0.00005, + exposure_quantum_s=1e-6, + gain_min=0.0, gain_max=36.0, exposure_k=0.70, max_exposure_scale_up=1.6, + max_exposure_scale_down=0.6) + metering = MeteringConfig( + center_roi=0.6, + bg_percentile=10.0, + delta_dn=8, + winsor_high=245, + ) + targets = TargetConfig( + clip_limit=0.02, # allow more clipping (bright field can clip) + percentile=70.0, + percentile_target=170.0, + min_metered_pixels=2000, # can be lower because backlight mask is usually strong + ) + +elif devs.reflector_up: + limits = LimitsConfig(exposure_min_s=0.0005, exposure_max_s=0.2, exposure_effective_min_s=0.002, + exposure_quantum_s=0.001, + gain_min=0.0, gain_max=36.0, exposure_k=0.80, max_exposure_scale_up=1.6, + max_exposure_scale_down=0.6) +else: + limits = LimitsConfig(exposure_min_s=0.002, exposure_max_s=0.5, exposure_effective_min_s=0.002, + exposure_quantum_s=0.001, + gain_min=0.0, gain_max=36.0, exposure_k=0.80, max_exposure_scale_up=1.6, + max_exposure_scale_down=0.6) + targets = TargetConfig( + clip_limit=0.01, + percentile=70.0, + percentile_target=160.0, + min_metered_pixels=5000, + ) + metering = MeteringConfig( + center_roi=0.7, + bg_percentile=10.0, + delta_dn=15, + winsor_high=230, + ) +ctrl = build_controller(sample_cam, lim=limits, tgt=targets, met=metering) + +# After centering and stationary: +st = time.perf_counter() +exp, gain, m = ctrl.run_once(settle_s=0.0, max_iters=20, verbose=True) +print(f"run_once took {time.perf_counter() - st} s") + +# expected = { +# "reflector_up + backlight_max": 0.001, # ~1 ms typical +# "reflector_up + backlight_off": 0.025, # adjust based on your observed behaviour +# "reflector_down": 0.052, # 52 ms you reported +# } +# tolerance = { +# "reflector_up + backlight_max": 5.0, +# "reflector_up + backlight_off": 5.0, +# "reflector_down": 5.0, +# } +# benchmark_controller(devs=devs, sample_cam=sample_cam, expected_exp_s=expected, trials=5, settle_s=0.01, max_iters=20, start_gain=0.0) \ No newline at end of file