feat(flomni): trigger source of the flOMNI loop at every Fermat scan start (soft trigger) #343

Merged
holler merged 1 commits from feat/flomni-soft-trigger into main 2026-10-27 08:40:32 +01:00
7 changed files with 191 additions and 6 deletions

No files matched your search

@@ -21,6 +21,7 @@ Background and reference (flOMNI control repository, `OMNY/flOMNI` on gitea):
| FZP scanner (only between scans) | `fzp_enable` (both axes or none), `fzp_range_V_x` / `_y`, `feedforward_enable`, `fzp_ff_priority` (feedforward first, 1 in production and all scan profiles since 2026-10-26), `move_beyond_um_x` / `_y`, `exposure_during_motion`, `abbe_enable`, `abbe_limit_um` |
| Sample stage moves (only between scans) | `target_shaper_scale`, `jump_max_um_x` / `_y` |
| Detector trigger | `trigger_tolerance_nm`, `trigger_hold_ms`, `trigger_min_delay_ms`, `trigger_on_move` |
| Trigger source (only between scans) | `trigger_soft` (never saved, set by the Fermat scan), `soft_exposure_ms`, `soft_latency_ms` |
| Laser tracker and thresholds | `tracking_target_y` / `_z`, `psd_threshold_low`, `interferometer_threshold`, `interferometer_threshold_rot` |
| FZP calibration (expert, only between scans) | `sample_hold_fzp_V`, `sample_hold_err_um`, `sample_hold_zero` |
@@ -59,6 +60,25 @@ sensor) stays within the tolerance in x and in y (each averaged over 0.5 ms) for
The former `flomni.interferometer_trigger_*` were removed 2026-10-23.
### Soft trigger (working without detector)
With the soft trigger no trigger leaves the loop (the detector trigger output stays 0) and the loop emulates the
detector: every trigger starts an exposure of `soft_exposure_ms` after `soft_latency_ms` (1 ms). Scans then run
without detector or trigger PLC. The loop always starts with the hardware trigger; `trigger_soft` is never saved as
default and is in no profile (flOMNI `FeedbackLoop/soft_trigger.md`).
**`flomni_fermat_scan` sets the trigger source at every scan start** (2026-10-26):
- **At the beamline → hardware.** At the beamline means: `ddg1` is configured with a real (not simulated) device
class, or one of the exposure box motors `ebtrx`, `ebtry`, `ebtrz`, `ebsupport` is in the session. If this cannot
be confirmed (web HMI not reachable, refused) the scan is aborted: with the soft trigger the detector would not
be triggered while flOMNI exposes internally. A failing check also counts as beamline.
- **Otherwise (simulated `ddg1`, no beamline devices) → soft** with `soft_exposure_ms` = `exp_time` ×
`frames_per_trigger` × 1000. If that fails the scan continues with a warning (alarm).
By hand: `flomni.rt_settings.trigger_set_soft(True, exposure_ms=100)`, `trigger_set_soft(False)`;
`trigger_show()` shows both groups.
### Laser tracker
`flomni.rt_settings.tracking_show()`, `tracking_set_target_y(v)` / `tracking_set_target_z(v)`,
@@ -85,16 +85,18 @@ class FlomniRealtimeSystem(WebHmiClient):
raise RealtimeSystemError(f"unknown setting '{name}'; known: {', '.join(values)}")
return values[name]
def show(self, group: str = None, expert: bool = False):
"""Print the settings (live value, .xmp value, markers), the violated rules and the scan state."""
def show(self, group=None, expert: bool = False):
"""Print the settings (live value, .xmp value, markers), the violated rules and the scan state; group: one
group name or several (tuple)."""
snap = self.read()
wanted = (group,) if isinstance(group, str) else tuple(group or ())
groups = {g["name"]: g for g in snap["groups"]}
sc = snap.get("scan") or {}
print(STAFF_ONLY)
print(f"Loop settings {snap.get('instrument', '')} ({self.route}) - "
+ (f"scan running (mode {_fmt(sc.get('mode'))})" if sc.get("active") else "no scan"))
for gname, g in groups.items():
if group and gname != group:
if wanted and gname not in wanted:
continue
if g.get("expert") and not expert and not group:
continue
@@ -203,8 +205,9 @@ class FlomniRealtimeSystem(WebHmiClient):
# the former flomni.interferometer_trigger_* (removed 2026-10-23)
def trigger_show(self):
"""Detector trigger (step scans): tolerance, hold time, minimum delay, trigger after every move."""
self.show(group="trigger")
"""Detector trigger (step scans): tolerance, hold time, minimum delay, trigger after every move; trigger source
(hardware / soft)."""
self.show(group=("trigger", "trigger_source"))
def trigger_set_tolerance_nm(self, tolerance_nm: float):
"""|relative position sample vs FZP| x and y within this (nm) for the hold time. 30 coarse, 10 standard, 2 finest;
@@ -220,6 +223,16 @@ class FlomniRealtimeSystem(WebHmiClient):
False: triggers at step-scan points only."""
return self.set(trigger_on_move=1 if enable else 0)
def trigger_set_soft(self, enable: bool, exposure_ms: float = None):
"""Soft trigger (working without detector / trigger PLC): True = no hardware trigger leaves the loop, every
trigger starts an emulated exposure of exposure_ms (default: unchanged); False = hardware trigger. Never saved
as default (the loop starts with the hardware trigger); the flOMNI Fermat scan sets it at every scan start
(hardware at the beamline, soft otherwise). Only between scans."""
changes = {"trigger_soft": 1 if enable else 0}
if enable and exposure_ms is not None:
changes["soft_exposure_ms"] = exposure_ms
return self.set(**changes)
def tracking_targets(self):
"""-> (target y, target z) of the laser tracker on the PSD."""
v = self.get()
@@ -253,7 +266,7 @@ class FlomniRealtimeSystem(WebHmiClient):
if name is None:
print(__doc__.split("\n\nConnection:")[0])
print("\nShortcuts: trigger_show(), trigger_set_tolerance_nm(v), trigger_set_hold_ms(v),"
" trigger_set_after_move(True/False), tracking_targets(), tracking_show(), tracking_set_target_y(v),"
" trigger_set_after_move(True/False), trigger_set_soft(True/False, exposure_ms), tracking_targets(), tracking_show(), tracking_set_target_y(v),"
" tracking_set_target_z(v), tracking_set_interferometer_threshold(v), tracking_set_psd_threshold_low(v);"
" profiles(), unsaved(), get(name)")
print('Explanations: help("all") or help("<setting>"), e.g. help("move_beyond_um_x")')
@@ -44,6 +44,7 @@ SETTINGS = (
"sample_hold_fzp_V", "sample_hold_err_um", "sample_hold_zero",
"target_shaper_scale", "jump_max_um_x", "jump_max_um_y", # group motion, scan-mode profiles (2026-10-26)
"fzp_ff_priority", # FZP feedforward priority (2026-10-26)
"trigger_soft", "soft_exposure_ms", "soft_latency_ms", # soft trigger (2026-10-26)
)
RETRY_AFTER_FAILURE_S = 60.0 # web HMI not reachable: no new attempt for this long (a tomogram does not wait each time)
+69
View File
@@ -69,6 +69,10 @@ class FlomniFermatScan(ScanBase):
MAX_CORRIDOR_SIZE = 3 # Corridor size is capped at this value (in um) for stability
MIN_FERMAT_ASYMMETRY = 0.1 # Practical working range for fermat_asymmetry (soft-clamped)
MAX_FERMAT_ASYMMETRY = 10.0
# Devices that exist only in the beamline configuration (bl_endstation.yaml: exposure box motors); with one of
# them, or a ddg1 that is not simulated, the session is at the beamline -> hardware trigger (flOMNI soft trigger,
# FeedbackLoop/soft_trigger.md, 2026-10-26)
BEAMLINE_DEVICES = ("ebtrx", "ebtry", "ebtrz", "ebsupport")
def __init__(
# fmt: off
@@ -345,6 +349,9 @@ class FlomniFermatScan(ScanBase):
"""
dev = self.dev
# Trigger source of the flOMNI loop: hardware at the beamline, soft otherwise
self.set_trigger_source()
# Prepare DDG1
dev.ddg1.set_trigger(TRIGGERSOURCE.EXT_RISING_EDGE.value)
@@ -377,6 +384,68 @@ class FlomniFermatScan(ScanBase):
"Signal strength of the laser tracker is too low for scanning. Realignment required!"
)
def at_beamline(self) -> tuple[bool, str]:
"""
Is this session at the beamline? True when ddg1 is configured with a real (not simulated) device class or
one of BEAMLINE_DEVICES is configured; False only when both say no. A failing check counts as beamline,
so the hardware trigger is the fallback.
Returns:
(at_beamline, reason)
"""
try:
devices = self.device_manager.devices
reasons = []
if "ddg1" in devices:
cls = str((getattr(devices["ddg1"], "_config", None) or {}).get("deviceClass", ""))
if ".sim." not in cls and "Simulated" not in cls:
reasons.append(f"ddg1 is {cls or 'not simulated'}")
found = [name for name in self.BEAMLINE_DEVICES if name in devices]
if found:
reasons.append("beamline devices " + ", ".join(found))
if reasons:
return True, "; ".join(reasons)
return False, "no beamline devices, ddg1 simulated or absent"
except Exception as exc: # pylint: disable=broad-except
return True, f"beamline check failed ({exc}), treated as beamline"
def set_trigger_source(self):
"""
Set the trigger source of the flOMNI loop (loop setting trigger_soft, web HMI via flomni_loop_settings):
at the beamline the hardware trigger (the loop's default); otherwise the soft trigger with the exposure
exp_time x frames_per_trigger, so a scan runs without detector or trigger PLC. At the beamline the scan is
aborted if the hardware trigger cannot be confirmed (with the soft trigger the detector would not be
triggered while flOMNI exposes internally).
"""
at_bl, reason = self.at_beamline()
if at_bl:
changes = {"trigger_soft": 0}
else:
changes = {"trigger_soft": 1,
"soft_exposure_ms": float(self.exp_time) * float(self.frames_per_trigger) * 1000.0}
try:
res = self.dev.flomni_loop_settings.api_post(
"/api/settings/apply", {"changes": changes, "profile": None, "adjust": False}
)
errors = [] if res and res.get("ok") else list((res or {}).get("errors") or ["no answer"])
except Exception as exc: # pylint: disable=broad-except
errors = [str(exc)]
source = "hardware" if at_bl else "soft"
if errors:
text = f"flOMNI trigger source {source} ({reason}) not set: {'; '.join(errors)}"
if at_bl:
raise ScanAbortion(text + ". At the beamline the hardware trigger must be confirmed.")
logger.warning(text)
self.device_manager.connector.raise_alarm(
severity=Alarms.WARNING,
info=messages.ErrorInfo(
error_message=text, compact_error_message=text, exception_type="TriggerSource",
device="flomni_loop_settings",
),
)
return
logger.info(f"flOMNI trigger source: {source} ({reason}), {changes}")
def flomni_rotation(self, angle: float):
"""
Rotate flomni to the requested angle.
@@ -212,3 +212,12 @@ def test_help(server, capsys):
assert "BEAMLINE SCIENTISTS" in capsys.readouterr().out
with pytest.raises(RealtimeSystemError, match="unknown setting"):
rts.help("nosuch")
def test_trigger_set_soft(server):
url, tok, _ = server
rts = FlomniRealtimeSystem(url=url, token_file=tok)
rts.trigger_set_soft(True, exposure_ms=200)
assert _Handler.requests[-1][1]["changes"] == {"trigger_soft": 1, "soft_exposure_ms": 200}
rts.trigger_set_soft(False, exposure_ms=200)
assert _Handler.requests[-1][1]["changes"] == {"trigger_soft": 0}
@@ -1,6 +1,8 @@
from unittest import mock
import numpy as np
import pytest
from bec_server.scan_server.scans import ScanAbortion
from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan
@@ -152,3 +154,74 @@ def test_randomize_offset_false_gives_unshifted_deterministic_lattice():
phi = 2 * np.pi * ((1 + np.sqrt(5)) / 2.0)
r = 0.57
np.testing.assert_allclose(a[0, :2], [r * np.sin(phi), r * np.cos(phi)])
# ── trigger source of the flOMNI loop (soft trigger, 2026-10-26) ──────────────
def _trigger_stub(devices, apply_result=None, apply_exc=None, exp_time=0.1, frames_per_trigger=2):
"""A stand-in for a FlomniFermatScan instance with a device list {name: deviceClass} and the loop-settings
device answering /api/settings/apply."""
scan = mock.MagicMock()
scan.BEAMLINE_DEVICES = FlomniFermatScan.BEAMLINE_DEVICES
scan.device_manager.devices = {
name: mock.MagicMock(_config={"deviceClass": cls}) for name, cls in devices.items()
}
scan.exp_time = exp_time
scan.frames_per_trigger = frames_per_trigger
if apply_exc is not None:
scan.dev.flomni_loop_settings.api_post.side_effect = apply_exc
else:
scan.dev.flomni_loop_settings.api_post.return_value = (
apply_result if apply_result is not None else {"ok": True, "errors": []}
)
scan.at_beamline = lambda: FlomniFermatScan.at_beamline(scan)
return scan
SIM_DDG1 = "csaxs_bec.devices.sim.simulated_beamline_devices.SimulatedDDG1"
REAL_DDG1 = "csaxs_bec.devices.epics.delay_generator_csaxs.DDG1"
MOTOR = "ophyd_devices.devices.psi_motor.EpicsUserMotorVME"
def test_at_beamline_detection():
assert FlomniFermatScan.at_beamline(_trigger_stub({"ddg1": SIM_DDG1}))[0] is False
assert FlomniFermatScan.at_beamline(_trigger_stub({}))[0] is False
assert FlomniFermatScan.at_beamline(_trigger_stub({"ddg1": REAL_DDG1}))[0] is True
# a beamline-only device counts even with a simulated ddg1
at_bl, reason = FlomniFermatScan.at_beamline(_trigger_stub({"ddg1": SIM_DDG1, "ebtrx": MOTOR}))
assert at_bl is True and "ebtrx" in reason
# a failing check counts as beamline (hardware trigger is the fallback)
scan = _trigger_stub({})
scan.device_manager.devices = None
assert FlomniFermatScan.at_beamline(scan)[0] is True
def test_trigger_source_soft_away_from_beamline():
scan = _trigger_stub({"ddg1": SIM_DDG1}, exp_time=0.1, frames_per_trigger=2)
FlomniFermatScan.set_trigger_source(scan)
path, body = scan.dev.flomni_loop_settings.api_post.call_args[0]
assert path == "/api/settings/apply"
assert body["changes"]["trigger_soft"] == 1
assert abs(body["changes"]["soft_exposure_ms"] - 200.0) < 1e-9
def test_trigger_source_hardware_at_beamline():
scan = _trigger_stub({"ddg1": REAL_DDG1, "ebtry": MOTOR})
FlomniFermatScan.set_trigger_source(scan)
body = scan.dev.flomni_loop_settings.api_post.call_args[0][1]
assert body["changes"] == {"trigger_soft": 0}
def test_trigger_source_not_confirmed_at_beamline_aborts():
scan = _trigger_stub({"ddg1": REAL_DDG1}, apply_exc=RuntimeError("web HMI not reachable"))
with pytest.raises(ScanAbortion, match="hardware trigger must be confirmed"):
FlomniFermatScan.set_trigger_source(scan)
scan = _trigger_stub({"ddg1": REAL_DDG1}, apply_result={"ok": False, "errors": ["a scan is running"]})
with pytest.raises(ScanAbortion, match="a scan is running"):
FlomniFermatScan.set_trigger_source(scan)
def test_trigger_source_failure_away_only_warns():
scan = _trigger_stub({"ddg1": SIM_DDG1}, apply_exc=RuntimeError("web HMI not reachable"))
FlomniFermatScan.set_trigger_source(scan)
scan.device_manager.connector.raise_alarm.assert_called_once()