diff --git a/src/aare/daq/devices.py b/src/aare/daq/devices.py index ceffcf07..02febc74 100644 --- a/src/aare/daq/devices.py +++ b/src/aare/daq/devices.py @@ -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: diff --git a/src/aare/gui/sam_cam_test.py b/src/aare/gui/sam_cam_test.py deleted file mode 100644 index abe00a96..00000000 --- a/src/aare/gui/sam_cam_test.py +++ /dev/null @@ -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) \ No newline at end of file diff --git a/tests/unit/daq/test_workflows.py b/tests/unit/daq/test_workflows.py index 3a466b63..38543aec 100644 --- a/tests/unit/daq/test_workflows.py +++ b/tests/unit/daq/test_workflows.py @@ -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()