mx/AareDAQ
0
1
Fork 0
Code Issues 27 Pull Requests Actions Packages Projects Releases 88 Wiki Activity

Added test script for sam cam auto exposure

Browse Source
This commit is contained in:
appleb_m
2026-01-30 17:04:45 +01:00
parent e6d84a307d
commit ab6ecff677
1 changed files with 929 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aare/gui/sam_cam_test.py
+929
View File
@@ -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)