diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py index 36c5ec29..bf19f6af 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py +++ b/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 ( @@ -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) diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py new file mode 100644 index 00000000..79eaaec0 --- /dev/null +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_realtime_system.py @@ -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://: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)") diff --git a/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py b/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py index afa49424..3ef464b1 100644 --- a/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py +++ b/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py @@ -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,,). @@ -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 diff --git a/tests/tests_bec_ipython_client/test_flomni_realtime_system.py b/tests/tests_bec_ipython_client/test_flomni_realtime_system.py new file mode 100644 index 00000000..16265a7a --- /dev/null +++ b/tests/tests_bec_ipython_client/test_flomni_realtime_system.py @@ -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()