diff --git a/csaxs_bec/bec_ipython_client/plugins/cSAXS/cSAXS.py b/csaxs_bec/bec_ipython_client/plugins/cSAXS/cSAXS.py index da747954..0be06df9 100644 --- a/csaxs_bec/bec_ipython_client/plugins/cSAXS/cSAXS.py +++ b/csaxs_bec/bec_ipython_client/plugins/cSAXS/cSAXS.py @@ -34,6 +34,9 @@ class cSAXS( self.diagnostics = cSAXSDiagnostics() super().__init__(client=client) + # EPICS does not keep the slit offsets reliably; restore them from config + self.slits_apply_epics_offsets() + # ------------------------------------------------------------------ # Undulator # ------------------------------------------------------------------ diff --git a/csaxs_bec/bec_ipython_client/plugins/cSAXS/slits.py b/csaxs_bec/bec_ipython_client/plugins/cSAXS/slits.py index 31246a6a..7c076e68 100644 --- a/csaxs_bec/bec_ipython_client/plugins/cSAXS/slits.py +++ b/csaxs_bec/bec_ipython_client/plugins/cSAXS/slits.py @@ -1,7 +1,9 @@ from __future__ import annotations import builtins -from typing import Optional +import datetime +import os +from typing import Any, Optional # ------------------------------------------------------------ # Make dev visible if running inside BEC @@ -29,6 +31,16 @@ class cSAXSSlits: "sl6": "Slit 6 (exposure box 2)", } + # device tags of the beamline sections recorded by the full state dump + # (endstation instruments such as LamNI/flOMNI/OMNY carry their own tags) + _BEAMLINE_TAGS = ("machine", "bl_frontend", "bl_optics_hutch", "bl_endstation") + + _STATE_DIR = "~/data/raw/logs/beamline_state" + + # userParameter holding the EPICS user_offset of a motor; the IOC does not + # keep the offsets reliably, so the client writes them at startup + _EPICS_OFFSET_PARAM = "epics_offset" + def __init__(self, client): self.client = client self.device_manager = client.device_manager @@ -83,14 +95,256 @@ class cSAXSSlits: print(f"{label}:") print( - f" x center = {self._sl_fmt(xc)} " - f"x size = {self._sl_fmt(xs)}" + f" {prefix}xc = {self._sl_fmt(xc)} " + f"{prefix}xs = {self._sl_fmt(xs)}" ) print( - f" y center = {self._sl_fmt(yc)} " - f"y size = {self._sl_fmt(ys)}" + f" {prefix}yc = {self._sl_fmt(yc)} " + f"{prefix}ys = {self._sl_fmt(ys)}" ) print("") print("==========================================") - print("") \ No newline at end of file + print("") + + def slits_apply_epics_offsets(self) -> dict: + """ + Write the configured EPICS user offsets (userParameter 'epics_offset'). + + Called at cSAXS client startup. Only devices that define + userParameter 'epics_offset' are touched; the offset is written only + if it differs from the one currently in EPICS. Nothing moves - the + user readback shifts with the offset. + + Returns: + dict: {device_name: offset} for every offset that was written. + """ + written = {} + for name, device in self.device_manager.devices.items(): + try: + param = (getattr(device, "_config", {}) or {}).get("userParameter") or {} + if self._EPICS_OFFSET_PARAM not in param: + continue + if not getattr(device, "enabled", True): + continue + offset = float(param[self._EPICS_OFFSET_PARAM]) + current = float(device.user_offset.get()) + if abs(current - offset) < 1e-9: + continue + device.user_offset.put(offset) + written[name] = offset + print(f"{name}: EPICS user offset {current:.6f} -> {offset:.6f} (from config)") + except Exception as exc: + print(f"{name}: could not apply {self._EPICS_OFFSET_PARAM}: {exc}") + return written + + # -------------------------------------------------------- + # slit calibration + # -------------------------------------------------------- + + def _sl_read_value(self, device_name: str, cached: bool = False) -> Optional[float]: + """ + Return the readback of a device (fresh read by default) or None. + """ + try: + device = self.device_manager.devices[device_name] + reading = device.read(cached=cached) + return float(reading[device_name]["value"]) + except Exception: + return None + + def _sl_offset_entry(self, name: str, target: float) -> dict: + """ + Compute the offset that makes the displayed value of slit axis + *name* equal to *target*, assuming the slit is physically where + *target* should be. + + EPICS motors (sl1-sl3): user = dir * dial + user_offset + SmarAct virtual slits (sl4-sl6): value = f(blades) + offset (config) + In both cases the displayed value shifts 1:1 with the offset, so + new_offset = offset + (target - displayed). + """ + entry = {"name": name, "target": target, "kind": None, "displayed": None, + "offset": None, "new_offset": None} + try: + device = self.device_manager.devices[name] + except Exception: + return entry + + config = getattr(device, "_config", {}) or {} + if "VirtualSlit" in str(config.get("deviceClass", "")): + entry["kind"] = "smaract" + entry["offset"] = float((config.get("deviceConfig") or {}).get("offset", 0) or 0) + else: + entry["kind"] = "epics" + try: + entry["offset"] = float(device.user_offset.get()) + except Exception: + entry["offset"] = None + configured = (config.get("userParameter") or {}).get(self._EPICS_OFFSET_PARAM) + entry["configured"] = None if configured is None else float(configured) + + entry["displayed"] = self._sl_read_value(name) + if entry["displayed"] is not None and entry["offset"] is not None: + entry["new_offset"] = entry["offset"] + (target - entry["displayed"]) + return entry + + @staticmethod + def _sl_fmt_value(value: Any) -> str: + if isinstance(value, bool) or value is None: + return str(value) + if isinstance(value, float): + return f"{value:.6g}" + text = repr(value) + return text if len(text) <= 100 else text[:97] + "..." + + def _sl_beamline_state_lines(self) -> list[str]: + """ + Dump readings and configuration signals (offsets, limits, ...) of + all enabled devices tagged with one of _BEAMLINE_TAGS. + """ + lines = [] + for tag in self._BEAMLINE_TAGS: + try: + devices = self.device_manager.devices.get_devices_with_tags(tag) + except Exception: + devices = [] + lines.append("") + lines.append(f"---------- {tag} ----------") + for device in devices: + if not getattr(device, "enabled", True): + continue + lines.append(f"{device.name}") + signals = {} + for fcn in (device.read, device.read_configuration): + try: + reading = fcn(cached=True) or fcn(cached=False) or {} + signals.update(reading) + except Exception as exc: + lines.append(f" <{fcn.__name__} failed: {exc}>") + for sig_name, sig in signals.items(): + value = sig.get("value") if isinstance(sig, dict) else sig + lines.append(f" {sig_name:40s} {self._sl_fmt_value(value)}") + return lines + + def slits_calibration_report( + self, + full: bool = True, + save: bool = False, + filename: Optional[str] = None, + target_width: float = 1.0, + ) -> dict: + """ + Slit offset calibration report (and optional full beamline state dump). + + Run with all slits physically centered on the beam and opened to + target_width (default 1 mm). Reports for every slit axis the offset + that makes the center read 0 and the size read target_width: + EPICS slits (sl1-sl3): new user_offset, plus commands to apply it. + SmarAct slits (sl4-sl6): new 'offset' for deviceConfig in bl_endstation.yaml. + Nothing is changed on the hardware or in the config files. + + Args: + full: also record all beamline devices (machine, frontend, optics + incl. mirrors/monos, endstation; not LamNI/flOMNI/OMNY). + save: write the report to a text file. + filename: output file; default ~/data/raw/logs/beamline_state/ + beamline_state_.txt + target_width: slit opening [mm] the slits are currently set to. + + Returns: + dict: {device_name: new_offset} for all slit center/size axes. + + Examples: + csaxs.slits_calibration_report() + csaxs.slits_calibration_report(save=True) + csaxs.slits_calibration_report(full=False) + """ + now = datetime.datetime.now() + lines = [ + "========== cSAXS Slit Calibration Report ==========", + f"date: {now:%Y-%m-%d %H:%M:%S}", + f"assumed state: all slits centered, opening {target_width:g} mm", + "", + f"{'axis':8s} {'type':8s} {'displayed':>11s} {'offset':>11s}" + f" {'target':>8s} {'new offset':>11s} {'change':>10s}", + ] + + entries = [] + for prefix, label in self._SLIT_LABELS.items(): + lines.append(f"{label}") + for suffix, target in (("xc", 0.0), ("xs", target_width), ("yc", 0.0), ("ys", target_width)): + entry = self._sl_offset_entry(f"{prefix}{suffix}", target) + entries.append(entry) + change = ( + None if entry["new_offset"] is None else entry["new_offset"] - entry["offset"] + ) + lines.append( + f"{entry['name']:8s} {str(entry['kind'] or '---'):8s}" + f" {self._sl_fmt(entry['displayed']):>11s} {self._sl_fmt(entry['offset']):>11s}" + f" {target:8.4f} {self._sl_fmt(entry['new_offset']):>11s}" + f" {self._sl_fmt(change):>10s}" + ) + if ( + suffix.endswith("s") + and entry["displayed"] is not None + and abs(entry["displayed"] - target) > 0.5 * target_width + ): + lines.append( + f" WARNING: {entry['name']} reads {entry['displayed']:.4f}, far" + f" from {target_width:g} mm - is this slit really set up?" + ) + lines.append("") + + missing = [e["name"] for e in entries if e["new_offset"] is None] + if missing: + lines.append(f"could not compute (device missing/unreadable): {', '.join(missing)}") + lines.append("") + + for e in entries: + configured = e.get("configured") + if configured is not None and e["offset"] is not None and abs(configured - e["offset"]) > 1e-9: + lines.append( + f"NOTE: {e['name']} EPICS offset {e['offset']:.6f} differs from configured" + f" {self._EPICS_OFFSET_PARAM} {configured:.6f}" + ) + lines.append( + f"EPICS slits - new userParameter '{self._EPICS_OFFSET_PARAM}'" + " (bl_frontend.yaml for sl1, bl_optics_hutch.yaml for sl2/sl3);" + ) + lines.append("written to EPICS at client startup or with csaxs.slits_apply_epics_offsets():") + for e in entries: + if e["kind"] == "epics" and e["new_offset"] is not None: + lines.append(f" {e['name']}: {self._EPICS_OFFSET_PARAM}: {e['new_offset']:.6f}") + lines.append("") + lines.append("EPICS slits - to apply immediately in this session:") + for e in entries: + if e["kind"] == "epics" and e["new_offset"] is not None: + lines.append(f" dev.{e['name']}.user_offset.put({e['new_offset']:.6f})") + lines.append("") + lines.append("SmarAct slits - new 'offset' in deviceConfig (bl_endstation.yaml):") + for e in entries: + if e["kind"] == "smaract" and e["new_offset"] is not None: + lines.append(f" {e['name']}: offset: {e['new_offset']:.6f}") + lines.append("") + + if full: + lines.append("========== Beamline state (cached readings) ==========") + lines.extend(self._sl_beamline_state_lines()) + lines.append("") + + text = "\n".join(lines) + print(text) + + if save: + if filename is None: + filename = os.path.join( + self._STATE_DIR, f"beamline_state_{now:%Y%m%d_%H%M%S}.txt" + ) + filename = os.path.expanduser(filename) + os.makedirs(os.path.dirname(filename) or ".", exist_ok=True) + with open(filename, "w", encoding="utf-8") as fh: + fh.write(text + "\n") + print(f"Report written to {filename}") + + return {e["name"]: e["new_offset"] for e in entries} diff --git a/tests/tests_bec_ipython_client/test_cSAXS_slits.py b/tests/tests_bec_ipython_client/test_cSAXS_slits.py index b381bfbd..20b38d5b 100644 --- a/tests/tests_bec_ipython_client/test_cSAXS_slits.py +++ b/tests/tests_bec_ipython_client/test_cSAXS_slits.py @@ -43,3 +43,158 @@ def test_slits_show_all_reads_every_slit_device_cached(capsys): for dev in devices.values(): dev.read.assert_called_once_with(cached=True) assert "cSAXS Slits Overview" in capsys.readouterr().out + + +def _slit_devices(displayed: dict, epics_offset=0.1, smaract_offset=0.2): + devices = {} + for prefix in cSAXSSlits._SLIT_LABELS: + smaract = prefix in ("sl4", "sl5", "sl6") + for suffix in ("xc", "xs", "yc", "ys"): + name = f"{prefix}{suffix}" + dev = mock.MagicMock() + dev.read.return_value = {name: {"value": displayed[suffix]}} + if smaract: + cls = "VirtualSlitWidth" if suffix.endswith("s") else "VirtualSlitCenter" + dev._config = { + "deviceClass": f"ophyd_devices.devices.virtual_slit.{cls}", + "deviceConfig": {"offset": smaract_offset}, + } + else: + dev._config = {"deviceClass": "ophyd_devices.EpicsMotorEC", "deviceConfig": {}} + dev.user_offset.get.return_value = epics_offset + devices[name] = dev + return devices + + +def test_slits_calibration_report_offsets(capsys): + displayed = {"xc": 0.03, "xs": 1.05, "yc": -0.02, "ys": 0.9} + slits = _make_slits(_slit_devices(displayed)) + + result = slits.slits_calibration_report(full=False) + + assert len(result) == 24 + # EPICS: new user_offset = old + (target - displayed) + assert abs(result["sl1xc"] - (0.1 - 0.03)) < 1e-12 + assert abs(result["sl2xs"] - (0.1 + 1.0 - 1.05)) < 1e-12 + assert abs(result["sl3yc"] - (0.1 + 0.02)) < 1e-12 + # SmarAct: new config offset = old + (target - displayed) + assert abs(result["sl4ys"] - (0.2 + 1.0 - 0.9)) < 1e-12 + assert abs(result["sl6xc"] - (0.2 - 0.03)) < 1e-12 + + out = capsys.readouterr().out + assert "dev.sl1xc.user_offset.put(0.070000)" in out + assert "sl4ys: offset: 0.300000" in out + assert "dev.sl4" not in out + assert "Beamline state" not in out + + +def test_slits_calibration_report_uses_fresh_reads_and_target_width(): + displayed = {"xc": 0.0, "xs": 2.0, "yc": 0.0, "ys": 2.0} + devices = _slit_devices(displayed, epics_offset=0.0, smaract_offset=0.0) + slits = _make_slits(devices) + + result = slits.slits_calibration_report(full=False, target_width=2.0) + + assert all(abs(v) < 1e-12 for v in result.values()) + devices["sl1xc"].read.assert_called_with(cached=False) + + +def test_slits_calibration_report_missing_device_and_save(tmp_path): + displayed = {"xc": 0.0, "xs": 1.0, "yc": 0.0, "ys": 1.0} + devices = _slit_devices(displayed) + del devices["sl5xc"] + slits = _make_slits(devices) + fname = tmp_path / "sub" / "state.txt" + + result = slits.slits_calibration_report(full=False, save=True, filename=str(fname)) + + assert result["sl5xc"] is None + text = fname.read_text() + assert "could not compute" in text and "sl5xc" in text + assert "Slit Calibration Report" in text + + +def test_slits_calibration_report_full_dump(capsys): + displayed = {"xc": 0.0, "xs": 1.0, "yc": 0.0, "ys": 1.0} + slit_devs = _slit_devices(displayed) + devices = mock.MagicMock() + devices.__getitem__.side_effect = slit_devs.__getitem__ + + kb = mock.MagicMock() + kb.name = "kbvpitch" + kb.enabled = True + kb.read.return_value = {"kbvpitch": {"value": 3.25}} + kb.read_configuration.return_value = {"kbvpitch_user_offset": {"value": -0.5}} + disabled = mock.MagicMock() + disabled.name = "disabled_dev" + disabled.enabled = False + devices.get_devices_with_tags.side_effect = lambda tag: ( + [kb, disabled] if tag == "bl_optics_hutch" else [] + ) + client = mock.MagicMock() + client.device_manager.devices = devices + slits = cSAXSSlits(client) + + slits.slits_calibration_report(full=True) + + out = capsys.readouterr().out + assert "---------- bl_optics_hutch ----------" in out + assert "kbvpitch_user_offset" in out and "-0.5" in out + assert "3.25" in out + assert "disabled_dev" not in out + kb.read.assert_called_once_with(cached=True) + tags = [c.args[0] for c in devices.get_devices_with_tags.call_args_list] + assert tags == list(cSAXSSlits._BEAMLINE_TAGS) + + +def _epics_motor(user_parameter: dict | None, live_offset: float, enabled=True): + dev = mock.MagicMock() + dev.enabled = enabled + dev._config = {"deviceClass": "ophyd_devices.EpicsMotorEC", "deviceConfig": {}} + if user_parameter is not None: + dev._config["userParameter"] = user_parameter + dev.user_offset.get.return_value = live_offset + return dev + + +def test_slits_apply_epics_offsets_only_where_configured_and_different(): + devices = { + "sl1xc": _epics_motor({"epics_offset": 0.12}, live_offset=0.0), + "sl1xs": _epics_motor({"epics_offset": 0.5}, live_offset=0.5), # already right + "sl2xc": _epics_motor({"other": 1}, live_offset=0.3), # no epics_offset + "sl2xs": _epics_motor(None, live_offset=0.3), # no userParameter + "sl3xc": _epics_motor({"epics_offset": -1.0}, live_offset=0.0, enabled=False), + } + slits = _make_slits(devices) + + written = slits.slits_apply_epics_offsets() + + assert written == {"sl1xc": 0.12} + devices["sl1xc"].user_offset.put.assert_called_once_with(0.12) + for name in ("sl1xs", "sl2xc", "sl2xs", "sl3xc"): + devices[name].user_offset.put.assert_not_called() + + +def test_slits_apply_epics_offsets_survives_device_errors(capsys): + bad = _epics_motor({"epics_offset": 0.1}, live_offset=0.0) + bad.user_offset.get.side_effect = RuntimeError("no connection") + good = _epics_motor({"epics_offset": 0.2}, live_offset=0.0) + slits = _make_slits({"sl1xc": bad, "sl1yc": good}) + + written = slits.slits_apply_epics_offsets() + + assert written == {"sl1yc": 0.2} + assert "sl1xc: could not apply epics_offset" in capsys.readouterr().out + + +def test_slits_calibration_report_prints_epics_offset_param_and_mismatch(capsys): + displayed = {"xc": 0.03, "xs": 1.0, "yc": 0.0, "ys": 1.0} + devices = _slit_devices(displayed, epics_offset=0.1) + devices["sl1xc"]._config["userParameter"] = {"epics_offset": 0.25} + slits = _make_slits(devices) + + slits.slits_calibration_report(full=False) + + out = capsys.readouterr().out + assert "sl1xc: epics_offset: 0.070000" in out + assert "NOTE: sl1xc EPICS offset 0.100000 differs from configured epics_offset 0.250000" in out