removed unused devices, assocaited tests and old sam cam test file
This commit is contained in:
appleb_m
@@ -185,59 +185,6 @@ class BeamlineDevices:
|
||||
def set_zoom(self, value: float, /, wait: bool = True):
|
||||
self.__zoom.move(value, wait=wait)
|
||||
|
||||
# Collimator
|
||||
@property
|
||||
def collimator(self) -> float:
|
||||
return self.__collimator_pos.get()
|
||||
|
||||
@collimator.setter
|
||||
def collimator(self, value: float):
|
||||
self.set_collimator(value, wait=True)
|
||||
|
||||
def set_collimator(self, value: float, /, wait: bool = True):
|
||||
self.__collimator_pos.move(value, wait=wait)
|
||||
|
||||
# Scintillator
|
||||
@property
|
||||
def scintillator(self) -> float:
|
||||
return self.__scintillator_pos.get()
|
||||
|
||||
@scintillator.setter
|
||||
def scintillator(self, value: float):
|
||||
self.set_scintillator(value, wait=True)
|
||||
|
||||
def set_scintillator(self, value: float, /, wait: bool = True):
|
||||
self.__scintillator_pos.put(value, wait=wait)
|
||||
|
||||
# Reflector (backlight?)
|
||||
@property
|
||||
def reflector_up(self) -> bool:
|
||||
return self.__back_light_pos.position.upper() == StagePositionEnum.MEASURE.name
|
||||
|
||||
@reflector_up.setter
|
||||
def reflector_up(self, value: StagePositionEnum):
|
||||
self.set_reflector_up(value, wait=True)
|
||||
|
||||
def set_reflector_up(self, value: StagePositionEnum, /, wait: bool = True):
|
||||
self.__back_light_pos.move(value, wait=wait)
|
||||
|
||||
# Beamstop
|
||||
@property
|
||||
def beamstop_stage_up(self) -> bool:
|
||||
return False
|
||||
|
||||
@beamstop_stage_up.setter
|
||||
def beamstop_stage_up(self, value: bool):
|
||||
pass
|
||||
|
||||
@property
|
||||
def beamstop_z(self) -> float:
|
||||
return 35.0
|
||||
|
||||
@beamstop_z.setter
|
||||
def beamstop_z(self, value: float):
|
||||
pass
|
||||
|
||||
# Optics
|
||||
@property
|
||||
def energy_kev(self) -> float:
|
||||
|
||||
@@ -1,739 +0,0 @@
|
||||
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
|
||||
|
||||
# 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
|
||||
subsample: int = 2
|
||||
|
||||
|
||||
@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 = 36.0
|
||||
gain_max: float = 512.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.
|
||||
|
||||
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,
|
||||
get_frame_id: Callable[[], int] | 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.get_frame_id = get_frame_id
|
||||
|
||||
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 _wait_frames(self, frames: int, timeout_s: float = 0.5) -> bool:
|
||||
"""
|
||||
Wait for `frames` new frames (by frame counter), if get_frame_id is available.
|
||||
|
||||
Returns:
|
||||
True if the requested number of frames were observed, False on timeout.
|
||||
"""
|
||||
if self.get_frame_id is None:
|
||||
sleep(0.04 * frames)
|
||||
return True
|
||||
|
||||
start_id = int(self.get_frame_id())
|
||||
deadline = perf_counter() + float(timeout_s)
|
||||
target_id = start_id + int(frames)
|
||||
|
||||
while perf_counter() < deadline:
|
||||
if int(self.get_frame_id()) >= target_id:
|
||||
return True
|
||||
sleep(0.002)
|
||||
|
||||
return False
|
||||
|
||||
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.
|
||||
"""
|
||||
ok = self._wait_frames(frames=1, timeout_s=0.25)
|
||||
if not ok:
|
||||
# If frame IDs aren't advancing reliably, avoid reusing the same buffer.
|
||||
# Keep it small to preserve speed.
|
||||
sleep(min(0.02, max(0.0, float(new_exp_s))))
|
||||
|
||||
t_extra = max(0.0, float(new_exp_s) - (1.0 / float(fps)))
|
||||
if t_extra > 0:
|
||||
sleep(min(t_extra, 0.35))
|
||||
|
||||
if base_settle_s > 0:
|
||||
sleep(float(base_settle_s))
|
||||
|
||||
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 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,
|
||||
deadband_dn: float = 12.0,
|
||||
metering_unreliable_boost: float = 1.4,
|
||||
) -> tuple[float, float, Metrics]:
|
||||
|
||||
base_settle_s = float(settle_s)
|
||||
last_m = None
|
||||
|
||||
# Make EMA more responsive so it doesn't lag by ~10 iterations
|
||||
ema_alpha_p = 0.60
|
||||
ema_pv: float | None = None
|
||||
|
||||
# 1 stable frame is typically enough once the loop is well behaved
|
||||
stable_needed = 1
|
||||
stable_count = 0
|
||||
|
||||
# Use raw pv for the first few iterations to avoid EMA-lag overshoot
|
||||
raw_control_iters = 3
|
||||
|
||||
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_raw = float(_get_percentile_value(m, self.target.percentile))
|
||||
t3 = perf_counter()
|
||||
|
||||
last_m = m
|
||||
|
||||
if ema_pv is None:
|
||||
ema_pv = pv_raw
|
||||
else:
|
||||
ema_pv = ema_alpha_p * pv_raw + (1.0 - ema_alpha_p) * ema_pv
|
||||
|
||||
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_raw:.1f} (ema_p={ema_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)"
|
||||
)
|
||||
|
||||
# Stop condition: use raw pv (not EMA) so we don't "wait out" lag
|
||||
in_clip = (m.clip_frac <= self.target.clip_limit)
|
||||
in_p = (abs(self.target.percentile_target - pv_raw) <= deadband_dn)
|
||||
if in_clip and in_p:
|
||||
stable_count += 1
|
||||
if stable_count >= stable_needed:
|
||||
break
|
||||
else:
|
||||
stable_count = 0
|
||||
|
||||
# 1) clipping protection (unchanged)
|
||||
if m.clip_frac > self.target.clip_limit:
|
||||
r = min(8.0, m.clip_frac / max(1e-12, self.target.clip_limit))
|
||||
adaptive_drop = 0.85 - (r - 1.0) * (0.85 - 0.35) / (8.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)
|
||||
if self._set_exposure_if_changed(new_exp, exp):
|
||||
self._settle_after_change(new_exp, base_settle_s)
|
||||
continue
|
||||
|
||||
# 2) unreliable metering (unchanged)
|
||||
if m.n_metered < self.target.min_metered_pixels:
|
||||
boost = float(max(1.05, min(metering_unreliable_boost, self.limits.max_exposure_scale_up)))
|
||||
new_exp = self._clamp(exp * boost, self.limits.exposure_min_s, self.limits.exposure_max_s)
|
||||
if self._set_exposure_if_changed(new_exp, exp):
|
||||
self._settle_after_change(new_exp, base_settle_s)
|
||||
continue
|
||||
|
||||
# 3) main control: use pv_raw for the first few steps, then EMA
|
||||
pv_for_control = pv_raw if i < raw_control_iters else float(ema_pv)
|
||||
|
||||
err = float(self.target.percentile_target - pv_for_control)
|
||||
if abs(err) <= float(deadband_dn):
|
||||
continue
|
||||
|
||||
pv = max(1.0, float(pv_for_control))
|
||||
ratio = float(self.target.percentile_target) / pv
|
||||
|
||||
k = float(self.limits.exposure_k)
|
||||
scale = ratio ** k
|
||||
scale = max(self.limits.max_exposure_scale_down, min(self.limits.max_exposure_scale_up, scale))
|
||||
|
||||
new_exp = self._clamp(exp * scale, self.limits.exposure_min_s, self.limits.exposure_max_s)
|
||||
if self._set_exposure_if_changed(new_exp, exp):
|
||||
self._settle_after_change(new_exp, base_settle_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)
|
||||
# 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)
|
||||
|
||||
def get_frame_id() -> int:
|
||||
return int(sample_cam.uid.get())
|
||||
|
||||
return AutoExposureController(get_gray, get_exp, set_exp, get_gain, set_gain,
|
||||
metering=met, target=tgt, limits=lim, get_frame_id=get_frame_id)
|
||||
|
||||
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()}-ES-MS:")
|
||||
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.001,
|
||||
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.6)
|
||||
|
||||
metering = MeteringConfig(
|
||||
center_roi=0.7,
|
||||
bg_percentile=10.0,
|
||||
delta_dn=10,
|
||||
winsor_high=230,
|
||||
)
|
||||
targets = TargetConfig(
|
||||
clip_limit=0.02, # allow more clipping (bright field can clip)
|
||||
percentile=70.0,
|
||||
percentile_target=170.0,
|
||||
min_metered_pixels=5000, # can be lower because backlight mask is usually strong
|
||||
)
|
||||
|
||||
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.60, max_exposure_scale_up=1.6,
|
||||
max_exposure_scale_down=0.6)
|
||||
targets = TargetConfig(
|
||||
clip_limit=0.02,
|
||||
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, deadband_dn=10.0, metering_unreliable_boost=1.3)
|
||||
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)
|
||||
@@ -1,6 +1,6 @@
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from aare.daq.workflows import move_bsz, common_2rse, sa2se, sa2rse, sa2xtal_snapshot, dc2xtal_snapshot, xtal_snapshot2dc, xtal_snapshot2sa, dc2rse, se2sa, sa2dc, dc2sa, sa2xrf, sa2dh, dh2sa
|
||||
from aare.daq.workflows import common_2rse, sa2se, sa2rse, sa2xtal_snapshot, dc2xtal_snapshot, xtal_snapshot2dc, xtal_snapshot2sa, dc2rse, se2sa, sa2dc, dc2sa, sa2xrf, sa2dh, dh2sa
|
||||
from aare.daq.config import ABR_POS_MOUNT
|
||||
from aare.common.models import StagePositionEnum
|
||||
from aare.devices.area_detector import AutoEnum
|
||||
@@ -10,9 +10,6 @@ from aare.devices.bec_worker import BeamlineState
|
||||
@pytest.fixture
|
||||
def mock_devs():
|
||||
devs = MagicMock()
|
||||
devs.bsz.position = 0.0
|
||||
devs.beamstop_stage_up = False
|
||||
devs.reflector_up = False
|
||||
devs.bec_worker = MagicMock()
|
||||
devs.bec_worker.planner = MagicMock()
|
||||
devs.bec_worker.move_to = MagicMock()
|
||||
@@ -37,35 +34,8 @@ def _assert_bec_moved(devs, state):
|
||||
), f"BEC was not asked to move to {state}"
|
||||
|
||||
|
||||
def test_move_bsz_no_move(mock_devs):
|
||||
mock_devs.bsz.position = 1.0
|
||||
move_bsz(mock_devs, 1.05)
|
||||
assert mock_devs.beamstop_z.call_count == 0
|
||||
|
||||
|
||||
def test_move_bsz_with_move(mock_devs):
|
||||
mock_devs.bsz.position = 0.0
|
||||
mock_devs.beamstop_stage_up = False
|
||||
mock_devs.reflector_up = True
|
||||
|
||||
move_bsz(mock_devs, 1.0)
|
||||
|
||||
assert mock_devs.reflector_up is True
|
||||
assert mock_devs.beamstop_stage_up is False
|
||||
assert mock_devs.beamstop_z == 1.0
|
||||
|
||||
|
||||
def test_common_2rse_no_bec(mock_devs, mock_cfg):
|
||||
mock_devs.bec_worker = None
|
||||
common_2rse(mock_devs, mock_cfg)
|
||||
|
||||
assert mock_devs.collimator == 20.0
|
||||
assert mock_devs.scintillator == 20.0
|
||||
mock_devs.smargon_move_home.assert_called_once()
|
||||
assert mock_devs.beamstop_stage_up == StagePositionEnum.PARK
|
||||
|
||||
|
||||
def test_common_2rse_with_bec(mock_devs, mock_cfg):
|
||||
def test_common_2rse(mock_devs, mock_cfg):
|
||||
mock_devs.bec_worker = MagicMock()
|
||||
mock_devs.bec_worker.planner = MagicMock()
|
||||
mock_devs.bec_worker.move_to = MagicMock()
|
||||
|
||||
Reference in New Issue
Block a user