feat(flomni): realtime_system goes through the device flomni_loop_settings (RPC); token in its device config

Mirko 2026-10-23: the token goes into the device config (behind the beamline firewall like the other devices); the
user commands call the BEC device in the background.

- FlomniLoopSettings: api_get / api_post (RPC for flomni.realtime_system; only /api/settings* and /api/status;
  10 s timeout, no back-off). Not named set / put: the BEC device server treats those as ophyd writes (refused on
  a readOnly device; checked in bec_server rpc_handler.py)
- FlomniRealtimeSystem: requests through dev.flomni_loop_settings when it exists and is enabled (errors from the
  device -> RealtimeSystemError), else direct HTTP as before; show() names the route
- ptycho_flomni.yaml: token field (empty: token_file is used)
- tests: RPC methods and path check; routing through a device and back to direct HTTP when it is disabled
  (checked against bec_server: deviceConfig keys are matched to the constructor parameters, device_manager is
  passed, the constructor does no network access)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019aWcwG89wi7oVyrefukXX6

csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py

@@ -14,10 +14,9 @@ web page (tab "Loop settings") and BEC always behave the same way.
 
 Changes are live only: the next start of the control loop uses the .xmp files again (save_as_default keeps them).
 
-Connection: the web HMI URL (default http://<rtx host>:8080, and http://<rtx host> (port 80, nftables redirect to
-8080 on the loop machine) when 8080 cannot be reached; env FLOMNI_WEBHMI_URL) and an API token in a file
-(default ~/.flomni_webhmi_token, env FLOMNI_WEBHMI_TOKEN_FILE; the same token as in the web HMI's api_token_file on
-the loop machine, see flOMNI FeedbackLoop/loopsettings/README.md).
+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
@@ -33,12 +32,36 @@ def _fmt(v):
 
 
 class FlomniRealtimeSystem(WebHmiClient):
-    """Connection (URL with port-80 fallback, token, errors): WebHmiClient (devices/omny/webhmi_client.py), shared
-    with the baseline device flomni_loop_settings."""
+    """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:
@@ -59,7 +82,7 @@ class FlomniRealtimeSystem(WebHmiClient):
         snap = self.read()
         groups = {g["name"]: g for g in snap["groups"]}
         sc = snap.get("scan") or {}
-        print(f"Loop settings {snap.get('instrument', '')} ({self.url}) - "
+        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:

csaxs_bec/device_configs/ptycho_flomni.yaml

@@ -620,13 +620,15 @@ flomni_temphum:
 # ############################################################
 # 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.
-# token_file: a file with the web HMI API token on the machine of the device server (mode 600). The token itself can
-# be given as "token: <value>" instead, but only in a config that is not committed - never put it in this file.
+# 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: ""
     token_file: ~/.flomni_webhmi_token
     timeout: 2.0
   enabled: true

csaxs_bec/devices/omny/flomni_loop_settings.py

@@ -16,9 +16,12 @@ Signals: one per setting (float, NaN when not available; names = the setting nam
     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; only in a config that is not committed) or token_file (path to a file
-with it, default ~/.flomni_webhmi_token), optional url / host / timeout (default 2 s). Connection details:
+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
@@ -45,7 +48,7 @@ RETRY_AFTER_FAILURE_S = 60.0    # web HMI not reachable: no new attempt for this
 
 
 class FlomniLoopSettings(Device):
-    USER_ACCESS = ["show", "snapshot"]
+    USER_ACCESS = ["show", "snapshot", "api_get", "api_post"]
 
     available = Cpt(Signal, value=0, kind=Kind.normal)
     message = Cpt(Signal, value="", kind=Kind.normal)
@@ -63,6 +66,9 @@ class FlomniLoopSettings(Device):
         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
 
@@ -120,6 +126,26 @@ class FlomniLoopSettings(Device):
         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

docs/user/ptychography/flomni.md

@@ -804,8 +804,8 @@ Every scan stores the loop settings it ran with: the baseline device `flomni_loo
 setting, `available`, unsaved / inconsistent counts, the definition's sha1, a JSON of all settings);
 `dev.flomni_loop_settings.show()` prints what a scan would store now.
 
-FZP settings can only be changed between scans. BEC needs the API token of the web HMI in `~/.flomni_webhmi_token`
-(local contact; flOMNI `FeedbackLoop/loopsettings/README.md`).
+FZP settings can only be changed between scans. The commands go through the device `flomni_loop_settings`, whose
+config holds the web HMI API token (local contact; flOMNI `FeedbackLoop/loopsettings/README.md`).
 
 ### Detector trigger and interferometer tuning
 

tests/tests_bec_ipython_client/test_flomni_realtime_system.py

@@ -165,3 +165,35 @@ def test_port_80_fallback(monkeypatch):
     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()

tests/tests_devices/test_flomni_loop_settings.py

@@ -95,3 +95,14 @@ def test_token_file_and_snapshot(url, tmp_path):
     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