feat(csaxs): slit calibration report and EPICS offset restore
- csaxs.slits_calibration_report(): with all slits centered at a known opening, computes the offsets that make centers read 0 and sizes read target_width (EPICS user_offset for sl1-3, virtual-slit config offset for sl4-6); optional full dump of all machine/frontend/optics/endstation devices (readings + config signals) and save to text file - csaxs.slits_apply_epics_offsets(): writes userParameter 'epics_offset' to the EPICS user_offset where configured and different; called at cSAXS client startup since the IOC does not keep the offsets reliably - slits_show_all() labels values with the motor names Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
x12saandClaude Opus 5.5 committed
1 parent
0830a71759
commit
c0ee9e0378
3 files changed
+418
-6
No files matched your search
@@ -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
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@@ -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("")
|
||||
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_<timestamp>.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}
|
||||
@@ -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
|
||||
Reference in new issue
Block a user