csaxs_bec/bec_ipython_client/plugins/OMNY_shared/omny_general_tools.py

@@ -35,6 +35,26 @@ def umvr(*args):
     return scans.umv(*args, relative=True)
 
 
+def read_escape_suffix(stream, timeout: float = 0.05) -> str | None:
+    """The two characters after an ESC on a non-blocking stdin (arrow keys: "[A".."[D"),
+    or None if they do not arrive within `timeout` (ESC alone, or a slow terminal).
+
+    A plain stream.read(2) can return fewer characters, or None, before the rest of the
+    sequence has arrived; unpacking that raised and ended the tweak with a traceback."""
+    chars = ""
+    deadline = time.monotonic() + timeout
+    while len(chars) < 2 and time.monotonic() < deadline:
+        try:
+            got = stream.read(2 - len(chars))
+        except (IOError, TypeError):
+            got = None
+        if got:
+            chars += got
+        else:
+            time.sleep(0.005)
+    return chars if len(chars) == 2 else None
+
+
 class OMNYTools:
 
     HEADER = "\033[95m"
@@ -164,10 +184,10 @@ class OMNYTools:
         self, dev1, step1: float, dev2="none", step2: float = "0", special_command="none"
     ):
         if dev1 not in dev.enabled_devices:
-            print(f"Device 1 {dev} is not in enabled devices.")
+            print(f"Device 1 {dev1} is not in enabled devices.")
             return
         if dev2 not in dev.enabled_devices and dev2 != "none":
-            print(f"Device 2 {dev} is not in enabled devices.")
+            print(f"Device 2 {dev2} is not in enabled devices.")
             return
         # Save the current terminal settings
         fd = sys.stdin.fileno()
@@ -187,7 +207,8 @@ class OMNYTools:
                         print("\n\rExiting tweak mode\r")
                         break
                     elif key == "\x1b":  # Escape sequences for arrow keys
-                        next1, next2 = sys.stdin.read(2)
+                        suffix = read_escape_suffix(sys.stdin)
+                        next1, next2 = suffix if suffix else ("", "")
                         if next1 == "[":
                             if next2 == "A":
                                 if dev2 != "none":

csaxs_bec/bec_ipython_client/plugins/OMNY_shared/tomo_queue_mixin.py

@@ -1157,6 +1157,8 @@ class TomoQueueMixin:
         self.progress["estimated_finish_time"] = None
         self.progress["accumulated_idle_time"] = 0.0
         self.progress["heartbeat"] = None
+        # flOMNI: a completed scan is not resumed (tomo_scan_resume); reopened here on purpose
+        self.progress["tomo_completed"] = False
 
         job_id = job["id"]
         self._tomo_queue_proxy.update_by_id(job_id, status="incomplete")

csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-howto_calibrate_the_fzp_scanner.md

@@ -2,7 +2,7 @@
 
 Source of `local_contact_only-howto_calibrate_the_fzp_scanner.pdf`. The original text is kept unchanged; the
 section "Update 2026-10-20" at the end lists what the present flOMNI loop (PREEMPT_RT, 100 µs, branch `rt_preemt`)
-needs in addition.
+needs in addition (corrected 2026-10-23: "move-ahead" → **move-beyond**, see the section at the end).
 
 ```text
 This is a description of how to calibrate the FZP scanner of flOMNI
@@ -72,6 +72,11 @@ Default values for the fzp pid, fast FZP parameter are "1,0.0001,0.1,-
 4,0,0,0"
 ```
 
+Naming: "move ahead" in the text above (sample PIDs `FAST_FZP_PID_LOOP(0)` < 0) is called **move-beyond** from
+2026-10-23 on: the sample stage moves *beyond* the scan point towards the next one, the FZP compensates and so sits
+beyond its centre during the exposure, on the side that gives it more stroke for the next step. It is not a
+"pre-move" of the FZP to the next point (that is `TargetSelect` `use_final` 1, see below).
+
 ## Update 2026-10-20 (flOMNI loop on PREEMPT_RT, mpc2844)
 
 The loop has new modules since the text above was written (flOMNI `FeedbackLoop/tuning_guide.md`,
@@ -79,8 +84,8 @@ The loop has new modules since the text above was written (flOMNI `FeedbackLoop/
 Parameters) instead of OrchestraHMI. `i` = x or y, as above. Module names now: `PID_x`, `PID_y` (sample stage),
 `PID_x_fzp`, `PID_y_fzp` (FZP scanner), `AnalogInput_6289` (was `AnalogInput_6259`).
 
-**FAST_FZP_PID_LOOP has 7 elements** (`PID.cpp`). Sample stage PIDs: (0) move-ahead (negative, µm, point-by-point
-scans only), (4) FZP error threshold (µm) and (5) FZP drive limit (V) for holding the sample stage while the FZP
+**FAST_FZP_PID_LOOP has 7 elements** (`PID.cpp`). Sample stage PIDs: (0) move-beyond (negative, µm, point-by-point
+scans only; section "Move-beyond" below), (4) FZP error threshold (µm) and (5) FZP drive limit (V) for holding the sample stage while the FZP
 takes the step (5 = 0: never held), (6) 1 = hold the sample stage at 0. FZP PIDs: (0) mode (1 operation, 2 FZP
 centred on its cap sensor = disabled, 3 not commissioned, 4 debug), (1)–(3) min / max / gain of the old velocity
 correction (−4 before 2026-10-19, now 0: replaced by the feedforward).
@@ -97,14 +102,15 @@ correction (−4 before 2026-10-19, now 0: replaced by the feedforward).
 - FZP range: `PID_x_fzp` / `PID_y_fzp` `Ub` / `Lb` are ±1.5 V in operation (the cap sensors are calibrated over
   about ±2 V only). For the field of view above (2.8 × 1.8 µm) widen them, e.g. to ±10 V (FZP ≈ 0.13 µm/V in x,
   ≈ 0.05 µm/V in y). The feedforward stays off (`enable` 0,0) during this scan.
-- The detector trigger (`OnTargetTrigger`) checks the relative error sample vs FZP (tolerance / hold time from the
-  ConstEmitter, BEC `interferometer_trigger_*`, default 10 nm / 3 ms) and waits `MinDelay_ms` (15 ms).
+- The detector trigger (`OnTargetTrigger`) checks the relative error sample vs FZP (parameters
+  `ontarget_tolerance_um` / `ontarget_hold_ms`, BEC `flomni.realtime_system.trigger_*`, default 10 nm / 3 ms; until
+  2026-10-23 ConstEmitter lines) and waits `MinDelay_ms` (15 ms).
 
 **Step 3, default operating values (replace the values in the text above):**
 
 | Module | Parameter | Value |
 |---|---|---|
-| `PID_x`, `PID_y` | `FAST_FZP_PID_LOOP` | `0,0,0,0,0.1,0,0` (no move-ahead: −0.5 µm does not fit the ±1.5 V FZP range) |
+| `PID_x`, `PID_y` | `FAST_FZP_PID_LOOP` | `0,0,0,0,0.1,0,0` (no move-beyond: −0.5 µm is larger than the ±1.5 V FZP stroke, x ≈ ±0.19 µm, y ≈ ±0.08 µm) |
 | `PID_x_fzp` | `FAST_FZP_PID_LOOP`, `Ub` / `Lb` | `1,0.0001,0.1,0,0,0,0`, ±1.5 V |
 | `PID_y_fzp` | `FAST_FZP_PID_LOOP`, `Ub` / `Lb` | `1,0.0005,0.02,0,0,0,0`, ±1.5 V |
 | `Feedforward` | `enable`; `ff_limit`, `out_min` / `out_max` | 1,1; ±1.5 V (as the FZP `Ub` / `Lb`) |
@@ -122,7 +128,77 @@ over a wider range the FZP can take the small steps alone (flOMNI `feedback_repo
 - `PID_x_fzp` / `PID_y_fzp` `Ub` / `Lb` and the `Feedforward` limits to the calibrated range (e.g. ±7 V);
 - `TargetSelect_x` / `_y` `use_final` 1 (the FZP aims at the new point at once, the sample stage follows its
   S-curve) and `OnTargetTrigger` `WaitForSlew` 0 (trigger on the relative error only); optionally `MinDelay_ms` 0;
-- optionally move-ahead, sized to the FZP range: `PID_x` (0) e.g. −0.5, `PID_y` (0) e.g. −0.2.
+- optionally move-beyond, sized to the FZP stroke (≈ half of it, see below): at ±7 V (x ≈ ±0.9 µm, y ≈ ±0.36 µm)
+  `PID_x` (0) e.g. −0.5, `PID_y` (0) e.g. −0.2 (the values tested in `feedback_report.md` 6q).
 
 Measured with 101 ms exposures: 0.2–1 µm steps trigger after 4.5–5.8 ms instead of 17–32 ms; the error during the
 exposure grows with the step (1 µm: ≈ 6 nm max instead of 2.5 nm), so the setting depends on the experiment.
+
+## With the loop settings (since 2026-10-23)
+
+The steps above are profiles of the loop settings (web HMI tab *Loop settings*, BEC `flomni.realtime_system`;
+flOMNI `FeedbackLoop/loopsettings/`). The rules keep the modules consistent, so the individual parameters above do
+not have to be set by hand:
+
+| Step | Profile | BEC |
+|---|---|---|
+| 1, FZP disabled (alignment, affine matrix) | `fzp_off` | `flomni.realtime_system.apply_profile("fzp_off")` |
+| 2, FZP-only calibration scan (clear `MatrixCorr` first: not part of the settings) | `fzp_calibration` | `apply_profile("fzp_calibration")` |
+| 3, back to operation | `production` | `apply_profile("production")` |
+| afterwards, fast scans with the FZP at ±9 V (scan modes, flOMNI `FeedbackLoop/scan_modes.md`) | `high_res`, `normal`, `low_res` | `apply_profile("high_res")` |
+
+Single values: `flomni.realtime_system.set(fzp_range_V_x=7, fzp_range_V_y=7)`,
+`set(move_beyond_um_x=0.3)`, `set(exposure_during_motion=1)`; refused with the reason if a rule objects
+(`adjust=True` changes the dependent settings too). FZP settings only between scans. Live only until
+`save_as_default()`.
+
+## Move-beyond (`PID_x` / `PID_y` `FAST_FZP_PID_LOOP(0)` < 0), 2026-10-23
+
+**What it does** (`PID.cpp`, unchanged since the original loop). Only in point-by-point scans (scan state 2). The
+TrajectoryGenerator gives the next scan point (outputs 5 / 6, PID input 9). The sample PID then aims not at the
+target T but at T + m · sign(next − T), m = min(|FAST_FZP_PID_LOOP(0)|, |next − T|): the sample stage moves *beyond*
+the point towards the next one, by at most the step. The FZP loop keeps its own target T (it does not see m), so the
+FZP compensates the offset: during the exposure it holds the relative position at T while sitting at +m off its
+centre. At the next point the FZP then has stroke R + m in the step direction instead of R (R = stroke from the centre),
+so larger steps are taken by the FZP alone. Steps smaller than |FAST_FZP_PID_LOOP(0)|: the sample stage is one point
+ahead, the FZP moves between 0 and +step only. The offset is not shaped (InputSlew acts on T only); the sample stage
+moves to it with its own (slow) loop, i.e. partly during the exposure, and the FZP compensates that motion.
+
+**Is it still the same in the new loop?** Yes. The relative position is computed as before:
+FZP error = Cap − (Chan − target) (mode 1), `Position sample vs FZP` = Chan − Cap (PID output 3, RT_sampler). The
+sample PIDs still get the shaped target (InputSlew) + move-beyond and the raw interferometer. The FZP side now has
+two additions that act on the FZP side only: `TargetSelect` (`use_final` 0 = shaped target as before, 1 = final
+target) and `AbbeCorrection` (Chan corrected for the stage tilt, see below). The `Feedforward` uses the same FZP
+target and Chan, so it supplies the +m offset of the FZP directly (static term m / K).
+
+**Sizing.** During the exposure the FZP has to hold m + Abbe correction + noise / settling inside its stroke:
+|m| ≤ R − reserve, as a rule ≈ R / 2. R per axis = `Ub` × K (K ≈ 0.13 µm/V x, ≈ 0.05 µm/V y). The loop settings
+refuse more than 0.6 × R minus the Abbe limit (rule `move_beyond_*_fits`, constant `move_beyond_fraction`; the
+tested ±7 V values are 0.56 × R):
+
+| FZP `Ub`/`Lb` | R x / y | move-beyond x / y (≈ R/2) |
+|---|---|---|
+| ±1.5 V (production) | ±0.19 / ±0.08 µm | ≈ −0.1 / ≈ −0.04 µm (production: 0) |
+| ±7 V (after calibration) | ±0.9 / ±0.36 µm | −0.5 / −0.2 µm (tested, 6q) |
+
+The original default −0.5 µm in both axes dates from the ±9 V limits of the old loop (x ≈ ±1.2 µm, about R/2; in y
+±9 V ≈ ±0.45 µm, i.e. about the full stroke). When the FZP saturates while holding the offset, the relative error grows and the trigger
+waits.
+
+**Abbe correction.** `AbbeCorrection` adds a new term to the measured sample x / y on the FZP side only:
+x = Chan2_x + θz · offset_y_for_x, y = Chan1_y + θz · offset_x_for_y (θz = Chan3 / 13 mm). So yes: it is an error of
+the sample x / y position (stage tilt about z, not seen by the x / y interferometer beams) that is corrected by the
+FZP x / y, not by the sample stage (`PID_x` / `PID_y` keep the raw Chan, the rotZ piezo is too slow). The positions
+recorded for the reconstruction (`Position sample vs FZP`) contain it as well. It uses FZP stroke (limited to
+`correction_limit_um` 0.1 µm per axis; y at rest ≈ 2 nm, ≈ 30–37 nm during 20 µm x steps) — keep this in the reserve
+of the move-beyond, in y it is a large part of the ±0.08 µm stroke. Keep it off (`enable` 0) during the FZP
+calibration scan (step 2).
+
+**Notes.**
+- Hold of the sample stage, (4) / (5): input 8 of `PID_x` / `PID_y` was the FZP PID output *before* the
+  `Feedforward` (with the feedforward on only the remainder, not the real FZP drive). Since 2026-10-23 it is the
+  `Feedforward` output (FZP drive incl. feedforward), so (5) again compares the real FZP drive. No change with the
+  feedforward off or with (5) = 0 (production).
+- On the last point of a scan the next point output is 0, so the move-beyond points towards 0 (as in the old loop).
+- `TargetSelect` `use_final` 1 is a different thing: the FZP aims at the new point at once while the sample stage
+  follows the S-curve. Both can be combined (6q, E0).

csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-howto_limit_the_fzp_scanner_range.md

@@ -2,7 +2,8 @@
 
 Source of `local_contact_only-howto_limit_the_fzp_scanner_range.pdf`. The original text is kept unchanged; the
 section "Update 2026-10-20" lists what the present flOMNI loop (PREEMPT_RT, 100 µs, branch `rt_preemt`) needs in
-addition.
+addition. "ahead movement" (element 0) is called **move-beyond** from 2026-10-23 on: the sample stage moves beyond the
+scan point towards the next one, the FZP compensates and sits off-centre, which gives it stroke for the next step.
 
 ```text
 to limit the fzp scanner ranges
@@ -20,7 +21,7 @@ limit the analog output LB and UB
 The limited range is the present operating state: `PID_x_fzp` / `PID_y_fzp` `Ub` / `Lb` = ±1.5 V because the cap
 sensors are calibrated over about ±2 V only (flOMNI `feedback_report.md` 6n). FZP ≈ 0.13 µm/V in x, ≈ 0.05 µm/V in y,
 so ±1.5 V ≈ ±0.19 µm (x) / ±0.08 µm (y). In the sample stage PIDs (`PID_x`, `PID_y`) `FAST_FZP_PID_LOOP` is
-`0,0,0,0,0.1,0,0`: no move-ahead, (5) = 0 (the sample stage always compensates fully).
+`0,0,0,0,0.1,0,0`: no move-beyond, (5) = 0 (the sample stage always compensates fully).
 
 Changing the FZP range now involves, in addition to the text above:
 
@@ -31,6 +32,15 @@ Changing the FZP range now involves, in addition to the text above:
 - **`OnTargetTrigger`** `WaitForSlew` and **`TargetSelect_x` / `_y`** `use_final`: keep the defaults (1 / 0) with a
   small range. `use_final` 1 + `WaitForSlew` 0 ("exposure during the motion") only makes sense when the FZP range
   covers the scan step; steps larger than the range then still wait for the sample stage.
-- Move-ahead (0) must fit into the range: at ±1.5 V at most ≈ −0.05 µm (x) — so in operation it stays 0.
+- Move-beyond (0) must fit into the stroke: the FZP holds it during the exposure, together with the Abbe correction
+  (up to `correction_limit_um` 0.1 µm when enabled) and noise. Rule ≈ half the stroke (the loop settings refuse more than 0.6 × stroke minus the Abbe limit): at ±1.5 V ≈ −0.1 µm (x),
+  ≈ −0.04 µm (y); at ±7 V −0.5 / −0.2 µm (tested). Production: 0. Details: calibration how-to, section "Move-beyond".
+- (5) compares the real FZP drive incl. the feedforward (input 8 of `PID_x` / `PID_y` = `Feedforward` output since
+  2026-10-23; before it was the FZP PID output without the feedforward).
 - After a change: web HMI tab **Loop check** level 1 (expected settings) and, with the beamline ready, the scan test
   (level 4: positioning time, stability during the exposure per step size).
+
+**With the loop settings (since 2026-10-23):** `flomni.realtime_system.set(fzp_range_V_x=..., fzp_range_V_y=...)`
+(web HMI tab *Loop settings*) sets `PID_*_fzp` `Ub` / `Lb` and the `Feedforward` limits together; a move-beyond that
+no longer fits is refused, or clamped with `adjust=True` (rule `move_beyond_*_fits`). `FreqGen` limits stay
+manual (measurements only).

csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-realtime_system.md

@@ -0,0 +1,81 @@
+# flOMNI real-time loop settings (beamline scientists / local contact only)
+
+**Not for users.** `flomni.realtime_system` changes how the FZP scanner, the detector trigger and the laser tracker
+work, and `save_as_default()` makes a change the default of every following loop start. A wrong combination can spoil
+the data of a beamtime. The user guide only says that these settings exist, are set by the local contact and are
+stored with every scan; the details are here.
+
+Background and reference (flOMNI control repository, `OMNY/flOMNI` on gitea):
+
+- `FeedbackLoop/loopsettings/README.md` — every setting with the loop parameters it sets and its production value,
+  the rules, the profiles; `FeedbackLoop/loopsettings/flomni.json` — the definition (help texts, rules, profiles)
+- `FeedbackLoop/tuning_guide.md`, `FeedbackLoop/feedback_report.md` — characterization, measurements, why the values
+- RT-system `docs/loop-settings.md` — design, the web HMI side, installation
+- the FZP how-tos next to this file (`local_contact_only-howto_calibrate_the_fzp_scanner.md`,
+  `local_contact_only-howto_limit_the_fzp_scanner_range.md`) — move-beyond, FZP range, calibration steps as profiles
+
+## What can be set
+
+| Group | Settings |
+|---|---|
+| FZP scanner (only between scans) | `fzp_enable` (both axes or none), `fzp_range_V_x` / `_y`, `feedforward_enable`, `move_beyond_um_x` / `_y`, `exposure_during_motion`, `abbe_enable`, `abbe_limit_um` |
+| Detector trigger | `trigger_tolerance_nm`, `trigger_hold_ms`, `trigger_min_delay_ms`, `trigger_on_move` |
+| 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` |
+
+`flomni.realtime_system.help("all")` explains every setting, `help("<setting>")` one setting (what it does, range,
+live and saved value, the loop parameters it sets, its rules).
+
+## Commands
+
+- `flomni.realtime_system.show()` — all settings with live and saved value, violated rules, scan state
+  (`show(expert=True)` also the FZP calibration group)
+- `flomni.realtime_system.set(name=value, ...)` — change; refused with the reason if a rule objects (e.g. the
+  feedforward must be off while the FZP loop is off); `set(..., adjust=True)` changes the dependent settings the
+  rule names (never the ones given)
+- `flomni.realtime_system.check(name=value)` — dry run
+- `flomni.realtime_system.profiles()`, `apply_profile("production" | "fzp_off" | "fzp_calibration" | "high_res" | "high_res_1v5" | "normal" | "low_res")` (scan modes 2026-10-26: flOMNI `FeedbackLoop/scan_modes.md`)
+- `flomni.realtime_system.unsaved()` — what differs from the loop's parameter files
+- `flomni.realtime_system.save_as_default()` — writes the live values into the loop's parameter files: the next
+  loop start uses them. Only for a change that should stay (a backup of each file is kept on the loop machine).
+- `flomni.realtime_system.help()`, `help("all")`, `help("<setting>")`
+
+Changes are live only until the next start of the control loop, unless saved. FZP settings can only be changed
+between scans (refused while a scan runs); trigger and tracker settings any time.
+
+### Detector trigger
+
+In step scans the detector is triggered when the relative position sample vs FZP (interferometer minus FZP cap
+sensor) stays within the tolerance in x and in y (each averaged over 0.5 ms) for the hold time.
+
+- `flomni.realtime_system.trigger_show()` — tolerance, hold time, minimum delay, trigger after a move
+- `flomni.realtime_system.trigger_set_tolerance_nm(10)` — about 30 for a fast low-resolution scan, 10 standard,
+  2 for the highest resolution (a few ms more per point); 0 = the former criterion (on-target flags of the FZP loops)
+- `flomni.realtime_system.trigger_set_hold_ms(3)` — hold time in ms
+- `flomni.realtime_system.trigger_set_after_move(True)` — also one trigger after every position change of the
+  sample stage (e.g. `umv(dev.rtx, ...)`, `dev.rtx.controller.move_xyz(x, y, z)`) once on target; `False` (default):
+  scan points only
+
+The former `flomni.interferometer_trigger_*` were removed 2026-10-23.
+
+### Laser tracker
+
+`flomni.laser_parameters_show_all()`, `laser_parameters_set_targety(v)` / `_targetz(v)`,
+`laser_parameters_set_intensity_threshold_laser(v)`, `laser_parameters_set_psd_intensity_threshold_tracking_low(v)`
+and the arrow-key tweak `laser_parameters_tweak_tracking()` use the same settings (`tracking_target_y` / `_z`,
+`interferometer_threshold`, `psd_threshold_low`).
+
+## Stored with every scan
+
+The baseline device `flomni_loop_settings` is read at the start of every scan: one value per setting,
+`available` (0 when the loop machine could not be reached — the scan continues), the number of settings not saved
+as default or inconsistent, violated error rules, the definition with its sha1, and a JSON of all settings.
+`dev.flomni_loop_settings.show()` prints what a scan would store now.
+
+## Connection
+
+`flomni.realtime_system` sends its requests through the device `flomni_loop_settings` (device server). The device
+talks to the web HMI on the loop machine (`http://mpc2844:8080`, REST `/api/settings`; port 80 as fallback) with the
+API token in its device config (`ptycho_flomni.yaml`, `token:`; the line of `~/loop-data/webhmi/api_token` on
+mpc2844). Without that device (e.g. a simulated config) the client talks to the web HMI directly with
+`~/.flomni_webhmi_token`. The web HMI tab *Loop settings* shows and changes the same settings with the same rules.

csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py

@@ -17,6 +17,7 @@ from typeguard import typechecked
 
 from csaxs_bec.bec_ipython_client.plugins.cSAXS import cSAXSBeamlineChecks
 from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_optics_mixin import FlomniOpticsMixin
+from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_realtime_system import FlomniRealtimeSystem
 from csaxs_bec.bec_ipython_client.plugins.flomni.gui_tools import flomniGuiTools
 from csaxs_bec.bec_ipython_client.plugins.flomni.x_ray_eye_align import XrayEyeAlign
 from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.angle_file import (
@@ -39,6 +40,7 @@ from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.omny_general_tools import
     OMNYTools,
     PtychoReconstructor,
     TomoIDManager,
+    read_escape_suffix,
 )
 from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.ptouch_printer import PTouchLabelPrinter
 from csaxs_bec.devices.omny.galil.galil_ophyd import GalilError
@@ -81,7 +83,8 @@ class FlomniInitStagesMixin:
         sensor_voltage_target = dev.ftransy.user_parameter.get("sensor_voltage")
         sensor_voltage = float(dev.ftransy.controller.socket_put_and_receive("MG@AN[1]").strip())
 
-        if not np.isclose(sensor_voltage, sensor_voltage_target, 0.25):
+        # absolute tolerances (V, mm): the third positional argument of np.isclose is rtol
+        if not np.isclose(sensor_voltage, sensor_voltage_target, rtol=0, atol=0.25):
             print(
                 f"Sensor voltage of the gripper is {sensor_voltage}, while target from config is {sensor_voltage_target}"
             )
@@ -409,7 +412,7 @@ class FlomniSampleTransferMixin:
         self.ensure_osa_back()
 
         fsamx_in = dev.fsamx.user_parameter.get("in")
-        if not np.isclose(dev.fsamx.readback.get(), fsamx_in, 0.2):
+        if not np.isclose(dev.fsamx.readback.get(), fsamx_in, rtol=0, atol=0.2):
             raise FlomniError("fsamx not in position. Aborting.")
 
         fheater_in = dev.fheater.user_parameter.get("in")
@@ -503,26 +506,28 @@ class FlomniSampleTransferMixin:
     def laser_tracker_off(self):
         dev.rtx.controller.laser_tracker_off()
 
+    # laser tracker targets and thresholds: settings of the real-time loop (flomni.realtime_system, group
+    # "tracking"; until 2026-10-23 ConstEmitter lines set through the CommunicationServer e-commands)
     def laser_parameters_show_all(self):
-        dev.rtx.controller.emitter_show_all()
+        self.realtime_system.show(group="tracking")
 
     def laser_parameters_set_targety(self, val: float):
-        dev.rtx.controller.emitter_set_tracking_target_y(val)
+        self.realtime_system.set(tracking_target_y=val)
 
     def laser_parameters_set_targetz(self, val: float):
-        dev.rtx.controller.emitter_set_tracking_target_z(val)
+        self.realtime_system.set(tracking_target_z=val)
 
     def laser_parameters_set_intensity_threshold_laser(self, val: float):
-        dev.rtx.controller.emitter_set_intensity_threshold_laser(val)
+        self.realtime_system.set(interferometer_threshold=val)
 
     def laser_parameters_set_psd_intensity_threshold_tracking_low(self, val: float):
-        dev.rtx.controller.emitter_set_psd_intensity_threshold_tracking_low(val)
+        self.realtime_system.set(psd_threshold_low=val)
 
     def laser_parameters_tweak_tracking(self):
         """Interactive tweak of the two laser tracking target positions.
 
-        Arrow keys nudge the ConstEmitter tracking targets by a fixed step of
-        0.01 (absolute set of current +/- step):
+        Arrow keys nudge the tracking targets (realtime_system tracking_target_y / _z) by a fixed step of
+        0.02 (absolute set of current +/- step):
             left / right : target y  -/+
             up   / down  : target z  +/-
         Keys:
@@ -542,6 +547,7 @@ class FlomniSampleTransferMixin:
         import tty
 
         controller = dev.rtx.controller
+        rts = self.realtime_system
 
         if not controller.laser_tracker_check_enabled():
             print(
@@ -554,11 +560,7 @@ class FlomniSampleTransferMixin:
         step = 0.02
 
         def read_targets():
-            values = controller.emitter_get()
-            return (
-                values["targetpos_tracking_y"],
-                values["targetpos_tracking_z"],
-            )
+            return rts.tracking_targets()
 
         def signal_fzpx():
             # second channel (index 1) of show_signal_strength_interferometer,
@@ -597,25 +599,25 @@ class FlomniSampleTransferMixin:
                         print("\n\rExiting tweak mode\r")
                         break
                     elif key == "s":
-                        controller.emitter_set_tracking_target_y(start_y)
-                        controller.emitter_set_tracking_target_z(start_z)
+                        rts.set(tracking_target_y=start_y, tracking_target_z=start_z, quiet=True)
                         print("\rReturned to start positions\r")
                         print_status()
                     elif key == "\x1b":  # escape sequences for arrow keys
-                        next1, next2 = sys.stdin.read(2)
+                        suffix = read_escape_suffix(sys.stdin)
+                        next1, next2 = suffix if suffix else ("", "")
                         if next1 == "[":
                             target_y, target_z = read_targets()
                             if next2 == "A":  # up: target z +
-                                controller.emitter_set_tracking_target_z(target_z + step)
+                                rts.set(tracking_target_z=target_z + step, quiet=True)
                                 print_status()
                             elif next2 == "B":  # down: target z -
-                                controller.emitter_set_tracking_target_z(target_z - step)
+                                rts.set(tracking_target_z=target_z - step, quiet=True)
                                 print_status()
                             elif next2 == "C":  # right: target y +
-                                controller.emitter_set_tracking_target_y(target_y + step)
+                                rts.set(tracking_target_y=target_y + step, quiet=True)
                                 print_status()
                             elif next2 == "D":  # left: target y -
-                                controller.emitter_set_tracking_target_y(target_y - step)
+                                rts.set(tracking_target_y=target_y - step, quiet=True)
                                 print_status()
                 except IOError:
                     pass
@@ -710,81 +712,8 @@ class FlomniSampleTransferMixin:
         else:
             print("The rt feedback is \x1b[91mNOT\x1b[0m running.")
 
-    def interferometer_trigger_show(self):
-        """Show the detector-trigger stability criterion of the interferometer feedback.
-
-        A step-scan point is triggered when the relative position sample vs FZP (interferometer
-        minus FZP cap sensor) is within the tolerance in x and in y, each averaged over 0.5 ms,
-        for the hold time without interruption.
-        """
-        values = dev.rtx.controller.emitter_get_ontarget()
-        tol_nm = values["ontarget_tolerance_um"] * 1000
-        if tol_nm > 0:
-            print(f"Detector trigger: |position error| x and y <= {tol_nm:.1f} nm "
-                  f"for {values['ontarget_hold_ms']:.1f} ms.")
-        else:
-            print("Detector trigger: former criterion (on-target flags of the FZP PID loops).")
-        if values.get("trigger_on_move") is not None:
-            mode = "on" if values["trigger_on_move"] else "off"
-            print(f"Trigger after every position change (not only at scan points): {mode}.")
-
-    def interferometer_trigger_set_tolerance_nm(self, tolerance_nm: float):
-        """Set the detector-trigger tolerance in nm (per axis, x and y).
-
-        Typical: 30 nm for a fast low-resolution scan, 10 nm standard, 2 nm for the highest
-        resolution (needs a few ms more per point). 0 selects the former criterion (on-target flags
-        of the FZP PID loops).
-
-        Args:
-            tolerance_nm (float): tolerance in nm, 0 or 1 ... 1000.
-        """
-        if tolerance_nm != 0 and not 1 <= tolerance_nm <= 1000:
-            raise FlomniError(
-                "The trigger tolerance must be 0 (former criterion) or 1 ... 1000 nm."
-            )
-        dev.rtx.controller.emitter_set_ontarget_tolerance(tolerance_nm / 1000)
-        self.interferometer_trigger_show()
-
-    def interferometer_trigger_set_hold_ms(self, hold_ms: float):
-        """Set how long (ms) the position must stay within the tolerance before the trigger.
-
-        Typical 3 ms. Longer = safer against a position that is only passing through the tolerance.
-
-        Args:
-            hold_ms (float): hold time in ms, 0 ... 100.
-        """
-        if not 0 <= hold_ms <= 100:
-            raise FlomniError("The trigger hold time must be 0 ... 100 ms.")
-        dev.rtx.controller.emitter_set_ontarget_hold_ms(hold_ms)
-        self.interferometer_trigger_show()
-
-    def interferometer_trigger_set_after_move(self, enable: bool):
-        """Send one detector trigger after every position change, not only at scan points.
-
-        With True every direct move of the sample stage (e.g. umv of rtx / rty / rtz, or
-        dev.rtx.controller.move_xyz) gives exactly one trigger, as soon as the position is stable
-        (same tolerance and hold time as for scan points). False: triggers at scan points only.
-        """
-        dev.rtx.controller.emitter_set_trigger_on_move(1 if enable else 0)
-        self.interferometer_trigger_show()
-
-    def interferometer_trigger_help(self):
-        """Print the detector-trigger commands with a short explanation."""
-        lines = [
-            "Detector trigger (step scans): a point is triggered when the relative position",
-            "sample vs FZP is within the tolerance in x and y (0.5 ms average) for the hold time.",
-            "  flomni.interferometer_trigger_show()                 tolerance and hold time",
-            "  flomni.interferometer_trigger_set_tolerance_nm(val)  30 coarse, 10 standard,",
-            "                                                       2 finest; 0 = former",
-            "                                                       criterion (FZP PID flags)",
-            "  flomni.interferometer_trigger_set_hold_ms(val)       e.g. 3",
-            "  flomni.interferometer_trigger_set_after_move(True)   also one trigger after every",
-            "                                                       position change (False: off)",
-            "  dev.rtx.controller.move_xyz(x, y, z)                 move all three axes at once",
-            "The values are live; after a restart of the control loop the start values",
-            "(10 nm, 3 ms) from the loop parameter file apply again.",
-        ]
-        print("\n".join(lines))
+    # detector trigger: flomni.realtime_system.trigger_* (the former flomni.interferometer_trigger_* removed
+    # 2026-10-23, Mirko: "interferometer" was confusing next to flomni.laser_*)
 
     def lights_off(self):
         self.device_manager.devices.fsamx.controller.lights_off()
@@ -798,7 +727,7 @@ class FlomniSampleTransferMixin:
         self.flomnigui_raise()
 
         target_pos = -162
-        if np.isclose(dev.fsamx.readback.get(), target_pos, 0.01):
+        if np.isclose(dev.fsamx.readback.get(), target_pos, rtol=0, atol=0.5):
             return
 
         umv(dev.fsamroy, 0)
@@ -838,7 +767,7 @@ class FlomniSampleTransferMixin:
         sensor_voltage_target = dev.ftransy.user_parameter.get("sensor_voltage")
         sensor_voltage = float(dev.ftransy.controller.socket_put_and_receive("MG@AN[1]").strip())
 
-        if not np.isclose(sensor_voltage, sensor_voltage_target, 0.5):
+        if not np.isclose(sensor_voltage, sensor_voltage_target, rtol=0, atol=0.5):
             raise FlomniError(f"Sensor voltage is {sensor_voltage}, indicates an error. Aborting.")
 
     def ftransfer_get_sample(self, position: int):
@@ -1138,7 +1067,7 @@ class FlomniSampleTransferMixin:
         if 0 <= position < 21:
             return
         raise FlomniError(
-            f"The given position number [{position}] is not in the valid range of 0-21. "
+            f"The given position number [{position}] is not in the valid range of 0-20. "
         )
 
     def ftransfer_controller_enable_mount_mode(self):
@@ -1717,6 +1646,8 @@ class _ProgressProxy:
         "estimated_finish_time": None,
         "heartbeat": None,
         "accumulated_idle_time": 0.0,
+        # set when tomo_scan() reached its end: tomo_scan_resume() then has nothing to resume
+        "tomo_completed": False,
     }
 
     def __init__(self, client):
@@ -1792,7 +1723,7 @@ class Flomni(
         self.device_manager = client.device_manager
         self.special_angles = []
         self.special_angle_repeats = 20
-        self.special_angle_tolerance = 20
+        self.special_angle_tolerance = 0.5
         self._current_special_angles = []
         self._beam_is_okay = True
         self._progress_proxy = _ProgressProxy(self.client)
@@ -1819,6 +1750,9 @@ class Flomni(
         self._webpage_gen.start()
 
         self.OMNYTools = OMNYTools(self.client)
+        # expert functions of the real-time loop (settings with rules, via the web HMI on the loop machine;
+        # flomni_realtime_system.py): flomni.realtime_system.show() / .set(...) / .help()
+        self.realtime_system = FlomniRealtimeSystem(device_manager=self.device_manager)
         self.reconstructor = PtychoReconstructor(self.ptycho_reconstruct_foldername)
         self.tomo_id_manager = TomoIDManager()
         self.align = XrayEyeAlign(self.client, self)
@@ -1920,8 +1854,8 @@ class Flomni(
             try:
                 self.align.align(keep_shutter_open)
             except KeyboardInterrupt as exc:
-                fsamx_in = self._get_user_param_safe(dev.fsamx, "in")
-                if np.isclose(fsamx_in, dev.fsamx.readback.get(), 0.5):
+                fsamx_in = self._get_user_param_safe("fsamx", "in")  # name: dev[...] lookup (a device raised)
+                if np.isclose(fsamx_in, dev.fsamx.readback.get(), rtol=0, atol=0.5):
                     print("Stopping alignment. Returning to fsamx in position.")
                     self.feedback_disable()
                     umv(dev.fsamx, fsamx_in)
@@ -1982,10 +1916,14 @@ class Flomni(
     def set_special_angles(self, angles: list, repeats: int = 20, tolerance: float = 0.5):
         """Set the special angles for a tomo
 
+        The first projection of the tomogram within `tolerance` of a special angle is acquired
+        `repeats` times; each special angle is used once per tomogram (in any order), also across
+        tomo_scan_resume() in the same session.
+
         Args:
             angles (list): List of special angles
             repeats (int, optional): Number of repeats at a special angle. Defaults to 20.
-            tolerance (float, optional): Number of repeats at a special angle. Defaults to 0.5.
+            tolerance (float, optional): Angle tolerance in degrees. Defaults to 0.5.
 
         """
         self.special_angles = angles
@@ -2299,6 +2237,8 @@ class Flomni(
         self.write_alignment_scan_numbers(bec.queue.next_scan_number)
         start_angle = 0
         alignment_scan_numbers = []
+        # first scan number of the accepted attempt per angle (a retry shifts the scan numbers)
+        alignment_first_scans = []
 
         angle_end = start_angle + 180
         for angle in np.linspace(start_angle, angle_end, num=int(180 / 45) + 1, endpoint=True):
@@ -2316,16 +2256,24 @@ class Flomni(
                             bec.queue.request_queue_reset()
                             time.sleep(2)
                             error_caught = True
+                            print(f"Timeout at angle {angle}, repeating it (Ctrl-C to stop).")
                         else:
                             raise exc
                     # todo here was if blchk success, then setting to success true
-                    successful = True
+                    # repeat the angle after a timeout (it was skipped silently before, 2026-10-24)
+                    successful = not error_caught
 
                 end_scan_number = bec.queue.next_scan_number
+                alignment_first_scans.append(start_scan_number)
                 for scan_nr in range(start_scan_number, end_scan_number):
                     # self._write_tomo_scan_number(scan_nr, angle, 0)
                     alignment_scan_numbers.append(scan_nr)
 
+        consecutive = [alignment_first_scans[0] + k for k in range(len(alignment_first_scans))]
+        if alignment_first_scans != consecutive:
+            # a retry (or stitching) shifted the scan numbers: rewrite the file with the accepted scans
+            self.write_alignment_scan_numbers(alignment_first_scans[0], scans=alignment_first_scans)
+
         umv(dev.fsamroy, 0)
         self.OMNYTools.printgreenbold(
             "\n\nAlignment scan finished. Please run BEC_ptycho_align and load the new fit by flomni.read_alignment_offset() ."
@@ -2345,13 +2293,14 @@ class Flomni(
         print(scilog_content)
         self._scilog_write(scilog_content, "alignmentscan")
 
-    def write_alignment_scan_numbers(self, first_scan):
+    def write_alignment_scan_numbers(self, first_scan, scans=None):
         import os
 
         file = os.path.expanduser("~/data/raw/logs/ptychotomoalign_scannum.txt")
         os.makedirs(os.path.dirname(file), exist_ok=True)
 
-        scans = [first_scan + k for k in range(5)]
+        if scans is None:
+            scans = [first_scan + k for k in range(5)]
         angles = [0, 45, 90, 135, 180]
 
         x_vals = []
@@ -2643,10 +2592,16 @@ class Flomni(
             # succeeded, so a retried attempt still sees the same special
             # angle and repeat count.
             num_repeats = 1
-            is_special_angle = bool(
-                self._current_special_angles
-                and np.isclose(angle, self._current_special_angles[0], atol=0.5)
+            # any remaining special angle within the tolerance (not only the next one in the list)
+            special_index = next(
+                (
+                    k
+                    for k, special in enumerate(self._current_special_angles)
+                    if np.isclose(angle, special, rtol=0, atol=self.special_angle_tolerance)
+                ),
+                None,
             )
+            is_special_angle = special_index is not None
             if is_special_angle:
                 num_repeats = self.special_angle_repeats
             start_scan_number = bec.queue.next_scan_number
@@ -2656,7 +2611,7 @@ class Flomni(
                 self._at_each_angle(angle, **overrides)
             projection_duration = time.perf_counter() - projection_start
             if is_special_angle:
-                self._current_special_angles.pop(0)
+                self._current_special_angles.pop(special_index)
 
             end_scan_number = bec.queue.next_scan_number
             for scan_nr in range(start_scan_number, end_scan_number):
@@ -2673,8 +2628,8 @@ class Flomni(
             )
 
     def collect_empty_frames(self):
-        """Acquire 10 empty-frame (flat-field) images at angle 0, with fsamx
-        shifted out of the beam by half the field of view (fovx/2), in the
+        """Acquire 10 empty-frame (flat-field) images at angle 0, with the
+        sample moved out of the beam by half the field of view (fovx/2), in the
         direction opposite the normal alignment x-offset at angle 0.
 
         Called once at the start of a new tomo_scan() (not on resume).
@@ -2786,15 +2741,15 @@ class Flomni(
             if self.OMNYTools.yesno("Shall I continue?", "n"):
                 print("OK")
             else:
-                print("Stopping.")
-                return
+                # raise, not return: a return let tomo_queue_execute() mark the job "done" without data
+                raise FlomniError("Tomo scan stopped: the setup is not in measurement condition.")
 
         all_out, offending = filters_out_of_beam(dev)
         if not all_out:
             warning = f"Not all filters are out of the beam: {', '.join(offending)}."
             if not warn_and_confirm(self, warning, interactive=interactive, force=False):
-                print("Aborting tomo scan.")
-                return
+                # raise, not return (see above)
+                raise FlomniError(f"Tomo scan aborted: {warning}")
 
         self.flomnigui_show_progress()
 
@@ -2804,7 +2759,9 @@ class Flomni(
         bec.builtin_actors.scan_interlock.trigger_setting = "restart_scan"
         bec.builtin_actors.scan_interlock.enabled = True
 
-        self._current_special_angles = self.special_angles.copy()
+        if is_new_scan or not self.special_angles:
+            # on a resume keep the special angles not yet acquired (they were taken again before)
+            self._current_special_angles = list(self.special_angles)
         if is_new_scan:
             # pylint: disable=undefined-variable
             # Always attempt registration (even for an empty/test account)
@@ -2833,6 +2790,7 @@ class Flomni(
             # self.progress.reset() above, before the gates -- only the actual
             # start time is set here, once the scan is genuinely committed to.
             self.progress["tomo_start_time"] = datetime.datetime.now().isoformat()
+            self.progress["tomo_completed"] = False
             if file_angles is not None:
                 self._init_angle_file_progress(file_angles, projection_number)
             self.collect_empty_frames()
@@ -2857,6 +2815,9 @@ class Flomni(
                             start_angle is None
                             and self._subtomo_starts_near_zero(ii)
                             and self.zero_deg_reference_at_each_subtomo
+                            # new scan: the fixed 0-deg shot above and sub-tomogram 1's first
+                            # projection (0 deg) are already the two at the start (Mirko 2026-10-24)
+                            and not (is_new_scan and ii == 1)
                         ):
                             # Dedicated reference shot at exactly 0 degrees, taken
                             # every time the rotation passes back through 0, for
@@ -3009,6 +2970,8 @@ class Flomni(
             # (120s / 90s) after a tomogram that's actually already done.
             self.progress["heartbeat"] = None
 
+        # all projections acquired: nothing left for tomo_scan_resume() (also if a step below fails)
+        self.progress["tomo_completed"] = True
         self.progress["projection"] = self.progress["total_projections"]
         self.progress["subtomo_projection"] = self.progress["subtomo_total_projections"]
         self._print_progress()
@@ -3062,6 +3025,10 @@ class Flomni(
         if self.progress.get("tomo_start_time") is None:
             print("No tomo scan in progress to resume -- nothing to do.")
             return
+        if self.progress.get("tomo_completed"):
+            # it re-acquired the last projection (and repeated the end-of-scan steps) before, 2026-10-24
+            print("The last tomo scan was completed -- nothing to resume.")
+            return
 
         if not from_queue and not self._confirm_resume_outside_queue():
             print("Aborted.")
@@ -4062,7 +4029,7 @@ class Flomni(
 
             if self.tomo_type == 1:
                 self.tomo_angle_range = self._get_val(
-                    "Angular range (180 or 360)", self.tomo_angle_range, int
+                    "Angular range (180 or 360)", self.tomo_angle_range, int, choices=(180, 360)
                 )
                 # Default shown here must be the actual achievable total
                 # (int(range/stepsize)*8), not the raw algebraic inverse of
@@ -4179,14 +4146,23 @@ class Flomni(
                 self._prompt_angle_file()
 
     @staticmethod
-    def _get_val(msg: str, default_value, data_type, min_=None, max_=None):
+    def _get_val(msg: str, default_value, data_type, min_=None, max_=None, choices=None):
         """Prompt for a value, casting to data_type. When min_/max_ are
         given, re-prompts (rather than silently clamping) until the typed
         value is in range, mirroring the Tomo Parameters GUI's spinbox
         bounds (tomo_params.py) so tomo_parameters() enforces the same
-        limits instead of accepting anything of the right type."""
+        limits instead of accepting anything of the right type. Also
+        re-prompts on a value that cannot be cast, or that is not one of
+        `choices` (when given)."""
         while True:
-            val = data_type(input(f"{msg} ({default_value}): ") or default_value)
+            try:
+                val = data_type(input(f"{msg} ({default_value}): ") or default_value)
+            except ValueError:
+                print(f"Not a valid {data_type.__name__}. Please try again.")
+                continue
+            if choices is not None and val not in choices:
+                print(f"Value must be one of {', '.join(str(c) for c in choices)}. Please try again.")
+                continue
             if min_ is not None and val < min_:
                 print(f"Value must be >= {min_}. Please try again.")
                 continue

csaxs_bec/bec_ipython_client/plugins/flomni/flomni_optics_mixin.py

@@ -88,8 +88,10 @@ class FlomniOpticsMixin:
         if self.OMNYTools.yesno("Did the direct beam on the xray eye disappear?"):
             print("excellent.")
         else:
-            print("Aborting. With visible parts of the direct beam on the xray eye, it cannot be removed.")
-            return
+            # raise, not return: callers (tomo_alignment_scan) went on measuring with the eye in
+            raise FlomniOpticsError(
+                "Aborting. With visible parts of the direct beam on the xray eye, it cannot be removed."
+            )
         feyex_out = self._get_user_param_safe("feyex", "out")
 
         umv(dev.feyex, feyex_out)
@@ -104,8 +106,7 @@ class FlomniOpticsMixin:
                 "this could cause collisions when operating on the real "
                 "beamline."
             ):
-                print("Aborting feye_out(): fttrx1 not moved.")
-                return
+                raise FlomniOpticsError("Aborting feye_out(): fttrx1 not moved.")
         else:
             fttrx_in = self._get_user_param_safe("feyex", "fttrx_in")
             umv(dev.fttrx1, fttrx_in)
@@ -119,8 +120,7 @@ class FlomniOpticsMixin:
                 "this could cause collisions when operating on the real "
                 "beamline."
             ):
-                print("Aborting feye_in(): fttrx1 not moved.")
-                return
+                raise FlomniOpticsError("Aborting feye_in(): fttrx1 not moved.")
         else:
             fttrx_out = self._get_user_param_safe("feyex", "fttrx_out")
             umv(dev.fttrx1, fttrx_out)
@@ -420,7 +420,8 @@ class FlomniOpticsMixin:
 
         print("\nOSA Information:")
         print(f"  Current fosaz {fosaz_val:.1f}")
-        print(f"  The OSA will collide with a normal OMNY pin at fosaz \033[1m{(33-fosaz_val):.1f}\033[0m")
+        # collision where compute_osa_remaining_space() = 33 - foptz - fosaz reaches 0 (was 33 - fosaz)
+        print(f"  The OSA will collide with a normal OMNY pin at fosaz \033[1m{(33-foptz_val):.1f}\033[0m")
         print(f"  Remaining space (right now): \033[1m{remaining_current:.1f}\033[0m")
 
         diff = fosaz_val - fosaz_in  # >0: OSA is currently more "in" than nominal -> worse

csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py

@@ -0,0 +1,273 @@
+"""flomni.realtime_system: the operating settings of the flOMNI real-time loop.
+
+BEAMLINE SCIENTISTS / LOCAL CONTACT ONLY - not for users. These settings change how the FZP scanner, the detector
+trigger and the laser tracker work; a wrong combination can spoil a beamtime's data. The rules refuse combinations
+that do not work, but not ones that are only unsuitable for an experiment.
+
+The settings, the parameters they map onto and the rules between them live in the flOMNI instrument repository
+(FeedbackLoop/loopsettings/flomni.json) and are applied by the web HMI on the loop machine (RT-system
+webhmi/loopsettings.py). This client only talks to its REST API (/api/settings); it holds no logic of its own, so the
+web page (tab "Loop settings") and BEC always behave the same way.
+
+    flomni.realtime_system.show()                          all settings, rules, scan state
+    flomni.realtime_system.set(trigger_tolerance_nm=5)     change (refused if a rule objects)
+    flomni.realtime_system.set(fzp_enable=0, adjust=True)  ... and adjust the dependent settings
+    flomni.realtime_system.check(fzp_range_V_x=1.0)        dry run
+    flomni.realtime_system.apply_profile("fzp_off")        a named set of settings
+    flomni.realtime_system.save_as_default()               write the live values into the .xmp files
+
+Changes are live only: the next start of the control loop uses the .xmp files again (save_as_default keeps them).
+
+Connection: through the BEC device flomni_loop_settings (RPC; the API token is in its device config,
+ptycho_flomni.yaml) when it exists and is enabled, else directly to the web HMI (devices/omny/webhmi_client.py: URL
+http://<rtx host>:8080 with port-80 fallback, token from ~/.flomni_webhmi_token or FLOMNI_WEBHMI_TOKEN_FILE).
+"""
+
+from csaxs_bec.devices.omny.webhmi_client import WebHmiClient
+from csaxs_bec.devices.omny.webhmi_client import WebHmiError as RealtimeSystemError  # name kept for BEC users
+
+
+STAFF_ONLY = ("flomni.realtime_system: beamline scientists / local contact only - not for users "
+              "(plugins/flomni/docs/local_contact_only-realtime_system.md).")
+
+
+def _fmt(v):
+    if v is None:
+        return "-"
+    if isinstance(v, float):
+        return f"{v:.6g}"
+    return str(v)
+
+
+class FlomniRealtimeSystem(WebHmiClient):
+    """Requests go through the BEC device flomni_loop_settings (RPC api_get / api_post; the token is in its device
+    config) when it exists and is enabled; otherwise directly to the web HMI (WebHmiClient,
+    devices/omny/webhmi_client.py; token from token_file / ~/.flomni_webhmi_token)."""
+
+    DEVICE = "flomni_loop_settings"
+
+    def __init__(self, url=None, token_file=None, device_manager=None, timeout=5.0, token=None):
+        super().__init__(url=url, token=token, token_file=token_file, device_manager=device_manager, timeout=timeout)
+
+    def _device(self):
+        try:
+            dev = getattr(self._device_manager.devices, self.DEVICE)
+        except Exception:  # noqa: BLE001  (no device manager / no such device)
+            return None
+        return dev if getattr(dev, "enabled", True) else None
+
+    def _request(self, path, body=None):
+        dev = self._device()
+        if dev is None:
+            return super()._request(path, body)            # direct HTTP
+        try:
+            return dev.api_get(path) if body is None else dev.api_post(path, body)
+        except Exception as exc:  # noqa: BLE001  (RPC error with the device-side message)
+            raise RealtimeSystemError(f"via device {self.DEVICE}: {exc}") from None
+
+    @property
+    def route(self):
+        """'device flomni_loop_settings' or the direct URL."""
+        return f"device {self.DEVICE}" if self._device() is not None else self.url
+
+    # ── reading ───────────────────────────────────────────────────────────────
+
+    def read(self) -> dict:
+        """The full state as the web HMI reports it (settings, rules, scan state, profiles)."""
+        return self._request("/api/settings")
+
+    def get(self, name: str = None):
+        """Live value of one setting, or a dict of all values."""
+        values = {s["name"]: s["value"] for s in self.read()["settings"]}
+        if name is None:
+            return values
+        if name not in values:
+            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."""
+        snap = self.read()
+        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:
+                continue
+            if g.get("expert") and not expert and not group:
+                continue
+            rows = [s for s in snap["settings"] if s["group"] == gname]
+            if not rows:
+                continue
+            print(f"\n{g.get('title', gname)}" + ("  (changed only between scans)" if g.get("interlock") else ""))
+            for s in rows:
+                info = f"  ({s['info']['label']} {s['info']['value']:.4g} {s['info']['unit']})" if s.get("info") else ""
+                mark = ("  NOT SAVED (.xmp " + _fmt(s["saved"]) + ")" if s["unsaved"] else "") + \
+                       ("  INCONSISTENT: " + "; ".join(s["issues"]) if not s["consistent"] else "")
+                print(f"  {s['name']:30s} {_fmt(s['value']):>10s} {s.get('unit', ''):3s}{info}{mark}")
+        bad = [r for r in snap["rules"] if not r["ok"]]
+        if bad:
+            print()
+            for r in bad:
+                print(f"{r['severity'].upper()}: {r['message']}")
+        if not expert and not group and any(g.get("expert") for g in snap["groups"]):
+            print("\n(expert groups hidden: show(expert=True))")
+        print("\n" + (snap.get("note") or ""))
+
+    def profiles(self):
+        """Print the named profiles and their values."""
+        for p in self.read()["profiles"]:
+            print(f"{p['name']:18s} {p.get('title', '')}")
+            if p.get("help"):
+                print(f"{'':18s} {p['help']}")
+            print(f"{'':18s} " + ", ".join(f"{k}={_fmt(v)}" for k, v in p["values"].items()))
+
+    def unsaved(self):
+        """Print the parameters whose live value differs from the .xmp (settings only)."""
+        un = self._request("/api/settings/unsaved")
+        if not un:
+            print("All settings are saved: the .xmp files hold the live values.")
+        for u in un:
+            print(f"  {u['module']}.{u['param']}: .xmp {u['saved']} -> live {u['live']}")
+        return un
+
+    # ── changing ──────────────────────────────────────────────────────────────
+
+    @staticmethod
+    def _print_result(r, dry=False):
+        head = ("Check: " if dry else "") + ("applied" if r.get("applied") else ("ok" if r["ok"] else "REFUSED"))
+        if r["ok"] and not r["params"]:
+            head = "no change"
+        print(head + (f" (profile {r['profile']})" if r.get("profile") else ""))
+        for c in r["changes"]:
+            print(f"  {c['name']}: {_fmt(c['old'])} -> {_fmt(c['new'])}"
+                  + (f"   (adjusted by rule {c['adjusted']})" if c.get("adjusted") else ""))
+        for e in r["errors"]:
+            print(f"  ERROR: {e}")
+        for w in r["warnings"]:
+            print(f"  warning: {w}")
+        if r.get("applied"):
+            print("  " + r.get("note", "Live only: save_as_default() to keep it."))
+
+    def check(self, adjust: bool = False, **changes) -> dict:
+        """Dry run: what would change, which rules object. Nothing is sent to the loop."""
+        r = self._request("/api/settings/plan", {"changes": changes, "profile": None, "adjust": adjust})
+        self._print_result(r, dry=True)
+        return r
+
+    def set(self, adjust: bool = False, quiet: bool = False, **changes) -> dict:
+        """Change settings, e.g. set(trigger_tolerance_nm=5). All parameters change in one loop step.
+
+        A rule that objects refuses the request (RealtimeSystemError); with adjust=True the settings the rule names
+        are changed with it (never the ones given here). Live only until save_as_default(). quiet: print nothing
+        (interactive tools); a refusal still raises.
+        """
+        if not changes:
+            raise RealtimeSystemError("nothing to set: e.g. set(trigger_tolerance_nm=5); show() lists the names")
+        r = self._request("/api/settings/apply", {"changes": changes, "profile": None, "adjust": adjust})
+        if not quiet:
+            self._print_result(r)
+        if not r["ok"]:
+            raise RealtimeSystemError("; ".join(r["errors"]))
+        return r
+
+    def apply_profile(self, name: str, adjust: bool = False, **changes) -> dict:
+        """Apply a named profile (profiles() lists them); extra keyword values override the profile's."""
+        r = self._request("/api/settings/apply", {"changes": changes, "profile": name, "adjust": adjust})
+        self._print_result(r)
+        if not r["ok"]:
+            raise RealtimeSystemError("; ".join(r["errors"]))
+        return r
+
+    def save_as_default(self, *names: str, confirm: bool = True) -> dict:
+        """Write the live values of the settings' parameters (all, or the named settings) into the .xmp files, so the
+        next loop start uses them (the web HMI keeps a backup of each file)."""
+        un = self._request("/api/settings/unsaved")
+        if not un:
+            print("Nothing to save: the .xmp files hold the live values.")
+            return {"saved": []}
+        if not names:
+            print("Save as default (written into the .xmp files of the loop):")
+            for u in un:
+                print(f"  {u['module']}.{u['param']}: {u['saved']} -> {u['live']}")
+            if confirm and input("Save? [y/N] ").strip().lower() not in ("y", "yes"):
+                print("not saved")
+                return {"saved": []}
+        r = self._request("/api/settings/save", {"names": list(names) or None})
+        print(r.get("message", ""), "; ".join(r.get("saved", [])))
+        return r
+
+    # ── shortcuts (detector trigger, laser tracker) ────────────────────────────
+
+    # 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")
+
+    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;
+        0 = the former criterion (OnTarget_legacy flags of the FZP PIDs)."""
+        return self.set(trigger_tolerance_nm=tolerance_nm)
+
+    def trigger_set_hold_ms(self, hold_ms: float):
+        """How long the position must stay within the tolerance before the trigger (ms), typical 3."""
+        return self.set(trigger_hold_ms=hold_ms)
+
+    def trigger_set_after_move(self, enable: bool):
+        """True: one detector trigger after every direct position change (umv rtx / rty / rtz, move_xyz) once on target;
+        False: triggers at step-scan points only."""
+        return self.set(trigger_on_move=1 if enable else 0)
+
+    def tracking_targets(self):
+        """-> (target y, target z) of the laser tracker on the PSD."""
+        v = self.get()
+        return v["tracking_target_y"], v["tracking_target_z"]
+
+    def help(self, name: str = None):
+        """help(): the functions; help("all"): every setting with its explanation; help(name): one setting (what it
+        does, unit, range, the loop parameters it sets, live and saved value), from the definition on the loop
+        machine (the same text as the tooltips of the web HMI tab)."""
+        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(); profiles(), unsaved(), get(name)")
+            print('Explanations: help("all") or help("<setting>"), e.g. help("move_beyond_um_x")')
+            return
+        snap = self.read()
+        settings = {s["name"]: s for s in snap["settings"]}
+        groups = {g["name"]: g for g in snap["groups"]}
+        if name == "all":
+            print(STAFF_ONLY)
+            for gname, g in groups.items():
+                rows = [s for s in snap["settings"] if s["group"] == gname]
+                if not rows:
+                    continue
+                print(f"\n{g.get('title', gname)}" + ("  (only between scans)" if g.get("interlock") else ""))
+                for s in rows:
+                    unit = f" [{s['unit']}]" if s.get("unit") else ""
+                    print(f"  {s['name']}{unit}: {s.get('title', '')}")
+                    if s.get("help"):
+                        print(f"      {s['help']}")
+            return
+        s = settings.get(name)
+        if s is None:
+            raise RealtimeSystemError(f"unknown setting '{name}'; help('all') lists them")
+        g = groups.get(s["group"], {})
+        print(STAFF_ONLY)
+        print(f"{s['name']} - {s.get('title', '')} (group {g.get('title', s['group'])}"
+              + (", only between scans)" if g.get("interlock") else ")"))
+        if s.get("help"):
+            print(f"  {s['help']}")
+        rng = "0 / 1" if s.get("type") == "bool" else f"{_fmt(s.get('min'))} ... {_fmt(s.get('max'))} {s.get('unit', '')}"
+        print(f"  range: {rng}")
+        print(f"  live: {_fmt(s.get('value'))}   saved (.xmp): {_fmt(s.get('saved'))}"
+              + ("   NOT SAVED" if s.get("unsaved") else "") + ("" if s.get("consistent", True) else "   INCONSISTENT"))
+        if s.get("info"):
+            print(f"  {s['info'].get('label', '')} {s['info']['value']:.4g} {s['info'].get('unit', '')}")
+        print("  sets: " + ", ".join(f"{t['module']}.{t['param']}[{t['index']}] = {t.get('write', 'v')}"
+                                     for t in s.get("targets", [])))
+        rules = [r for r in snap.get("rules", []) if name in r.get("settings", [])]
+        if rules:
+            print("  rules: " + ", ".join(r["name"] + ("" if r.get("ok") else " (VIOLATED)") for r in rules))

csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/TODO_flomni_review_2026-10_findings.md

@@ -0,0 +1,48 @@
+# TODO: OMNY has bugs that were fixed in flomni (review 2026-10-24)
+
+Found while reviewing `flomni.py` and its mixins (branch `fzp-move-beyond`, csaxs_bec commits
+`1a5b53c`, `b939793` and the follow-up). OMNY was not changed; port these on the OMNY branch.
+
+## 1. X-ray eye Ctrl-C handler cannot return osamx (`omny.py`, `start_x_ray_eye_alignment`)
+
+```python
+osamx_in = self._get_user_param_safe(dev.osamx, "in")
+if np.isclose(osamx_in, dev.osamx.readback.get(), 0.5):
+```
+
+- `_get_user_param_safe()` looks the device up with `dev[device]` and needs the **name**
+  (`"osamx"`). With the device object the lookup raises (`KeyError`) inside the
+  `except KeyboardInterrupt` handler, so osamx is never returned to its in position and the
+  user sees a KeyError instead of the interrupt. flomni fix: `self._get_user_param_safe("fsamx", "in")`.
+- The third positional argument of `np.isclose` is **rtol**, not atol: `0.5` means 50 % of the
+  readback. Use `np.isclose(a, b, rtol=0, atol=0.5)` (mm).
+
+## 2. Special angles (`omny.py`, the per-angle loop of the tomo scan)
+
+```python
+if self._current_special_angles:
+    next_special_angle = self._current_special_angles[0]
+    if np.isclose(angle, next_special_angle, atol=0.5):
+        self._current_special_angles.pop(0)
+        num_repeats = self.special_angle_repeats
+else:
+    num_repeats = 1
+```
+
+- `num_repeats` is unbound when special angles are set but the angle is not the next one ->
+  `UnboundLocalError` (flomni had the same bug, fixed earlier with `num_repeats = 1` up front).
+- The angle is popped before the projection ran: a repeated attempt (timeout / beam check)
+  no longer sees it as special.
+- `special_angle_tolerance` is never used (hard-coded 0.5); `__init__` sets it to 20 while
+  `set_special_angles()` defaults to 0.5; the `tolerance` docstring says "Number of repeats".
+- Only the first list entry can match: a special angle that is never reached blocks all later ones.
+
+flomni now: match any remaining special angle within `special_angle_tolerance`, pop it only after
+the projection succeeded, keep the remaining list across a resume in the same session
+(`Flomni._tomo_scan_at_angle`, tests `tests/tests_bec_ipython_client/test_flomni_special_angles.py`).
+
+## Already shared (nothing to do for OMNY)
+
+- `OMNYTools.tweak_cursor()` (OMNY_shared) no longer ends with a traceback when an arrow-key
+  escape sequence arrives in two parts or ESC is pressed alone (`read_escape_suffix()`), and its
+  "not in enabled devices" message names the device.

csaxs_bec/device_configs/ptycho_flomni.yaml

@@ -616,6 +616,28 @@ flomni_temphum:
   deviceTags:
     - ptycho_flomni
 # ############################################################
+# ############ flOMNI real-time loop settings #################
+# ############################################################
+# The loop settings (FZP, trigger, tracker; flOMNI FeedbackLoop/loopsettings) stored with every scan: read from the
+# web HMI on the loop machine at the start of each scan (baseline). Web HMI down: available = 0, the scan continues.
+# The user commands flomni.realtime_system go through this device (RPC), so the token is needed only here.
+# token: the web HMI API token (~/loop-data/webhmi/api_token on mpc2844; behind the beamline firewall, Mirko
+# 2026-10-23). When empty, token_file (on the machine of the device server) is used.
+flomni_loop_settings:
+  description: flOMNI real-time loop settings (web HMI loop settings, snapshot at scan start)
+  deviceClass: csaxs_bec.devices.omny.flomni_loop_settings.FlomniLoopSettings
+  deviceConfig:
+    host: mpc2844.psi.ch
+    token: "3CcGHMdz16G7zuC6iPHEoZPWvKDY1HkCZnTlP1msZbg"
+    token_file: ~/.flomni_webhmi_token
+    timeout: 2.0
+  enabled: true
+  onFailure: buffer
+  readOnly: true
+  readoutPriority: baseline
+  deviceTags:
+    - ptycho_flomni
+# ############################################################
 # ########## OMNY / flOMNI / LamNI fast shutter ##############
 # REPLACED BY SIMULATED BEAMLINE DEVICE !!!!!!!!!!!!!!!!
 # ############################################################
@@ -706,37 +728,37 @@ calculated_signal:
 #   deviceTags:
 #     - ptycho_flomni
 
-omny_panda_continuous:
-  readoutPriority: async
-  deviceClass: csaxs_bec.devices.panda_box.panda_box_omny.PandaBoxOMNY
-  deviceConfig:
-    host: omny-panda.psi.ch
-    raw_stream_mode: true
-    # Only actually Arm() PandA for scans that consume its data -- generic BEC staging
-    # (stage_all_devices/pre_scan_all_devices) otherwise arms this device for every scan type
-    # (grid_scan, line_scan, alignment moves, ...), not just the ptycho scan that reads it. See
-    # docs/developer/panda_box_free_running_setup.md and
-    # AI_docs/PANDA_POSITION_VALIDATION.md sec 7 (stale frame_counter exposure).
-    arm_scan_allowlist:
-      - flomni_fermat_scan
-    signal_alias:
-      FMC_IN.VAL1.Value: cap_voltage_fzp_y
-      FMC_IN.VAL2.Value: cap_voltage_fzp_x
-      INENC1.VAL.Value: interf_st_fzp_y
-      INENC2.VAL.Value: interf_st_fzp_x
-      INENC3.VAL.Value: interf_st_rotz
-      INENC4.VAL.Value: interf_st_rotx
-      PCAP.GATE_DURATION.Value: pcap_gate_duration_value
-      # TTLIN1 (detector trigger) feeds both blocks below - confirmed 2026-09-15 against the
-      # omny-panda hardware, see docs/developer/panda_box_free_running_setup.md. PCAP.BITS0
-      # is a shared 32-bit word; PandaBoxOMNY masks it to bit 0 (TTLIN1.VAL) before
-      # publishing gate_detector_active. COUNTER1 counts TTLIN1's falling edge, i.e. one
-      # increment per completed exposure; it also doubles as the on_complete signal. Confirmed
-      # working end-to-end 2026-09-15; now under speed testing beyond the initial 100 Hz.
-      PCAP.BITS0.Value: gate_detector_active
-      COUNTER1.OUT.Value: frame_counter
-  enabled: true
-  readOnly: false
-  softwareTrigger: false
-  deviceTags:
-    - ptycho_flomni
+# omny_panda_continuous:
+#   readoutPriority: async
+#   deviceClass: csaxs_bec.devices.panda_box.panda_box_omny.PandaBoxOMNY
+#   deviceConfig:
+#     host: omny-panda.psi.ch
+#     raw_stream_mode: true
+#     # Only actually Arm() PandA for scans that consume its data -- generic BEC staging
+#     # (stage_all_devices/pre_scan_all_devices) otherwise arms this device for every scan type
+#     # (grid_scan, line_scan, alignment moves, ...), not just the ptycho scan that reads it. See
+#     # docs/developer/panda_box_free_running_setup.md and
+#     # AI_docs/PANDA_POSITION_VALIDATION.md sec 7 (stale frame_counter exposure).
+#     arm_scan_allowlist:
+#       - flomni_fermat_scan
+#     signal_alias:
+#       FMC_IN.VAL1.Value: cap_voltage_fzp_y
+#       FMC_IN.VAL2.Value: cap_voltage_fzp_x
+#       INENC1.VAL.Value: interf_st_fzp_y
+#       INENC2.VAL.Value: interf_st_fzp_x
+#       INENC3.VAL.Value: interf_st_rotz
+#       INENC4.VAL.Value: interf_st_rotx
+#       PCAP.GATE_DURATION.Value: pcap_gate_duration_value
+#       # TTLIN1 (detector trigger) feeds both blocks below - confirmed 2026-09-15 against the
+#       # omny-panda hardware, see docs/developer/panda_box_free_running_setup.md. PCAP.BITS0
+#       # is a shared 32-bit word; PandaBoxOMNY masks it to bit 0 (TTLIN1.VAL) before
+#       # publishing gate_detector_active. COUNTER1 counts TTLIN1's falling edge, i.e. one
+#       # increment per completed exposure; it also doubles as the on_complete signal. Confirmed
+#       # working end-to-end 2026-09-15; now under speed testing beyond the initial 100 Hz.
+#       PCAP.BITS0.Value: gate_detector_active
+#       COUNTER1.OUT.Value: frame_counter
+#   enabled: true
+#   readOnly: false
+#   softwareTrigger: false
+#   deviceTags:
+#     - ptycho_flomni

csaxs_bec/device_configs/simulated_omny/simulated_bl_endstation_flomni.yaml

@@ -24,5 +24,5 @@ fsh:
 flomni:
  - !include ../ptycho_flomni.yaml
 
-machine:
- - !include ../machine.yml
+# machine:
+#  - !include ../machine.yml

csaxs_bec/devices/omny/flomni_loop_settings.py

@@ -0,0 +1,169 @@
+"""flomni_loop_settings: the operating settings of the flOMNI real-time loop as a baseline device.
+
+Read at the start of every scan (readoutPriority: baseline; the flOMNI Fermat scan calls read_baseline_devices in
+prepare_scan), so each scan file holds the loop settings it ran with: FZP on/off, FZP range, feedforward,
+move-beyond, exposure during the motion, Abbe correction, detector trigger, tracker targets and thresholds
+(definition: flOMNI FeedbackLoop/loopsettings/flomni.json; served by the web HMI on the loop machine, GET
+/api/settings). Settings only, not all loop parameters (decision Mirko 2026-10-23).
+
+Signals: one per setting (float, NaN when not available; names = the setting names), and
+    available       1 = read from the web HMI, 0 = not available (the scan continues; message says why)
+    message         "" or the reason it is not available
+    unsaved         number of settings whose live value differs from the .xmp (not saved as default)
+    inconsistent    number of settings whose parameters disagree (e.g. FZP x on, y off)
+    rules_violated  number of violated error rules
+    definition      path and sha1 of the definition file on the loop machine
+    snapshot_json   JSON: every setting (value, saved, unsaved, consistent, unit), the violated rules, the
+                    scan state, the definition path and sha1
+
+deviceConfig: token (the API token itself), else token_file (path to a file with it, default
+~/.flomni_webhmi_token), optional url / host / timeout (default 2 s for the scan snapshot). Connection details:
+devices/omny/webhmi_client.py.
+
+flomni.realtime_system (the user commands) calls api_get / api_post of this device by RPC, so the token is needed
+only here.
+"""
+
+import json
+import math
+import time
+
+from ophyd import Component as Cpt
+from ophyd import Device, Kind, Signal
+
+from csaxs_bec.devices.omny.webhmi_client import WebHmiClient, WebHmiError
+
+# the settings of flomni.json (2026-10-23); a setting added to the definition later is in snapshot_json only until
+# it is added here
+SETTINGS = (
+    "fzp_enable", "fzp_range_V_x", "fzp_range_V_y", "feedforward_enable", "move_beyond_um_x", "move_beyond_um_y",
+    "exposure_during_motion", "abbe_enable", "abbe_limit_um",
+    "trigger_tolerance_nm", "trigger_hold_ms", "trigger_min_delay_ms", "trigger_on_move",
+    "tracking_target_y", "tracking_target_z", "psd_threshold_low", "interferometer_threshold",
+    "interferometer_threshold_rot",
+    "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)
+)
+
+RETRY_AFTER_FAILURE_S = 60.0    # web HMI not reachable: no new attempt for this long (a tomogram does not wait each time)
+
+
+class FlomniLoopSettings(Device):
+    USER_ACCESS = ["show", "snapshot", "api_get", "api_post"]
+
+    available = Cpt(Signal, value=0, kind=Kind.normal)
+    message = Cpt(Signal, value="", kind=Kind.normal)
+    unsaved = Cpt(Signal, value=-1, kind=Kind.normal)
+    inconsistent = Cpt(Signal, value=-1, kind=Kind.normal)
+    rules_violated = Cpt(Signal, value=-1, kind=Kind.normal)
+    definition = Cpt(Signal, value="", kind=Kind.normal)
+    snapshot_json = Cpt(Signal, value="", kind=Kind.normal)
+
+    # one signal per setting
+    locals().update({_n: Cpt(Signal, value=math.nan, kind=Kind.normal) for _n in SETTINGS})
+
+    def __init__(self, prefix="", *, name, token=None, token_file=None, url=None, host=None, timeout=2.0,
+                 device_manager=None, **kwargs):
+        super().__init__(prefix, name=name, **kwargs)
+        self._client = WebHmiClient(url=url, token=token, token_file=token_file, host=host,
+                                    device_manager=device_manager, timeout=timeout, client_name="BEC baseline")
+        # requests of flomni.realtime_system (RPC from the client, below): longer timeout, no back-off
+        self._api = WebHmiClient(url=url, token=token, token_file=token_file, host=host,
+                                 device_manager=device_manager, timeout=max(timeout, 10.0), client_name="BEC")
+        self._failed_at = None
+        self._last = None
+
+    def _unavailable(self, reason):
+        self.available.put(0)
+        self.message.put(str(reason))
+        for n in ("unsaved", "inconsistent", "rules_violated"):
+            getattr(self, n).put(-1)
+        self.definition.put("")
+        self.snapshot_json.put("")
+        for n in SETTINGS:
+            getattr(self, n).put(math.nan)
+
+    def update(self):
+        """Read the settings from the web HMI into the signals. Never raises: not available -> available 0."""
+        now = time.time()
+        if self._failed_at is not None and now - self._failed_at < RETRY_AFTER_FAILURE_S:
+            return                                  # keep the 'not available' values of the last failure
+        try:
+            snap = self._client.settings()
+        except WebHmiError as exc:
+            self._failed_at = now
+            self._unavailable(f"{exc} (next attempt after {RETRY_AFTER_FAILURE_S:.0f} s)")
+            return
+        except Exception as exc:  # noqa: BLE001  (a broken answer must not stop a scan)
+            self._failed_at = now
+            self._unavailable(f"unexpected answer from the web HMI: {exc}")
+            return
+        self._failed_at = None
+        self._last = snap
+        values = {s["name"]: s for s in snap.get("settings", [])}
+        for n in SETTINGS:
+            v = values.get(n, {}).get("value")
+            getattr(self, n).put(float(v) if isinstance(v, (int, float)) else math.nan)   # bool is an int too
+        self.unsaved.put(sum(1 for s in values.values() if s.get("unsaved")))
+        self.inconsistent.put(sum(1 for s in values.values() if not s.get("consistent", True)))
+        self.rules_violated.put(sum(1 for r in snap.get("rules", []) if not r.get("ok") and r.get("severity") == "error"))
+        self.definition.put(f"{snap.get('definition', '')} sha1 {snap.get('definition_sha1', '?')}")
+        missing = [n for n in values if n not in SETTINGS]
+        self.message.put(f"settings not stored as own signals (in snapshot_json): {', '.join(missing)}" if missing else "")
+        # compact: values and flags only (help texts and parameter mappings are in the definition, see its sha1)
+        compact = {
+            "instrument": snap.get("instrument"), "definition": snap.get("definition"),
+            "definition_sha1": snap.get("definition_sha1"), "scan": snap.get("scan"),
+            "settings": {n: {"value": s.get("value"), "saved": s.get("saved"), "unsaved": s.get("unsaved"),
+                             "consistent": s.get("consistent"), "unit": s.get("unit", "")}
+                         for n, s in values.items()},
+            "rules_violated": [{"name": r.get("name"), "severity": r.get("severity"), "message": r.get("message")}
+                               for r in snap.get("rules", []) if not r.get("ok")],
+        }
+        self.snapshot_json.put(json.dumps(compact, separators=(",", ":")))
+        self.available.put(1)
+
+    def read(self):
+        self.update()
+        return super().read()
+
+    # ── RPC for flomni.realtime_system: the user commands go through this device, so the API token is needed only
+    # in its device config (device server), not on the client. Not called set / put: the device server treats
+    # those names as ophyd writes (refused on a readOnly device).
+
+    _API_PATHS = ("/api/settings", "/api/status")
+
+    def _check_path(self, path):
+        if not any(path == p or path.startswith(p + "/") for p in self._API_PATHS):
+            raise WebHmiError(f"path {path!r} not allowed (only /api/settings*, /api/status)")
+
+    def api_get(self, path: str = "/api/settings") -> dict:
+        """GET a web HMI loop-settings endpoint; returns its JSON. Errors raise (passed to the client)."""
+        self._check_path(path)
+        return self._api._request(path)
+
+    def api_post(self, path: str, body: dict = None) -> dict:
+        """POST to a web HMI loop-settings endpoint (plan / apply / save); returns its JSON."""
+        self._check_path(path)
+        return self._api._request(path, body or {})
+
+    def snapshot(self) -> dict:
+        """Read now and return {setting: value} (empty when not available)."""
+        self._failed_at = None                      # an explicit request always tries
+        self.update()
+        if not self.available.get():
+            return {}
+        return {n: getattr(self, n).get() for n in SETTINGS}
+
+    def show(self):
+        """Print the settings as they would be stored with a scan now."""
+        snap = self.snapshot()
+        if not snap:
+            print(f"loop settings not available: {self.message.get()}")
+            return
+        for n, v in snap.items():
+            print(f"  {n:30s} {v:g}")
+        print(f"unsaved {self.unsaved.get()}, inconsistent {self.inconsistent.get()}, "
+              f"rules violated {self.rules_violated.get()}; {self.definition.get()}")
+        if self.message.get():
+            print(self.message.get())

csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py

@@ -13,6 +13,7 @@ from ophyd_devices.utils.socket import SocketIO, raise_if_disconnected
 from prettytable import PrettyTable
 
 from csaxs_bec.devices.omny.rt.rt_ophyd import (
+    format_scan_timing,
     RtCommunicationError,
     RtError,
     RtReadbackSignal,
@@ -45,17 +46,6 @@ class RtFlomniController(Controller):
         "laser_tracker_check_signalstrength",
         "laser_tracker_check_enabled",
         "is_axis_moving",
-        "emitter_show_all",
-        "emitter_get",
-        "emitter_set_tracking_target_y",
-        "emitter_set_tracking_target_z",
-        "emitter_set_intensity_threshold_laser",
-        "emitter_set_psd_intensity_threshold_tracking_low",
-        "emitter_get_ontarget",
-        "emitter_set_ontarget_tolerance",
-        "emitter_set_ontarget_hold_ms",
-        "emitter_help",
-        "emitter_set_trigger_on_move",
         "move_xyz",
     ]
 
@@ -404,84 +394,9 @@ class RtFlomniController(Controller):
             return True
         return False
 
-    def emitter_get(self) -> dict:
-        """Read the runtime constants served by the ConstEmitter Orchestra module.
-
-        These are values that used to live in individual module parameter files
-        (tracking targets and intensity thresholds) and are now emitted centrally
-        so they can be tweaked live. The rotation-loop laser threshold
-        (intensity_threshold_rot_laser) is a pure startup constant and is
-        intentionally not reported here.
-
-        Returns:
-            dict: current emitter values.
-        """
-        ret = self.socket_put_and_receive("es")
-        # remove trailing \n
-        ret = ret.split("\n")[0]
-        values = [float(val) for val in ret.split(",")]
-        if len(values) != 4:
-            raise RtCommunicationError(
-                f"Expected to receive 4 emitter values. Instead received {values}"
-            )
-        return {
-            "targetpos_tracking_y": values[0],
-            "targetpos_tracking_z": values[1],
-            "intensity_threshold_laser": values[2],
-            "PSD_intensity_threshold_tracking_low": values[3],
-        }
-
-    def emitter_show_all(self):
-        """Print all ConstEmitter runtime constants as a table."""
-        t = PrettyTable()
-        t.title = "ConstEmitter Runtime Constants"
-        t.field_names = ["Name", "Value"]
-        for key, val in self.emitter_get().items():
-            t.add_row([key, val])
-        for key, val in self.emitter_get_ontarget().items():
-            t.add_row([key, val])
-        print(t)
-
-    def emitter_get_ontarget(self) -> dict:
-        """Read the detector-trigger stability criterion (OnTargetTrigger in the control loop).
-
-        A step-scan point is triggered when the relative position sample vs FZP is within
-        ``ontarget_tolerance_um`` in x and in y (each averaged over 0.5 ms) for ``ontarget_hold_ms``
-        without interruption.
-
-        Returns:
-            dict: ontarget_tolerance_um, ontarget_hold_ms, trigger_on_move (None from a server
-            that does not report it yet).
-        """
-        ret = self.socket_put_and_receive("eg")
-        ret = ret.split("\n")[0]
-        values = [float(val) for val in ret.split(",")]
-        if len(values) not in (2, 3):
-            raise RtCommunicationError(
-                f"Expected to receive 2 or 3 on-target values. Instead received {values}"
-            )
-        return {
-            "ontarget_tolerance_um": values[0],
-            "ontarget_hold_ms": values[1],
-            "trigger_on_move": values[2] if len(values) == 3 else None,
-        }
-
-    def emitter_set_ontarget_tolerance(self, val_um: float) -> None:
-        """Set the detector-trigger tolerance in um, per axis.
-
-        E.g. 0.030 for a low-resolution scan, 0.002 for the highest resolution. A value <= 0 selects
-        the former criterion (on-target flags of the FZP PID loops, set in their parameter files).
-        """
-        self.socket_put(f"eo{val_um:.6f}")
-
-    def emitter_set_ontarget_hold_ms(self, val_ms: float) -> None:
-        """Set the time in ms the position must stay within the tolerance before the trigger."""
-        self.socket_put(f"eh{val_ms:.3f}")
-
-    def emitter_set_trigger_on_move(self, val: int) -> None:
-        """1: send one detector trigger after every direct position change (pa / p), once on target
-        (same criterion as a step-scan point); 0: triggers at step-scan points only."""
-        self.socket_put(f"em{1 if val else 0}")
+    # emitter_* (ConstEmitter get / set through the CommunicationServer e-commands) removed 2026-10-23: the
+    # ConstEmitter is no longer in the loop; its values are loop settings, set through the web HMI
+    # (flomni.realtime_system, csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py).
 
     def move_xyz(self, x: float, y: float, z: float) -> None:
         """Move rtx, rty and rtz with one command (CommunicationServer p<x>,<y>,<z>).
@@ -502,59 +417,6 @@ class RtFlomniController(Controller):
             targets.append(target_val)
         self.socket_put(f"p{targets[0]:.4f},{targets[1]:.4f},{targets[2]:.4f}")
 
-    def emitter_help(self) -> None:
-        """Print the ConstEmitter commands (loop runtime constants) with a short explanation."""
-        t = PrettyTable()
-        t.title = "ConstEmitter commands (rt controller)"
-        t.field_names = ["Method", "Wire", "Explanation"]
-        t.align = "l"
-        rows = [
-            ("emitter_show_all()", "es, eg", "print all runtime constants"),
-            ("emitter_get()", "es", "dict: tracking targets y/z, laser and PSD thresholds"),
-            ("emitter_get_ontarget()", "eg", "dict: detector-trigger tolerance (um), hold (ms)"),
-            ("emitter_set_tracking_target_y(val)", "ey", "target of the tracking loop y (PSD)"),
-            ("emitter_set_tracking_target_z(val)", "ez", "target of the tracking loop z (PSD)"),
-            ("emitter_set_intensity_threshold_laser(val)", "et",
-             "interferometer signal below this: slew limiters and x/y/z, FZP PIDs stop"),
-            ("emitter_set_psd_intensity_threshold_tracking_low(val)", "el",
-             "PSD intensity below this: tracking PIDs y/z stop"),
-            ("emitter_set_ontarget_tolerance(val_um)", "eo",
-             "detector trigger: |relative error| x and y <= val um (e.g. 0.030 coarse, 0.002 fine);"
-             " <= 0: FZP PID flags"),
-            ("emitter_set_ontarget_hold_ms(val_ms)", "eh",
-             "detector trigger: ... held for val ms without interruption"),
-            ("emitter_set_trigger_on_move(val)", "em",
-             "1: one detector trigger after every direct move (pa / p) once on target; 0: off"),
-        ]
-        for row in rows:
-            t.add_row(row)
-        print(t)
-        print(
-            "Values are live in the control loop; the start values after a loop restart come from"
-            " ConstEmitter.xmp."
-        )
-
-    def emitter_set_tracking_target_y(self, val: float) -> None:
-        """Set the y tracking target position (PID_tracking_y)."""
-        self.socket_put(f"ey{val:.4f}")
-
-    def emitter_set_tracking_target_z(self, val: float) -> None:
-        """Set the z tracking target position (PID_tracking_z)."""
-        self.socket_put(f"ez{val:.4f}")
-
-    def emitter_set_intensity_threshold_laser(self, val: float) -> None:
-        """Set the shared laser intensity threshold.
-
-        Wired into the slew-rate limiters (x/y/z) and the PID loops
-        (x/y/z and the fast FZP x/y loops). Does not affect the
-        rotation loops, which use a separate fixed threshold.
-        """
-        self.socket_put(f"et{val:.4f}")
-
-    def emitter_set_psd_intensity_threshold_tracking_low(self, val: float) -> None:
-        """Set the low PSD intensity threshold used by the tracking PIDs (y and z)."""
-        self.socket_put(f"el{val:.4f}")
-
     def pid_y(self) -> float:
         ret = float(self.socket_put_and_receive("G").strip())
         return ret
@@ -955,6 +817,11 @@ class RtFlomniFlyer(Device):
         )
         self.average_stdeviations_x_st_fzp = 0
         self.average_stdeviations_y_st_fzp = 0
+        # end-of-scan statistics: kickoff parameters from the scan, start / end of the scan
+        self._exp_time = None
+        self._frames_per_trigger = 1
+        self._t_kickoff = None
+        self._t_scan_end = None
         self.readout_thread = None
         self.scan_done_event = threading.Event()
         self.scan_done_event.set()
@@ -995,6 +862,8 @@ class RtFlomniFlyer(Device):
                     signals = self._get_signals_from_table(return_table)
                     self.data.set(signals)
 
+        self._t_scan_end = time.monotonic()  # scan finished in the RT loop (statistics)
+
         if self.shutdown_event.wait(0.05):
             logger.info("Shutdown event set, stopping readout.")
             # if we are here, the shutdown_event is set. We can exit the readout loop.
@@ -1016,11 +885,30 @@ class RtFlomniFlyer(Device):
         status.set_finished()
         self.progress.put(value=read_counter, max_value=number_of_positions_planned, done=True)
 
-        logger.info(
-            "Flomni statistics: Average of all standard deviations: x"
-            f" {self.average_stdeviations_x_st_fzp/read_counter*1000:.1f}, y"
-            f" {self.average_stdeviations_y_st_fzp/read_counter*1000:.1f}"
+        self._report_statistics(read_counter)
+
+    def _report_statistics(self, read_counter: int):
+        """At the end of a scan: mean of the per-point standard deviations (sample vs FZP during each exposure,
+        RT_sampler) to the device-server log and to the BEC command line (Mirko 2026-10-26)."""
+        if read_counter == 0:
+            logger.info("Flomni statistics: no scan points read, no average of the standard deviations")
+            return
+        msg = (
+            "flOMNI scan statistics: average of the standard deviations during the exposures"
+            f" ({read_counter} points, sample vs FZP): x"
+            f" {self.average_stdeviations_x_st_fzp/read_counter*1000:.1f} nm, y"
+            f" {self.average_stdeviations_y_st_fzp/read_counter*1000:.1f} nm"
+            + format_scan_timing(read_counter, self._t_kickoff, self._t_scan_end, self._exp_time,
+                                 self._frames_per_trigger)
         )
+        logger.info(msg)
+        try:
+            self.controller.device_manager.connector.send_client_info(
+                msg, scope="flomni_scan_statistics", show_asap=True
+            )
+        except Exception as exc:  # pylint: disable=broad-except
+            # the statistics are information only: never fail the readout for them
+            logger.warning(f"Flomni statistics not sent to the client: {exc}")
 
     def _get_signals_from_table(self, return_table) -> dict:
         self.average_stdeviations_x_st_fzp += float(return_table[4])
@@ -1054,12 +942,22 @@ class RtFlomniFlyer(Device):
         )
         self.readout_thread.start()
 
+    def configure(self, d: dict) -> tuple:
+        """The device server calls this with the kickoff parameters of the scan before kickoff():
+        exp_time / frames_per_trigger for the end-of-scan statistics (not passed: no exposure part)."""
+        d = dict(d or {})
+        self._exp_time = d.pop("exp_time", None)
+        self._frames_per_trigger = d.pop("frames_per_trigger", 1) or 1
+        return super().configure(d) if d else ({}, {})
+
     def kickoff(self) -> DeviceStatus:
         self.shutdown_event.clear()
         self.scan_done_event.clear()
         while not self.controller._min_scan_buffer_reached and not self.shutdown_event.wait(0.001):
             ...
         self.controller.start_scan()
+        self._t_kickoff = time.monotonic()
+        self._t_scan_end = None
         self.shutdown_event.wait(0.1)
         status = DeviceStatus(self)
         status.set_finished()

csaxs_bec/devices/omny/rt/rt_lamni_ophyd.py

@@ -13,7 +13,7 @@ from ophyd_devices.utils.controller import Controller, retry_once, threadlocked
 from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
 from prettytable import PrettyTable
 
-from csaxs_bec.devices.omny.rt.rt_ophyd import RtCommunicationError, RtError
+from csaxs_bec.devices.omny.rt.rt_ophyd import RtCommunicationError, RtError, format_scan_timing
 
 logger = bec_logger.logger
 
@@ -752,6 +752,11 @@ class RtLamniFlyer(Device):
         self.average_stdeviations_x_st_fzp = 0
         self.average_stdeviations_y_st_fzp = 0
         self.average_lamni_angle = 0
+        # end-of-scan statistics: kickoff parameters from the scan, start / end of the scan
+        self._exp_time = None
+        self._frames_per_trigger = 1
+        self._t_kickoff = None
+        self._t_scan_end = None
         self.readout_thread = None
         self.scan_done_event = threading.Event()
         self.scan_done_event.set()
@@ -767,12 +772,22 @@ class RtLamniFlyer(Device):
         )
         self.readout_thread.start()
 
+    def configure(self, d: dict) -> tuple:
+        """The device server calls this with the kickoff parameters of the scan before kickoff():
+        exp_time / frames_per_trigger for the end-of-scan statistics (not passed: no exposure part)."""
+        d = dict(d or {})
+        self._exp_time = d.pop("exp_time", None)
+        self._frames_per_trigger = d.pop("frames_per_trigger", 1) or 1
+        return super().configure(d) if d else ({}, {})
+
     def kickoff(self) -> DeviceStatus:
         self.shutdown_event.clear()
         self.scan_done_event.clear()
         while not self.controller._min_scan_buffer_reached and not self.shutdown_event.wait(0.001):
             ...
         self.controller.start_scan()
+        self._t_kickoff = time.monotonic()
+        self._t_scan_end = None
         self.shutdown_event.wait(0.1)
         status = DeviceStatus(self)
         status.set_finished()
@@ -834,6 +849,8 @@ class RtLamniFlyer(Device):
                     signals = self._get_signals_from_table(return_table)
                     self.data.set(signals)
 
+        self._t_scan_end = time.monotonic()  # scan finished in the RT loop (statistics)
+
         if self.shutdown_event.wait(0.05):
             logger.info("Shutdown event set, stopping readout.")
             # if we are here, the shutdown_event is set. We can exit the readout loop.
@@ -855,9 +872,32 @@ class RtLamniFlyer(Device):
         status.set_finished()
         self.progress.put(value=read_counter, max_value=number_of_positions_planned, done=True)
 
-        logger.info(
-            f"LamNI statistics: Average of all standard deviations: x {self.average_stdeviations_x_st_fzp}, y {self.average_stdeviations_y_st_fzp}, angle {self.average_lamni_angle}."
+        self._report_statistics(read_counter)
+
+    def _report_statistics(self, read_counter: int):
+        """At the end of a scan: mean of the per-point standard deviations (sample vs FZP during each exposure,
+        RT_sampler) and the mean angle to the device-server log and to the BEC command line (Mirko 2026-10-26,
+        as flOMNI). Before, the log line printed the sums, not the averages."""
+        if read_counter == 0:
+            logger.info("LamNI statistics: no scan points read, no average of the standard deviations")
+            return
+        msg = (
+            "LamNI scan statistics: average of the standard deviations during the exposures"
+            f" ({read_counter} points, sample vs FZP): x"
+            f" {self.average_stdeviations_x_st_fzp/read_counter*1000:.1f} nm, y"
+            f" {self.average_stdeviations_y_st_fzp/read_counter*1000:.1f} nm;"
+            f" average angle {self.average_lamni_angle/read_counter:.4f}"
+            + format_scan_timing(read_counter, self._t_kickoff, self._t_scan_end, self._exp_time,
+                                 self._frames_per_trigger)
         )
+        logger.info(msg)
+        try:
+            self.controller.device_manager.connector.send_client_info(
+                msg, scope="lamni_scan_statistics", show_asap=True
+            )
+        except Exception as exc:  # pylint: disable=broad-except
+            # the statistics are information only: never fail the readout for them
+            logger.warning(f"LamNI statistics not sent to the client: {exc}")
 
     def _get_signals_from_table(self, return_table) -> dict:
         self.average_stdeviations_x_st_fzp += float(return_table[5])

csaxs_bec/devices/omny/rt/rt_ophyd.py

@@ -14,6 +14,26 @@ from ophyd_devices.utils.socket import SocketSignal
 logger = bec_logger.logger
 
 
+def format_scan_timing(n_points: int, t_start, t_end, exp_time=None, frames_per_trigger=1) -> str:
+    """Time split of a flyer scan for the end-of-scan statistics (Mirko 2026-10-26, BEC side only): scan time from
+    the kickoff to the end of the scan in the RT loop (wall clock of the device server), exposure = points x
+    requested exp_time x frames_per_trigger (not measured: the detector may expose longer), overhead = the rest
+    (positioning, trigger, detector readout). '' when the times are missing."""
+    if not n_points or t_start is None or t_end is None or t_end <= t_start:
+        return ""
+    total = t_end - t_start
+    txt = f"; scan time {total:.1f} s ({total / n_points * 1000:.0f} ms/point)"
+    if exp_time:
+        expo = n_points * float(exp_time) * int(frames_per_trigger or 1)
+        over = total - expo
+        txt += (
+            f": exposure {expo:.1f} s ({expo / total * 100:.0f} %, requested {float(exp_time) * 1000:g} ms"
+            f" x {int(frames_per_trigger or 1)}), overhead {over:.1f} s ({over / n_points * 1000:.0f} ms/point:"
+            " positioning, trigger, detector)"
+        )
+    return txt
+
+
 class RtCommunicationError(ControllerCommunicationError):
     pass
 

csaxs_bec/devices/omny/webhmi_client.py

@@ -0,0 +1,116 @@
+"""Client of the web HMI on an OMNY-family loop machine (RT-system webhmi): REST requests with the API token.
+
+Shared by the BEC client plugin (flomni.realtime_system, bec_ipython_client/plugins/flomni/flomni_realtime_system.py)
+and the baseline device FlomniLoopSettings (devices/omny/flomni_loop_settings.py), so both talk to the web HMI the
+same way. Standard library only (runs in the IPython client and in the device server).
+
+URL: explicit `url`, else env FLOMNI_WEBHMI_URL, else http://<host>:8080 with <host> = `host`, else the host of the
+rtx device config, else mpc2844.psi.ch. With the default URL a request that cannot reach port 8080 is repeated on
+port 80 of the same host (nftables redirect 80 -> 8080 on the loop machine, the port the PSI gateway passes); that URL
+is then kept. An explicit URL gets no fallback.
+
+Token: `token` (the value, e.g. from a device config that is not committed), else `token_file`, else env
+FLOMNI_WEBHMI_TOKEN_FILE, else ~/.flomni_webhmi_token. The same line as the web HMI's api_token_file on the loop
+machine (flOMNI FeedbackLoop/loopsettings/README.md). Never commit a token.
+"""
+
+import json
+import os
+import urllib.error
+import urllib.request
+
+DEFAULT_TOKEN_FILE = "~/.flomni_webhmi_token"
+DEFAULT_PORT = 8080
+DEFAULT_HOST = "mpc2844.psi.ch"
+
+
+class WebHmiError(Exception):
+    """A request was refused (rules, scan running) or the web HMI could not be reached (.unreachable = True)."""
+
+    unreachable = False
+
+
+class WebHmiClient:
+    def __init__(self, url=None, token=None, token_file=None, host=None, device_manager=None, timeout=5.0,
+                 client_name="BEC"):
+        self._url = url
+        self._fallback = None                   # port-80 URL that answered when 8080 did not (kept for the session)
+        self._token_value = token
+        self._token_file = token_file
+        self._host = host
+        self._device_manager = device_manager
+        self.timeout = timeout
+        self.client_name = client_name
+
+    @property
+    def url(self):
+        if self._fallback:
+            return self._fallback
+        if self._url:
+            return self._url.rstrip("/")
+        env = os.environ.get("FLOMNI_WEBHMI_URL")
+        if env:
+            return env.rstrip("/")
+        host = self._host or DEFAULT_HOST
+        if not self._host:
+            try:                                # the host of the rt controller (device config of rtx)
+                host = self._device_manager.devices.rtx._config["deviceConfig"]["host"] or host
+            except Exception:  # noqa: BLE001
+                pass
+        return f"http://{host}:{DEFAULT_PORT}"
+
+    def _explicit_url(self):
+        return bool(self._url or os.environ.get("FLOMNI_WEBHMI_URL"))
+
+    def _token(self):
+        if self._token_value:
+            return str(self._token_value).strip()
+        path = os.path.expanduser(self._token_file or os.environ.get("FLOMNI_WEBHMI_TOKEN_FILE") or DEFAULT_TOKEN_FILE)
+        try:
+            with open(path) as f:
+                tok = f.readline().strip()
+        except OSError:
+            raise WebHmiError(
+                f"No API token for the web HMI: {path} not readable. Copy the token from the loop machine "
+                "(~/loop-data/webhmi/api_token of the user control) into this file (mode 600), or set "
+                "FLOMNI_WEBHMI_TOKEN_FILE.") from None
+        if not tok:
+            raise WebHmiError(f"API token file {path} is empty")
+        return tok
+
+    def _request(self, path, body=None):
+        try:
+            return self._request_once(self.url, path, body)
+        except WebHmiError as exc:
+            # default URL on port 8080 not reachable: the loop machine also serves the web HMI on port 80
+            if exc.unreachable and not self._explicit_url() and not self._fallback:
+                alt = self.url.replace(f":{DEFAULT_PORT}", "")
+                r = self._request_once(alt, path, body)
+                self._fallback = alt
+                return r
+            raise
+
+    def _request_once(self, url, path, body=None):
+        data = None if body is None else json.dumps(body).encode()
+        req = urllib.request.Request(url + path, data=data, method="GET" if body is None else "POST",
+                                     headers={"Authorization": "Bearer " + self._token(),
+                                              "X-Client": self.client_name, "Content-Type": "application/json"})
+        try:
+            with urllib.request.urlopen(req, timeout=self.timeout) as r:
+                return json.load(r)
+        except urllib.error.HTTPError as exc:
+            try:
+                detail = json.load(exc).get("detail", exc.reason)
+            except Exception:  # noqa: BLE001
+                detail = exc.reason
+            if exc.code == 401:
+                detail = f"{detail} (API token not accepted by the web HMI: compare with its api_token_file)"
+            raise WebHmiError(f"web HMI {url}{path}: {exc.code} {detail}") from None
+        except (urllib.error.URLError, OSError) as exc:
+            err = WebHmiError(f"web HMI {url} not reachable: {exc}")
+            err.unreachable = True
+            raise err from None
+
+    def settings(self) -> dict:
+        """GET /api/settings: settings, rules, scan state, profiles."""
+        return self._request("/api/settings")

csaxs_bec/scans/flomni_fermat_scan.py

@@ -238,7 +238,11 @@ class FlomniFermatScan(ScanBase):
         """
 
         # send off the flyer
-        self.actions.kickoff(device=self.dev.rt_positions)
+        self.actions.kickoff(
+            device=self.dev.rt_positions,
+            # for the end-of-scan statistics of the flyer (exposure vs overhead)
+            parameters={"exp_time": self.exp_time, "frames_per_trigger": self.frames_per_trigger},
+        )
 
         # start the readout loop of the flyer
         status = self.actions.complete(device=self.dev.rt_positions, wait=False)

csaxs_bec/scans/lamni_fermat_scan.py

@@ -218,7 +218,11 @@ class LamniFermatScan(ScanBase):
         This is where the main scan logic should be implemented.
         """
 
-        self.actions.kickoff(device="rt_positions")
+        self.actions.kickoff(
+            device="rt_positions",
+            # for the end-of-scan statistics of the flyer (exposure vs overhead)
+            parameters={"exp_time": self.exp_time, "frames_per_trigger": self.frames_per_trigger},
+        )
 
         status = self.actions.complete(device="rt_positions", wait=False)
 

docs/user/ptychography/flomni.md

@@ -221,23 +221,12 @@ _There is also an enable without reset, which is used during tomography scans, w
 #### Detector trigger: position stability
 
 In step scans the detector is triggered when the sample is stable at the scan point: the relative position
-sample vs FZP (interferometer minus FZP cap sensor) must stay within a tolerance in x and in y (each averaged
-over 0.5 ms) for a hold time. Both values are live and can be changed between scans:
+sample vs FZP must stay within a tolerance in x and in y for a short hold time (standard 10 nm, 3 ms). The local
+contact can choose a coarser tolerance for fast low-resolution scans or a finer one for the highest resolution.
+Measured with 3 µm steps (2026-10-20): the trigger comes about 36 ms (x) / 48 ms (y) after the step at 10 nm,
+40 / 55 ms at 2 nm; after the trigger the position stays within about 4 nm even at a 30 nm tolerance.
 
-- `flomni.interferometer_trigger_show()` — current tolerance and hold time
-- `flomni.interferometer_trigger_set_tolerance_nm(10)` — tolerance per axis in nm: about 30 for a fast
-  low-resolution scan, 10 standard, 2 for the highest resolution (a few ms more per point);
-  0 selects the former criterion (on-target flags of the FZP loops)
-- `flomni.interferometer_trigger_set_hold_ms(3)` — hold time in ms
-- `flomni.interferometer_trigger_set_after_move(True)` — additionally one trigger after every position change
-  of the sample stage (e.g. `umv(dev.rtx, ...)`), as soon as the position is stable with the same tolerance and
-  hold time; `False` (default): triggers at scan points only
-- `dev.rtx.controller.move_xyz(x, y, z)` — moves rtx, rty and rtz with one command (one trigger for the move)
-- `flomni.interferometer_trigger_help()` — this list
-
-After a restart of the control loop the start values (10 nm, 3 ms) apply again. Measured with 3 µm steps
-(2026-10-20): the trigger comes about 36 ms (x) / 48 ms (y) after the step at 10 nm, 40 / 55 ms at 2 nm;
-after the trigger the position stays within about 4 nm even at a 30 nm tolerance.
+- `dev.rtx.controller.move_xyz(x, y, z)` — moves rtx, rty and rtz with one command
 
 ### Scanning in 2D and sample alignment
 
@@ -781,13 +770,13 @@ removed later.
 
 [See short version](flomni.md#sample-storage-and-transfer)
 
-### Detector trigger and interferometer tuning
+### Real-time loop settings
 
-The trigger tolerance and hold time: [Detector trigger: position stability](#detector-trigger-position-stability)
-(`flomni.interferometer_trigger_*`). The control loops of the sample stage, the FZP scanner and the laser
-tracker were characterized and retuned in October 2026; results, the present parameters and how to retune them
-are documented in the flOMNI control repository (`FeedbackLoop/feedback_report.md` with PDF,
-`FeedbackLoop/tuning_guide.md`).
+The operating settings of the real-time control loop (FZP scanner, detector trigger, laser tracker) are set by the
+local contact; please do not change them. Every scan stores the settings it ran with (baseline device
+`flomni_loop_settings`).
+
+Details for beamline scientists: `csaxs_bec/bec_ipython_client/plugins/flomni/docs/local_contact_only-realtime_system.md`.
 
 ### Heater
 

tests/tests_bec_ipython_client/test_flomni_realtime_system.py

@@ -0,0 +1,214 @@
+"""Tests for flomni.realtime_system (flomni_realtime_system.py): the client of the web HMI loop-settings API on the
+flOMNI loop machine. A small local HTTP server stands in for the web HMI (RT-system webhmi, /api/settings*); it checks
+the token header and answers like the real server. No BEC service, no loop."""
+
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+import pytest
+
+from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_realtime_system import (
+    FlomniRealtimeSystem,
+    RealtimeSystemError,
+)
+
+TOKEN = "test-token-0123456789abcdef"
+
+SNAP = {
+    "instrument": "flOMNI", "note": "Live only.", "scan": {"mode": 0, "active": False, "text": ""},
+    "groups": [{"name": "trigger", "title": "Detector trigger"}, {"name": "tracking", "title": "Tracker"}],
+    "settings": [
+        {"name": "trigger_tolerance_nm", "group": "trigger", "unit": "nm", "value": 10.0, "saved": 10.0,
+         "unsaved": False, "consistent": True, "issues": [], "info": None},
+        {"name": "tracking_target_y", "group": "tracking", "unit": "", "value": 0.13, "saved": 0.13,
+         "unsaved": False, "consistent": True, "issues": [], "info": None},
+        {"name": "tracking_target_z", "group": "tracking", "unit": "", "value": 0.2, "saved": 0.2,
+         "unsaved": False, "consistent": True, "issues": [], "info": None},
+    ],
+    "rules": [], "profiles": [{"name": "production", "title": "Production", "help": "", "values": {}}],
+}
+
+
+class _Handler(BaseHTTPRequestHandler):
+    requests = []
+
+    def log_message(self, *args):
+        pass
+
+    def _reply(self, code, body):
+        data = json.dumps(body).encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+    def _auth(self):
+        if self.headers.get("Authorization") != "Bearer " + TOKEN:
+            self._reply(401, {"detail": "login required"})
+            return False
+        return True
+
+    def do_GET(self):
+        if not self._auth():
+            return
+        if self.path == "/api/settings":
+            self._reply(200, SNAP)
+        elif self.path == "/api/settings/unsaved":
+            self._reply(200, [])
+        else:
+            self._reply(404, {"detail": "not found"})
+
+    def do_POST(self):
+        if not self._auth():
+            return
+        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+        _Handler.requests.append((self.path, body, self.headers.get("X-Client")))
+        changes = body.get("changes", {})
+        if "move_beyond_um_x" in changes:
+            self._reply(200, {"ok": False, "errors": ["Move-beyond x does not fit"], "warnings": [], "changes": [],
+                              "params": [], "profile": None, "applied": False})
+            return
+        ch = [{"name": k, "old": 0, "new": v, "adjusted": None} for k, v in changes.items()]
+        self._reply(200, {"ok": True, "errors": [], "warnings": [], "changes": ch,
+                          "params": [{"module": "M", "param": "p", "old": [0], "new": [1]}] if ch or body.get("profile") else [],
+                          "profile": body.get("profile"), "applied": self.path.endswith("apply"),
+                          "note": "Live only: save as default to keep them."})
+
+
+@pytest.fixture
+def server(tmp_path):
+    _Handler.requests = []
+    srv = HTTPServer(("127.0.0.1", 0), _Handler)
+    threading.Thread(target=srv.serve_forever, daemon=True).start()
+    tok = tmp_path / "token"
+    tok.write_text(TOKEN + "\n")
+    yield f"http://127.0.0.1:{srv.server_address[1]}", str(tok), tmp_path
+    srv.shutdown()
+
+
+def test_read_and_shortcuts(server, capsys):
+    url, tok, _ = server
+    rts = FlomniRealtimeSystem(url=url, token_file=tok)
+    assert rts.get("trigger_tolerance_nm") == 10.0
+    assert rts.tracking_targets() == (0.13, 0.2)
+    rts.show()
+    out = capsys.readouterr().out
+    assert "trigger_tolerance_nm" in out and "Live only." in out
+    with pytest.raises(RealtimeSystemError):
+        rts.get("nosuch")
+
+
+def test_set_sends_changes_and_token(server, capsys):
+    url, tok, _ = server
+    rts = FlomniRealtimeSystem(url=url, token_file=tok)
+    r = rts.trigger_set_tolerance_nm(5)
+    assert r["applied"]
+    path, body, client = _Handler.requests[-1]
+    assert path == "/api/settings/apply" and body == {"changes": {"trigger_tolerance_nm": 5}, "profile": None,
+                                                       "adjust": False} and client == "BEC"
+    assert "Live only" in capsys.readouterr().out
+    rts.set(tracking_target_y=0.2, quiet=True)
+    assert capsys.readouterr().out == ""
+    rts.apply_profile("production", adjust=True)
+    assert _Handler.requests[-1][1] == {"changes": {}, "profile": "production", "adjust": True}
+    rts.check(fzp_enable=0, adjust=True)
+    assert _Handler.requests[-1][0] == "/api/settings/plan"
+
+
+def test_refused_raises(server):
+    url, tok, _ = server
+    with pytest.raises(RealtimeSystemError, match="does not fit"):
+        FlomniRealtimeSystem(url=url, token_file=tok).set(move_beyond_um_x=0.5)
+    with pytest.raises(RealtimeSystemError, match="nothing to set"):
+        FlomniRealtimeSystem(url=url, token_file=tok).set()
+
+
+def test_token_errors(server):
+    url, _, tmp = server
+    with pytest.raises(RealtimeSystemError, match="No API token"):
+        FlomniRealtimeSystem(url=url, token_file=str(tmp / "missing")).get()
+    bad = tmp / "bad"
+    bad.write_text("wrong-token-xxxxxxxxxxxx")
+    with pytest.raises(RealtimeSystemError, match="401"):
+        FlomniRealtimeSystem(url=url, token_file=str(bad)).get()
+
+
+def test_unreachable():
+    with pytest.raises(RealtimeSystemError, match="not reachable"):
+        FlomniRealtimeSystem(url="http://127.0.0.1:9", token_file=__file__, timeout=1).get()
+
+
+def test_port_80_fallback(monkeypatch):
+    """Default URL (rtx host, port 8080) not reachable: the same request goes to port 80 (nftables redirect on the
+    loop machine) and that URL is kept; an explicit URL gets no fallback."""
+    monkeypatch.delenv("FLOMNI_WEBHMI_URL", raising=False)
+    dm = type("DM", (), {})()
+    dm.devices = type("D", (), {})()
+    dm.devices.rtx = type("R", (), {"_config": {"deviceConfig": {"host": "loophost"}}})()
+    rts = FlomniRealtimeSystem(device_manager=dm, token_file=__file__)
+    calls = []
+
+    def once(url, path, body=None):
+        calls.append(url)
+        if url.endswith(":8080"):
+            err = RealtimeSystemError("not reachable")
+            err.unreachable = True
+            raise err
+        return {"ok": True}
+
+    monkeypatch.setattr(rts, "_request_once", once)
+    assert rts._request("/api/settings") == {"ok": True}
+    assert calls == ["http://loophost:8080", "http://loophost"] and rts.url == "http://loophost"
+    explicit = FlomniRealtimeSystem(url="http://other:8080", token_file=__file__)
+    monkeypatch.setattr(explicit, "_request_once", once)
+    with pytest.raises(RealtimeSystemError):
+        explicit._request("/api/settings")
+
+
+def test_routes_through_the_device():
+    """With the BEC device flomni_loop_settings present and enabled, requests go to it by RPC (api_get / api_post),
+    no token on the client; errors of the device come back as RealtimeSystemError."""
+    calls = []
+
+    class Dev:
+        enabled = True
+
+        def api_get(self, path):
+            calls.append(("get", path))
+            return {"settings": [{"name": "trigger_tolerance_nm", "value": 10.0}]}
+
+        def api_post(self, path, body):
+            calls.append(("post", path, body))
+            if "move_beyond_um_x" in body.get("changes", {}):
+                raise RuntimeError("web HMI not reachable")
+            return {"ok": True, "errors": [], "warnings": [], "changes": [], "params": [], "applied": False}
+
+    dm = type("DM", (), {})()
+    dm.devices = type("D", (), {})()
+    dm.devices.flomni_loop_settings = Dev()
+    rts = FlomniRealtimeSystem(device_manager=dm, token_file="/nonexistent")   # token never read
+    assert rts.get("trigger_tolerance_nm") == 10.0 and rts.route == "device flomni_loop_settings"
+    rts.check(fzp_enable=0)
+    assert calls[-1] == ("post", "/api/settings/plan", {"changes": {"fzp_enable": 0}, "profile": None, "adjust": False})
+    with pytest.raises(RealtimeSystemError, match="via device flomni_loop_settings: web HMI not reachable"):
+        rts.set(move_beyond_um_x=0.5)
+    dm.devices.flomni_loop_settings.enabled = False                        # disabled: direct HTTP
+    with pytest.raises(RealtimeSystemError, match="No API token"):
+        rts.get()
+
+
+def test_help(server, capsys):
+    """help(name) / help("all"): explanations from the definition, marked as beamline-scientist only."""
+    url, tok, _ = server
+    rts = FlomniRealtimeSystem(url=url, token_file=tok)
+    rts.help("trigger_tolerance_nm")
+    out = capsys.readouterr().out
+    assert "local contact only" in out and "trigger_tolerance_nm" in out and "live: 10" in out
+    rts.help("all")
+    assert "Detector trigger" in capsys.readouterr().out
+    rts.help()
+    assert "BEAMLINE SCIENTISTS" in capsys.readouterr().out
+    with pytest.raises(RealtimeSystemError, match="unknown setting"):
+        rts.help("nosuch")

tests/tests_bec_ipython_client/test_flomni_review_fixes.py

@@ -0,0 +1,229 @@
+"""Fixes from the flomni.py review 2026-10-24: tomo_scan_resume() after a completed scan, special angles kept
+on resume, the alignment scan repeats an angle after a timeout, absolute np.isclose tolerances."""
+
+import builtins
+import types
+
+import pytest
+
+import csaxs_bec.bec_ipython_client.plugins.flomni.flomni as flomni_module
+from csaxs_bec.bec_ipython_client.plugins.flomni.flomni import Flomni, FlomniError, _ProgressProxy
+
+
+class _FakeClient:
+    def __init__(self):
+        self._vars = {}
+
+    def get_global_var(self, key):
+        return self._vars.get(key)
+
+    def set_global_var(self, key, value):
+        self._vars[key] = value
+
+
+def _make_flomni():
+    obj = object.__new__(Flomni)
+    obj.client = _FakeClient()
+    obj._progress_proxy = _ProgressProxy(obj.client)
+    return obj
+
+
+def test_resume_after_completed_scan_does_nothing(monkeypatch):
+    flomni = _make_flomni()
+    flomni.progress["tomo_start_time"] = "2026-10-24T10:00:00"
+    flomni.progress["tomo_completed"] = True
+    monkeypatch.setattr(Flomni, "tomo_type", 1, raising=False)
+    flomni.tomo_scan = lambda **kw: pytest.fail("tomo_scan must not be called")
+    flomni._confirm_resume_outside_queue = lambda: True
+
+    flomni.tomo_scan_resume()
+
+
+def test_resume_of_interrupted_scan_still_resumes(monkeypatch):
+    flomni = _make_flomni()
+    flomni.progress["tomo_start_time"] = "2026-10-24T10:00:00"
+    flomni.progress["subtomo"] = 3
+    flomni.progress["angle"] = 45.0
+    monkeypatch.setattr(Flomni, "tomo_type", 1, raising=False)
+    calls = []
+    flomni.tomo_scan = lambda **kw: calls.append(kw)
+    flomni._confirm_resume_outside_queue = lambda: True
+
+    flomni.tomo_scan_resume()
+
+    assert calls == [{"subtomo_start": 3, "start_angle": 45.0}]
+
+
+def test_progress_reset_clears_completed():
+    flomni = _make_flomni()
+    flomni.progress["tomo_completed"] = True
+    flomni.progress.reset()
+    assert flomni.progress.get("tomo_completed") is False
+
+
+class _Timeout(flomni_module.AlarmBase):
+    def __init__(self):
+        Exception.__init__(self, "timeout")
+        self.alarm_type = "TimeoutError"
+
+
+class _Queue:
+    def __init__(self):
+        self.next_scan_number = 100
+        self.resets = 0
+
+    def request_queue_reset(self):
+        self.resets += 1
+
+
+def test_alignment_scan_repeats_an_angle_after_timeout(monkeypatch, tmp_path):
+    flomni = _make_flomni()
+    queue = _Queue()
+    fake_dev = types.SimpleNamespace(
+        rtx=types.SimpleNamespace(
+            controller=types.SimpleNamespace(feedback_is_running=lambda: True)
+        ),
+        fsamroy=object(),
+    )
+    monkeypatch.setitem(builtins.__dict__, "bec", types.SimpleNamespace(queue=queue))
+    monkeypatch.setitem(builtins.__dict__, "dev", fake_dev)
+    monkeypatch.setattr(flomni_module, "umv", lambda *a: None, raising=False)
+    monkeypatch.setattr(flomni_module, "filters_out_of_beam", lambda dev: (True, []))
+    monkeypatch.setattr(flomni_module.time, "sleep", lambda s: None)
+    monkeypatch.setattr(Flomni, "sample_name", "test", raising=False)
+    monkeypatch.setenv("HOME", str(tmp_path))
+    flomni.get_alignment_offset = lambda angle: (1.0, 0.0, 0.0)
+    flomni._check_eye_out_and_optics_in = lambda: True
+    flomni.OMNYTools = types.SimpleNamespace(printgreenbold=lambda msg: None)
+    flomni._scilog_write = lambda *a: True
+    angles = []
+
+    def projection(angle, _internal=False):
+        angles.append(angle)
+        queue.next_scan_number += 1  # one scan per attempt (the timed-out one included)
+        if angle == 90 and angles.count(90) == 1:
+            raise _Timeout()
+
+    flomni.tomo_scan_projection = projection
+
+    flomni.tomo_alignment_scan()
+
+    assert angles == [0, 45, 90, 90, 135, 180]
+    assert queue.resets == 1
+    lines = (tmp_path / "data/raw/logs/ptychotomoalign_scannum.txt").read_text().splitlines()
+    assert lines[0] == "100 101 103 104 105"  # the accepted scan per angle, not 100..104
+
+
+@pytest.mark.parametrize(
+    "readback, ok", [(-1.14, True), (-1.30, True), (-1.40, False), (-0.80, False)]
+)
+def test_heater_down_needs_fsamx_in_within_0_2_mm(monkeypatch, readback, ok):
+    flomni = _make_flomni()
+    fsamx = types.SimpleNamespace(
+        user_parameter={"in": -1.14}, readback=types.SimpleNamespace(get=lambda: readback)
+    )
+    fheater = types.SimpleNamespace(
+        _config={"deviceConfig": {"axis_Id": "H"}}, user_parameter={"in": 5}
+    )
+    monkeypatch.setitem(
+        builtins.__dict__, "dev", types.SimpleNamespace(fsamx=fsamx, fheater=fheater)
+    )
+    monkeypatch.setattr(flomni_module, "dev", builtins.__dict__["dev"], raising=False)
+    moved = []
+    monkeypatch.setattr(flomni_module, "umv", lambda *a: moved.append(a), raising=False)
+    flomni.ensure_osa_back = lambda: None
+    if ok:
+        flomni.move_fheater_down()
+        assert moved
+    else:
+        with pytest.raises(FlomniError):
+            flomni.move_fheater_down()
+        assert not moved
+
+
+class _DevContainer(dict):
+    """dict keyed by device name, like bec_lib's DeviceContainer (attribute access too)."""
+
+    def __getattr__(self, name):
+        return self[name]
+
+
+def test_xray_eye_ctrl_c_returns_fsamx_in(monkeypatch):
+    """The KeyboardInterrupt handler passed the device instead of its name: dev[device] raised."""
+    import csaxs_bec.bec_ipython_client.plugins.flomni.flomni_optics_mixin as optics_module
+
+    flomni = _make_flomni()
+    fsamx = types.SimpleNamespace(
+        user_parameter={"in": -1.14}, readback=types.SimpleNamespace(get=lambda: -1.3)
+    )
+    fake_dev = _DevContainer(fsamx=fsamx)
+    monkeypatch.setitem(builtins.__dict__, "dev", fake_dev)
+    monkeypatch.setattr(flomni_module, "dev", fake_dev, raising=False)
+    monkeypatch.setattr(optics_module, "dev", fake_dev, raising=False)
+    moved = []
+    monkeypatch.setattr(flomni_module, "umv", lambda *a: moved.append(a), raising=False)
+    flomni.OMNYTools = types.SimpleNamespace(yesno=lambda *a, **k: True)
+    flomni.feedback_disable = lambda: None
+
+    class _Align:
+        def __init__(self, *a):
+            pass
+
+        def align(self, keep_shutter_open):
+            raise KeyboardInterrupt
+
+    monkeypatch.setattr(flomni_module, "XrayEyeAlign", _Align)
+
+    with pytest.raises(KeyboardInterrupt):
+        flomni.start_x_ray_eye_alignment()
+    assert moved == [(fsamx, -1.14)]
+
+
+def test_tray_position_message_says_0_to_20():
+    flomni = _make_flomni()
+    flomni.check_position_is_valid(20)
+    with pytest.raises(FlomniError, match="0-20"):
+        flomni.check_position_is_valid(21)
+
+
+def test_get_val_reasks_angle_range_and_bad_numbers(monkeypatch, capsys):
+    answers = iter(["270", "abc", "360"])
+    monkeypatch.setattr(builtins, "input", lambda prompt="": next(answers))
+    assert Flomni._get_val("Angular range", 180, int, choices=(180, 360)) == 360
+    out = capsys.readouterr().out
+    assert "one of 180, 360" in out and "Not a valid int" in out
+
+
+def test_feye_out_raises_when_direct_beam_still_visible(monkeypatch):
+    import csaxs_bec.bec_ipython_client.plugins.flomni.flomni_optics_mixin as optics_module
+
+    flomni = _make_flomni()
+    flomni.csaxs = None
+    flomni.foptics_in = lambda: None
+    flomni.flomnigui_show_xeyealign = lambda: None
+    flomni.xrayeye_update_frame = lambda: None
+    flomni.OMNYTools = types.SimpleNamespace(yesno=lambda *a, **k: False)
+    fake_dev = types.SimpleNamespace(fsh=types.SimpleNamespace(fshclose=lambda: None))
+    monkeypatch.setattr(optics_module, "dev", fake_dev, raising=False)
+    monkeypatch.setitem(builtins.__dict__, "dev", fake_dev)
+    with pytest.raises(optics_module.FlomniOpticsError, match="direct beam"):
+        flomni.feye_out()
+
+
+class _SlowStream:
+    """Non-blocking stdin stand-in: returns the queued chunks one per read, None in between."""
+
+    def __init__(self, chunks):
+        self.chunks = list(chunks)
+
+    def read(self, n):
+        return self.chunks.pop(0) if self.chunks else None
+
+
+def test_read_escape_suffix_waits_for_split_sequence():
+    from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.omny_general_tools import (
+        read_escape_suffix,
+    )
+
+    assert read_escape_suffix(_SlowStream(["[", None, "A"])) == "[A"
+    assert read_escape_suffix(_SlowStream([])) is None  # ESC alone

tests/tests_bec_ipython_client/test_flomni_special_angles.py

@@ -32,6 +32,7 @@ def _make_flomni(monkeypatch):
     obj.tomo_angle_range = 180
     obj.special_angles = []
     obj.special_angle_repeats = 3
+    obj.special_angle_tolerance = 0.5
     obj._current_special_angles = []
     obj._print_progress = lambda: None
     obj._format_duration = lambda seconds: "0s"
@@ -86,3 +87,34 @@ def test_special_angle_repeats_survive_a_retry(monkeypatch):
 
     assert len(calls) == 1 + 3, "the retry must still repeat the special angle"
     assert flomni._current_special_angles == []
+
+
+def test_special_angle_tolerance_is_used(monkeypatch):
+    """special_angle_tolerance was ignored (hard-coded 0.5) before 2026-10-24."""
+    flomni = _make_flomni(monkeypatch)
+    flomni.special_angles = [90.0]
+    flomni._current_special_angles = [90.0]
+    flomni.special_angle_tolerance = 2.0
+    calls = []
+    flomni._at_each_angle = lambda angle: calls.append(angle)
+
+    flomni._tomo_scan_at_angle(91.5, 1)
+
+    assert calls == [91.5] * 3
+    assert flomni._current_special_angles == []
+
+
+def test_special_angles_match_in_any_order(monkeypatch):
+    """An angle not reached first (e.g. a reverse sub-tomogram) no longer blocks the later ones."""
+    flomni = _make_flomni(monkeypatch)
+    flomni.special_angles = [30.0, 120.0]
+    flomni._current_special_angles = [30.0, 120.0]
+    calls = []
+    flomni._at_each_angle = lambda angle: calls.append(angle)
+
+    flomni._tomo_scan_at_angle(120.0, 1)
+    flomni._tomo_scan_at_angle(120.0, 1)  # used once only
+    flomni._tomo_scan_at_angle(30.0, 1)
+
+    assert calls == [120.0] * 3 + [120.0] + [30.0] * 3
+    assert flomni._current_special_angles == []

tests/tests_bec_ipython_client/test_flomni_tomo_angles.py

@@ -681,7 +681,11 @@ def test_new_scan_resets_progress_before_eye_optics_gate_declines(monkeypatch, t
         "must not acquire -- the gate declined"
     )
 
-    flomni.tomo_scan()  # returns early, no exception
+    from csaxs_bec.bec_ipython_client.plugins.flomni.flomni import FlomniError
+
+    # raises since 2026-10-24 (a plain return let the tomo queue mark the job "done")
+    with pytest.raises(FlomniError, match="not in measurement condition"):
+        flomni.tomo_scan()
 
     assert flomni.progress["projection"] == 0
     assert flomni.progress["total_projections"] == 1
@@ -765,3 +769,24 @@ def test_flomni_gui_progress_survives_zero_total(monkeypatch):
 
     assert ring_values == [0.0, 50.0]
     assert "Angles from file" in labels[0]
+
+
+@pytest.mark.parametrize("zero_deg_reference", [False, True])
+def test_type1_new_scan_takes_two_zero_deg_projections_at_the_start(monkeypatch, zero_deg_reference):
+    """The fixed 0-deg shot + sub-tomogram 1's first projection (0 deg); the damage-check reference
+    before sub-tomogram 1 would have been a third one in a row (removed 2026-10-24)."""
+    flomni = make_flomni_for_golden_resume(monkeypatch, tomo_type=1)
+    flomni.tomo_angle_range = 180
+    flomni.tomo_angle_stepsize = 45.0
+    flomni.tomo_type1_always_forward = False
+    flomni.zero_deg_reference_at_each_subtomo = zero_deg_reference
+    calls = []
+    flomni._tomo_scan_at_angle = lambda angle, subtomo, **kw: calls.append((float(angle), subtomo))
+
+    flomni.tomo_scan()
+
+    assert calls[:2] == [(0.0, 1), (0.0, 1)]
+    assert calls[2][0] != 0.0
+    # sub-tomogram 3 (forward, starts near 0) still gets its reference shot when enabled
+    sub3 = [c for c in calls if c[1] == 3]
+    assert (sub3[0][0] == 0.0) == zero_deg_reference

tests/tests_devices/test_flomni_loop_settings.py

@@ -0,0 +1,108 @@
+"""Tests for the baseline device flomni_loop_settings (devices/omny/flomni_loop_settings.py): the loop settings of
+the flOMNI web HMI stored with every scan. A small local HTTP server stands in for the web HMI (/api/settings)."""
+
+import json
+import math
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+import pytest
+
+from csaxs_bec.devices.omny import flomni_loop_settings as fls
+from csaxs_bec.devices.omny.flomni_loop_settings import SETTINGS, FlomniLoopSettings
+
+TOKEN = "device-token-0123456789abcdef"
+
+SNAP = {
+    "instrument": "flOMNI", "definition": "/home/control/OMNY/flOMNI/FeedbackLoop/loopsettings/flomni.json",
+    "definition_sha1": "a" * 40, "scan": {"mode": 0, "active": False, "text": ""},
+    "settings": [
+        {"name": "fzp_enable", "value": 1, "unsaved": False, "consistent": True},
+        {"name": "fzp_range_V_x", "value": 1.5, "unsaved": True, "consistent": True},
+        {"name": "trigger_tolerance_nm", "value": 10.0, "unsaved": False, "consistent": False},
+        {"name": "new_setting_later", "value": 3.0, "unsaved": False, "consistent": True},
+    ],
+    "rules": [{"name": "r1", "severity": "error", "ok": False}, {"name": "r2", "severity": "warning", "ok": False}],
+}
+
+
+class _Handler(BaseHTTPRequestHandler):
+    hits = 0
+
+    def log_message(self, *args):
+        pass
+
+    def do_GET(self):
+        _Handler.hits += 1
+        code, body = (200, SNAP) if self.headers.get("Authorization") == "Bearer " + TOKEN else (401, {"detail": "no"})
+        data = json.dumps(body).encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+
+@pytest.fixture
+def url():
+    _Handler.hits = 0
+    srv = HTTPServer(("127.0.0.1", 0), _Handler)
+    threading.Thread(target=srv.serve_forever, daemon=True).start()
+    yield f"http://127.0.0.1:{srv.server_address[1]}"
+    srv.shutdown()
+
+
+def test_read_stores_settings(url):
+    dev = FlomniLoopSettings(name="flomni_loop_settings", url=url, token=TOKEN)
+    r = dev.read()
+    assert r["flomni_loop_settings_available"]["value"] == 1
+    assert r["flomni_loop_settings_fzp_enable"]["value"] == 1.0
+    assert r["flomni_loop_settings_fzp_range_V_x"]["value"] == 1.5
+    assert math.isnan(r["flomni_loop_settings_move_beyond_um_x"]["value"])        # not in the answer
+    assert r["flomni_loop_settings_unsaved"]["value"] == 1
+    assert r["flomni_loop_settings_inconsistent"]["value"] == 1
+    assert r["flomni_loop_settings_rules_violated"]["value"] == 1                  # error rules only
+    assert r["flomni_loop_settings_definition"]["value"].endswith("sha1 " + "a" * 40)
+    assert "new_setting_later" in r["flomni_loop_settings_message"]["value"]
+    snap = json.loads(r["flomni_loop_settings_snapshot_json"]["value"])
+    assert snap["settings"]["fzp_enable"]["value"] == 1 and "profiles" not in snap
+    assert snap["settings"]["new_setting_later"]["value"] == 3.0                  # later settings: in the JSON
+    assert [r["name"] for r in snap["rules_violated"]] == ["r1", "r2"]
+    assert set(SETTINGS) <= {k.replace("flomni_loop_settings_", "") for k in r}
+
+
+def test_unavailable_does_not_raise_and_backs_off(url, monkeypatch):
+    dev = FlomniLoopSettings(name="ls", url=url, token="wrong-token-xxxxxxxxxxxx")
+    r = dev.read()
+    assert r["ls_available"]["value"] == 0 and "401" in r["ls_message"]["value"]
+    assert math.isnan(r["ls_fzp_enable"]["value"]) and r["ls_unsaved"]["value"] == -1
+    hits = _Handler.hits
+    dev.read()                                   # within the back-off: no new request
+    assert _Handler.hits == hits
+    monkeypatch.setattr(fls, "RETRY_AFTER_FAILURE_S", 0.0)
+    dev.read()
+    assert _Handler.hits == hits + 1
+
+
+def test_unreachable():
+    dev = FlomniLoopSettings(name="ls", url="http://127.0.0.1:9", token=TOKEN, timeout=0.5)
+    r = dev.read()
+    assert r["ls_available"]["value"] == 0 and "not reachable" in r["ls_message"]["value"]
+
+
+def test_token_file_and_snapshot(url, tmp_path):
+    tok = tmp_path / "token"
+    tok.write_text(TOKEN + "\n")
+    dev = FlomniLoopSettings(name="ls", url=url, token_file=str(tok))
+    assert dev.snapshot()["fzp_range_V_x"] == 1.5
+
+
+def test_api_methods_for_realtime_system(url):
+    """api_get / api_post: the RPC used by flomni.realtime_system (token only in the device config)."""
+    dev = FlomniLoopSettings(name="ls", url=url, token=TOKEN)
+    assert dev.api_get("/api/settings")["instrument"] == "flOMNI"
+    with pytest.raises(Exception, match="not allowed"):
+        dev.api_get("/api/params")                 # only the loop-settings endpoints
+    with pytest.raises(Exception, match="not allowed"):
+        dev.api_post("/api/loop/stop", {})
+    assert "api_get" in FlomniLoopSettings.USER_ACCESS and "set" not in FlomniLoopSettings.USER_ACCESS

tests/tests_devices/test_rt_flomni.py

@@ -4,6 +4,7 @@ import pytest
 from ophyd_devices.tests.utils import SocketMock
 
 from csaxs_bec.devices.omny.rt import RtFlomniController, RtFlomniMotor
+from csaxs_bec.devices.omny.rt.rt_flomni_ophyd import RtFlomniFlyer
 from csaxs_bec.devices.omny.rt.rt_ophyd import RtError
 
 
@@ -98,3 +99,31 @@ def test_feedback_enable_without_reset_raises(rt_flomni):
                 rt_flomni.feedback_enable_without_reset()
                 laser_tracker_on.assert_called_once()
                 assert mock.call(b"l3\n") in rt_flomni.sock.put.mock_calls
+
+
+def test_flyer_reports_average_stdev_to_the_client(rt_flomni):
+    flyer = RtFlomniFlyer(name="rt_positions", socket_cls=SocketMock, device_manager=rt_flomni.device_manager)
+    flyer.controller = rt_flomni
+    for row in (["0", "0", "1", "1", "0.002", "2", "2", "0.001", "0", "0"],
+                ["1", "0", "1", "1", "0.004", "2", "2", "0.003", "0", "0"]):
+        flyer._get_signals_from_table(row)
+    flyer.configure({"exp_time": 0.1, "frames_per_trigger": 2})        # kickoff parameters of the scan
+    flyer._t_kickoff, flyer._t_scan_end = 100.0, 101.0
+    flyer._report_statistics(2)
+    send = rt_flomni.device_manager.connector.send_client_info
+    msg = send.call_args.args[0]
+    assert "x 3.0 nm, y 2.0 nm" in msg and "2 points" in msg
+    assert "scan time 1.0 s (500 ms/point)" in msg and "exposure 0.4 s (40 %" in msg and "overhead 0.6 s (300 ms/point" in msg
+    assert send.call_args.kwargs["show_asap"] is True
+    send.reset_mock()
+    flyer._report_statistics(0)                    # no points: nothing sent, no ZeroDivisionError
+    send.assert_not_called()
+
+
+def test_format_scan_timing():
+    from csaxs_bec.devices.omny.rt.rt_ophyd import format_scan_timing
+
+    assert format_scan_timing(0, 0, 1) == "" and format_scan_timing(5, None, 1) == ""
+    assert format_scan_timing(10, 0, 2.0) == "; scan time 2.0 s (200 ms/point)"     # no exp_time: no split
+    txt = format_scan_timing(10, 0, 2.0, 0.1, 1)
+    assert "exposure 1.0 s (50 %" in txt and "overhead 1.0 s (100 ms/point" in txt

tests/tests_devices/test_rt_lamni_statistics.py

@@ -0,0 +1,26 @@
+from unittest import mock
+
+from ophyd_devices.tests.utils import SocketMock
+
+from csaxs_bec.devices.omny.rt.rt_lamni_ophyd import RtLamniController, RtLamniFlyer
+
+
+def test_lamni_flyer_reports_average_stdev_to_the_client():
+    RtLamniController._reset_controller()
+    flyer = RtLamniFlyer(name="rt_positions", socket_cls=SocketMock, device_manager=mock.MagicMock())
+    flyer.controller = mock.MagicMock()
+    for i, (sx, sy) in enumerate(((0.002, 0.001), (0.004, 0.003))):
+        row = ["0"] * 21
+        row[0], row[5], row[8], row[19] = str(i), str(sx), str(sy), "0.5"
+        flyer._get_signals_from_table(row)
+    flyer.configure({"exp_time": 0.05})                                  # frames_per_trigger default 1
+    flyer._t_kickoff, flyer._t_scan_end = 10.0, 10.5
+    flyer._report_statistics(2)
+    send = flyer.controller.device_manager.connector.send_client_info
+    msg = send.call_args.args[0]
+    assert "x 3.0 nm, y 2.0 nm" in msg and "2 points" in msg and "average angle 0.5000" in msg
+    assert "scan time 0.5 s (250 ms/point)" in msg and "exposure 0.1 s (20 %" in msg
+    send.reset_mock()
+    flyer._report_statistics(0)                    # no points: nothing sent, no ZeroDivisionError
+    send.assert_not_called()
+    RtLamniController._reset_controller()