feat(flomni): flomni.realtime_system - loop settings through the web HMI; ConstEmitter emitter_* removed
The ConstEmitter is no longer in the flOMNI loop (flOMNI 773df74, branch fzp-move-beyond): its values are module parameters, set with their rules by the web HMI on mpc2844 (RT-system webhmi/loopsettings.py, REST /api/settings). - flomni_realtime_system.py (new): FlomniRealtimeSystem, flomni.realtime_system: show / get / check / set (refused with the reason, adjust=True for the dependent settings) / profiles / apply_profile / unsaved / save_as_default, shortcuts trigger_* and tracking_targets; stdlib HTTP, API token from ~/.flomni_webhmi_token (FLOMNI_WEBHMI_TOKEN_FILE), URL http://<rtx host>:8080 (FLOMNI_WEBHMI_URL). No logic of its own: the same rules as the web page. - flomni.py: flomni.realtime_system; laser_parameters_* and laser_parameters_tweak_tracking() through it; interferometer_trigger_* are aliases of realtime_system.trigger_* (renamed, Mirko: "interferometer" was confusing next to flomni.laser_*) - rt_flomni_ophyd.py: emitter_* methods and their USER_ACCESS entries removed (the e-commands now answer with an error line); move_xyz stays - tests: test_flomni_realtime_system.py against a small local HTTP server standing in for the web HMI Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019aWcwG89wi7oVyrefukXX6
This commit is contained in:
4 files changed
+439
-225
No files matched your search
@@ -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 (
|
||||
@@ -503,25 +504,27 @@ 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
|
||||
Arrow keys nudge the tracking targets (realtime_system tracking_target_y / _z) by a fixed step of
|
||||
0.01 (absolute set of current +/- step):
|
||||
left / right : target y -/+
|
||||
up / down : target z +/-
|
||||
@@ -542,6 +545,7 @@ class FlomniSampleTransferMixin:
|
||||
import tty
|
||||
|
||||
controller = dev.rtx.controller
|
||||
rts = self.realtime_system
|
||||
|
||||
if not controller.laser_tracker_check_enabled():
|
||||
print(
|
||||
@@ -554,11 +558,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,8 +597,7 @@ 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
|
||||
@@ -606,16 +605,16 @@ class FlomniSampleTransferMixin:
|
||||
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 +709,36 @@ 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.
|
||||
# detector trigger: renamed to flomni.realtime_system.trigger_* (2026-10-23, Mirko: "interferometer" was
|
||||
# confusing next to flomni.laser_*); the values are settings of the real-time loop (OnTargetTrigger parameters,
|
||||
# until 2026-10-23 ConstEmitter lines). The old names stay as aliases.
|
||||
def _renamed(self, old, new):
|
||||
print(f"flomni.{old} is now flomni.realtime_system.{new}")
|
||||
|
||||
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_show(self):
|
||||
"""Alias of flomni.realtime_system.trigger_show()."""
|
||||
self._renamed("interferometer_trigger_show()", "trigger_show()")
|
||||
self.realtime_system.trigger_show()
|
||||
|
||||
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()
|
||||
"""Alias of flomni.realtime_system.trigger_set_tolerance_nm()."""
|
||||
self._renamed("interferometer_trigger_set_tolerance_nm()", "trigger_set_tolerance_nm()")
|
||||
self.realtime_system.trigger_set_tolerance_nm(tolerance_nm)
|
||||
|
||||
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()
|
||||
"""Alias of flomni.realtime_system.trigger_set_hold_ms()."""
|
||||
self._renamed("interferometer_trigger_set_hold_ms()", "trigger_set_hold_ms()")
|
||||
self.realtime_system.trigger_set_hold_ms(hold_ms)
|
||||
|
||||
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()
|
||||
"""Alias of flomni.realtime_system.trigger_set_after_move()."""
|
||||
self._renamed("interferometer_trigger_set_after_move()", "trigger_set_after_move()")
|
||||
self.realtime_system.trigger_set_after_move(enable)
|
||||
|
||||
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))
|
||||
"""Alias of flomni.realtime_system.help()."""
|
||||
self._renamed("interferometer_trigger_help()", "help()")
|
||||
self.realtime_system.help()
|
||||
|
||||
def lights_off(self):
|
||||
self.device_manager.devices.fsamx.controller.lights_off()
|
||||
@@ -1819,6 +1773,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)
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
"""flomni.realtime_system: the operating settings of the flOMNI real-time loop (expert functions).
|
||||
|
||||
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: the web HMI URL (default http://<rtx host>:8080, 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).
|
||||
"""
|
||||
|
||||
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."""
|
||||
|
||||
|
||||
def _fmt(v):
|
||||
if v is None:
|
||||
return "-"
|
||||
if isinstance(v, float):
|
||||
return f"{v:.6g}"
|
||||
return str(v)
|
||||
|
||||
|
||||
class FlomniRealtimeSystem:
|
||||
def __init__(self, url=None, token_file=None, device_manager=None, timeout=5.0):
|
||||
self._url = url
|
||||
self._token_file = token_file
|
||||
self._device_manager = device_manager
|
||||
self.timeout = timeout
|
||||
|
||||
# ── connection ────────────────────────────────────────────────────────────
|
||||
|
||||
@property
|
||||
def url(self):
|
||||
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 _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):
|
||||
data = None if body is None else json.dumps(body).encode()
|
||||
req = urllib.request.Request(self.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 {self.url}{path}: {exc.code} {detail}") from None
|
||||
except (urllib.error.URLError, OSError) as exc:
|
||||
raise RealtimeSystemError(f"web HMI {self.url} not reachable: {exc}") from None
|
||||
|
||||
# ── 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(f"Loop settings {snap.get('instrument', '')} ({self.url}) - "
|
||||
+ (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) ────────────────────────────
|
||||
|
||||
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):
|
||||
"""Print the functions with a short explanation."""
|
||||
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)")
|
||||
@@ -45,17 +45,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 +393,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 +416,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
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
"""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()
|
||||
Reference in new issue
Block a user