make diagnostics robust against missing devices; add multi-signal support for polarization diodes
This commit is contained in:
@@ -1,10 +1,23 @@
|
||||
from __future__ import annotations
|
||||
import builtins
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Access to dev (as before)
|
||||
# ------------------------------------------------------------
|
||||
if builtins.__dict__.get("dev") is not None:
|
||||
dev = builtins.__dict__.get("dev")
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Safe device access
|
||||
# ------------------------------------------------------------
|
||||
def _safe_dev(name):
|
||||
try:
|
||||
return getattr(dev, name)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Base classes
|
||||
# ============================================================
|
||||
@@ -19,15 +32,25 @@ class SingleSignalDiag:
|
||||
self._signal = signal_dev
|
||||
|
||||
def show_all(self):
|
||||
cfg = self._gain.get()
|
||||
print(self._label)
|
||||
print("-" * len(self._label))
|
||||
|
||||
if self._gain is None or self._signal is None:
|
||||
print("device not available in config\n")
|
||||
return
|
||||
|
||||
try:
|
||||
cfg = self._gain.get()
|
||||
except Exception as e:
|
||||
print(f"error accessing gain: {e}\n")
|
||||
return
|
||||
|
||||
try:
|
||||
val = self._signal.signal.get()
|
||||
sval = f"{val:.4f}"
|
||||
except Exception:
|
||||
sval = "N/A"
|
||||
|
||||
print(self._label)
|
||||
print("-" * len(self._label))
|
||||
print(f"gain : {cfg.gain:.0e}")
|
||||
print(f"coupling : {cfg.coupling}")
|
||||
print(f"speed : {cfg.speed}")
|
||||
@@ -35,9 +58,61 @@ class SingleSignalDiag:
|
||||
print("")
|
||||
print("hint:")
|
||||
print(f" gain : csaxs.diagnostics.{self._name}.gain(1e5)")
|
||||
print(f" raw : dev.{self._signal.name}.signal.get()")
|
||||
print(f" raw : dev.{getattr(self._signal, 'name', '?')}.signal.get()")
|
||||
print("")
|
||||
|
||||
def gain(self, value):
|
||||
if self._gain is None:
|
||||
print("gain device not available")
|
||||
return
|
||||
self._gain.set_gain(value)
|
||||
|
||||
|
||||
class MultiSignalDiag:
|
||||
"""Diagnostics for multiple signals sharing one gain."""
|
||||
|
||||
def __init__(self, name, label, gain_dev, signal_devs: dict):
|
||||
self._name = name
|
||||
self._label = label
|
||||
self._gain = gain_dev
|
||||
self._signals = signal_devs # dict: label -> device
|
||||
|
||||
def show_all(self):
|
||||
print(self._label)
|
||||
print("-" * len(self._label))
|
||||
|
||||
if self._gain is None or any(v is None for v in self._signals.values()):
|
||||
print("device not available in config\n")
|
||||
return
|
||||
|
||||
try:
|
||||
cfg = self._gain.get()
|
||||
except Exception as e:
|
||||
print(f"error accessing gain: {e}\n")
|
||||
return
|
||||
|
||||
print(f"gain : {cfg.gain:.0e}")
|
||||
print(f"coupling : {cfg.coupling}")
|
||||
print(f"speed : {cfg.speed}")
|
||||
print("readback:")
|
||||
|
||||
for key, sig in self._signals.items():
|
||||
try:
|
||||
val = sig.signal.get()
|
||||
sval = f"{val:.4f}"
|
||||
except Exception:
|
||||
sval = "N/A"
|
||||
print(f" {key:<10}: {sval}")
|
||||
|
||||
print("")
|
||||
print("hint:")
|
||||
print(f" gain : csaxs.diagnostics.{self._name}.gain(1e5)")
|
||||
print("")
|
||||
|
||||
def gain(self, value):
|
||||
if self._gain is None:
|
||||
print("gain device not available")
|
||||
return
|
||||
self._gain.set_gain(value)
|
||||
|
||||
|
||||
@@ -51,11 +126,20 @@ class BPMDiag:
|
||||
self._rb = rb_dev
|
||||
|
||||
def show_all(self):
|
||||
cfg = self._gain.get()
|
||||
rb = self._rb.get()
|
||||
|
||||
print(self._label)
|
||||
print("-" * len(self._label))
|
||||
|
||||
if self._gain is None or self._rb is None:
|
||||
print("device not available in config\n")
|
||||
return
|
||||
|
||||
try:
|
||||
cfg = self._gain.get()
|
||||
rb = self._rb.get()
|
||||
except Exception as e:
|
||||
print(f"error accessing device: {e}\n")
|
||||
return
|
||||
|
||||
print(f"gain : {cfg.gain:.0e}")
|
||||
print(f"coupling : {cfg.coupling}")
|
||||
print(f"speed : {cfg.speed}")
|
||||
@@ -66,9 +150,13 @@ class BPMDiag:
|
||||
print("")
|
||||
print("hint:")
|
||||
print(f" gain : csaxs.diagnostics.{self._name}.gain(1e6)")
|
||||
print(f" raw : dev.{self._rb.name}.pos_x.get()")
|
||||
print(f" raw : dev.{getattr(self._rb, 'name', '?')}.pos_x.get()")
|
||||
print("")
|
||||
|
||||
def gain(self, value):
|
||||
if self._gain is None:
|
||||
print("gain device not available")
|
||||
return
|
||||
self._gain.set_gain(value)
|
||||
|
||||
|
||||
@@ -84,49 +172,49 @@ class cSAXSDiagnostics:
|
||||
self.bpm_xbox1 = BPMDiag(
|
||||
name="bpm_xbox1",
|
||||
label="BPM XBox1 (OP hutch)",
|
||||
gain_dev=dev.gain_bpm_xbox1,
|
||||
rb_dev=dev.bpm_xbox1_slowrb,
|
||||
gain_dev=_safe_dev("gain_bpm_xbox1"),
|
||||
rb_dev=_safe_dev("bpm_xbox1_slowrb"),
|
||||
)
|
||||
|
||||
self.bpm_xbox2 = BPMDiag(
|
||||
name="bpm_xbox2",
|
||||
label="BPM XBox2 (ES hutch)",
|
||||
gain_dev=dev.gain_bpm_xbox2,
|
||||
rb_dev=dev.bpm_xbox2_slowrb,
|
||||
gain_dev=_safe_dev("gain_bpm_xbox2"),
|
||||
rb_dev=_safe_dev("bpm_xbox2_slowrb"),
|
||||
)
|
||||
|
||||
# Single-signal diagnostics
|
||||
self.bim = SingleSignalDiag(
|
||||
name="bim",
|
||||
label="BIM XBox3 (ES)",
|
||||
gain_dev=dev.gain_bim_xbox3,
|
||||
signal_dev=dev.bim_xbox3_slowrb,
|
||||
gain_dev=_safe_dev("gain_bim_xbox3"),
|
||||
signal_dev=_safe_dev("bim_xbox3_slowrb"),
|
||||
)
|
||||
|
||||
self.beamstop = SingleSignalDiag(
|
||||
name="beamstop",
|
||||
label="Beamstop diode (flight tube)",
|
||||
gain_dev=dev.gain_beamstop_diode,
|
||||
signal_dev=dev.beamstop_intensity,
|
||||
gain_dev=_safe_dev("gain_beamstop_diode"),
|
||||
signal_dev=_safe_dev("beamstop_intensity"),
|
||||
)
|
||||
|
||||
self.polarization = SingleSignalDiag(
|
||||
# Multi-signal (fixed your diode issue)
|
||||
self.polarization = MultiSignalDiag(
|
||||
name="polarization",
|
||||
label="Polarization diodes (XBox1)",
|
||||
gain_dev=dev.gain_diodes_xbox1,
|
||||
signal_dev=dev.diode_horizontal_xbox1_slowrb,
|
||||
gain_dev=_safe_dev("gain_diodes_xbox1"),
|
||||
signal_devs={
|
||||
"horizontal": _safe_dev("diode_horizontal_xbox1_slowrb"),
|
||||
"vertical": _safe_dev("diode_vertical_xbox1_slowrb"),
|
||||
},
|
||||
)
|
||||
|
||||
def show_all(self):
|
||||
print("CSAXS diagnostics")
|
||||
print("=================")
|
||||
print("")
|
||||
print("=================\n")
|
||||
|
||||
self.bpm_xbox1.show_all()
|
||||
print("")
|
||||
self.bpm_xbox2.show_all()
|
||||
print("")
|
||||
self.bim.show_all()
|
||||
print("")
|
||||
self.beamstop.show_all()
|
||||
print("")
|
||||
self.polarization.show_all()
|
||||
self.polarization.show_all()
|
||||
Reference in New Issue
Block a user