feat(flomni): trigger source of the flOMNI loop at every Fermat scan start (soft trigger) #343
No files matched your search
+20
@@ -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)`,
|
||||
|
||||
BIN
Binary file not shown.
@@ -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,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()
|
||||
Reference in new issue
Block a user