bec/csaxs_bec
0
0 Fork 3
Code Issues 17 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

feat(flomni): baseline device flomni_loop_settings - the loop settings stored with every scan

Browse Source 
Decisions Mirko 2026-10-23: settings only (not all loop parameters), snapshot at the scan start, web HMI down ->
marked unavailable and the scan continues, flOMNI only for now, token part of the device configuration.

- devices/omny/flomni_loop_settings.py (new): FlomniLoopSettings, readoutPriority baseline (the Fermat scan reads
  the baseline devices in prepare_scan): one signal per setting (21, NaN when not available), available, message,
  unsaved / inconsistent counts, violated error rules, definition path + sha1, compact snapshot_json; never
  raises; after a failure no new attempt for 60 s (a tomogram does not wait for the timeout at every projection);
  show() / snapshot() for the user
- devices/omny/webhmi_client.py (new): the HTTP client (URL with port-80 fallback, token value or file, errors)
  shared by the device and flomni.realtime_system (which now uses it; behaviour unchanged)
- ptycho_flomni.yaml: flomni_loop_settings with token_file (the token value only in a config that is not
  committed), host mpc2844.psi.ch, timeout 2 s
- tests: tests_devices/test_flomni_loop_settings.py (stub web HMI: values, counts, unavailable / back-off,
  unreachable, token file); checked end-to-end against the web HMI with the flOMNI definition and fakecore (all
  21 settings filled)
- user guide: the settings are stored with every scan

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019aWcwG89wi7oVyrefukXX6
This commit is contained in:
Claude committed 2026-10-23 16:36:46 +00:00
1 parent 1806b721b1
commit e418f96a1e
6 files changed
+386 -86
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py
+7 -86
View File
@@ -20,17 +20,8 @@ Connection: the web HMI URL (default http://<rtx host>:8080, and http://<rtx hos
the loop machine, see flOMNI FeedbackLoop/loopsettings/README.md).
"""
import json
import os
import urllib.error
import urllib.request
DEFAULT_TOKEN_FILE = "~/.flomni_webhmi_token"
DEFAULT_PORT = 8080
class RealtimeSystemError(Exception):
"""A request was refused (rules, scan running) or the web HMI could not be reached."""
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
def _fmt(v):
@@ -41,82 +32,12 @@ def _fmt(v):
return str(v)
class FlomniRealtimeSystem:
def __init__(self, url=None, token_file=None, device_manager=None, timeout=5.0):
self._url = url
self._fallback = None # port-80 URL that answered when 8080 did not (kept for the session)
self._token_file = token_file
self._device_manager = device_manager
self.timeout = timeout
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."""
# ── connection ────────────────────────────────────────────────────────────
@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 = "mpc2844.psi.ch"
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):
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 RealtimeSystemError(
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 RealtimeSystemError(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 RealtimeSystemError as exc:
# default URL on port 8080 not reachable: the loop machine also serves the web HMI on port 80
# (nftables redirect 80 -> 8080, the port the PSI gateway passes); not for an explicit URL
if getattr(exc, "unreachable", False) 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": "BEC",
"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 RealtimeSystemError(f"web HMI {url}{path}: {exc.code} {detail}") from None
except (urllib.error.URLError, OSError) as exc:
err = RealtimeSystemError(f"web HMI {url} not reachable: {exc}")
err.unreachable = True
raise err from None
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)
# ── reading ───────────────────────────────────────────────────────────────
csaxs_bec/device_configs/ptycho_flomni.yaml
+20
View File
@@ -616,6 +616,26 @@ 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.
# 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.
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_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 !!!!!!!!!!!!!!!!
# ############################################################
csaxs_bec/devices/omny/flomni_loop_settings.py
+142
View File
@@ -0,0 +1,142 @@
"""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; 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:
devices/omny/webhmi_client.py.
"""
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",
)
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"]
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")
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()
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/webhmi_client.py
+116
View File
@@ -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")
docs/user/ptychography/flomni.md
+4
View File
@@ -800,6 +800,10 @@ them. The rules and the mapping onto the loop parameters are kept on the loop ma
control loop; `save_as_default()` writes them into the loop's parameter files
- `flomni.realtime_system.help()`
Every scan stores the loop settings it ran with: the baseline device `flomni_loop_settings` (one value per
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`).
tests/tests_devices/test_flomni_loop_settings.py
+97
View File
@@ -0,0 +1,97 @@
"""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