diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-realtime_system.md b/csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-realtime_system.md index 165239ba..eb9d1348 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-realtime_system.md +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-realtime_system.md @@ -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)`, diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-realtime_system.pdf b/csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-realtime_system.pdf index cf5b5236..fb4ee1d0 100644 Binary files a/csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-realtime_system.pdf and b/csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-realtime_system.pdf differ diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py index b539dbd9..598d262f 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py @@ -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(""), e.g. help("move_beyond_um_x")') diff --git a/csaxs_bec/devices/omny/flomni_loop_settings.py b/csaxs_bec/devices/omny/flomni_loop_settings.py index dbb32383..23bf5faf 100644 --- a/csaxs_bec/devices/omny/flomni_loop_settings.py +++ b/csaxs_bec/devices/omny/flomni_loop_settings.py @@ -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) diff --git a/csaxs_bec/scans/flomni_fermat_scan.py b/csaxs_bec/scans/flomni_fermat_scan.py index fc1b820c..6cfac1bc 100644 --- a/csaxs_bec/scans/flomni_fermat_scan.py +++ b/csaxs_bec/scans/flomni_fermat_scan.py @@ -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. diff --git a/tests/tests_bec_ipython_client/test_flomni_realtime_system.py b/tests/tests_bec_ipython_client/test_flomni_realtime_system.py index d44bbf96..04921234 100644 --- a/tests/tests_bec_ipython_client/test_flomni_realtime_system.py +++ b/tests/tests_bec_ipython_client/test_flomni_realtime_system.py @@ -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} diff --git a/tests/tests_scans/test_flomni_fermat_scan.py b/tests/tests_scans/test_flomni_fermat_scan.py index 2e2e4c0e..98a66823 100644 --- a/tests/tests_scans/test_flomni_fermat_scan.py +++ b/tests/tests_scans/test_flomni_fermat_scan.py @@ -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()