feat(ddg): add show_all() status command for the delay generators #320

Merged
holler merged 1 commits from feat/ddg_show_all into main 2026-09-15 13:11:49 +02:00
4 changed files with 133 additions and 1 deletions
@@ -219,6 +219,7 @@ class DDG1(PSIDeviceBase, DelayGeneratorCSAXS):
"get_move_start_latency",
"set_move_start_latency",
"prepare_mcs_on_trigger",
"show_all",
]
# TODO Consider using the 'fsh' device instead.
@@ -120,7 +120,12 @@ class DDG2(PSIDeviceBase, DelayGeneratorCSAXS):
"""
# Methods exposed to the BEC client, i.e. callable as dev.ddg2.<name>()
USER_ACCESS = ["get_readout_times", "set_readout_times", "effective_readout_times"]
USER_ACCESS = [
"get_readout_times",
"set_readout_times",
"effective_readout_times",
"show_all",
]
def __init__(
self,
@@ -18,6 +18,7 @@ from bec_lib.logger import bec_logger
from ophyd import Component as Cpt
from ophyd import Device, EpicsSignal, EpicsSignalRO, Kind, Signal
from ophyd_devices import StatusBase, SubscriptionStatus
from prettytable import PrettyTable
from typeguard import typechecked
from csaxs_bec.devices.epics.delay_generator_csaxs.error_registry import ERROR_CODES
@@ -764,6 +765,109 @@ class DelayGeneratorCSAXS(Device):
else:
return f"Unknown error code: {msg}"
def show_all(self) -> None:
"""Print the current delay pair, IO and trigger/burst settings.
Covers everything needed to reconstruct the timing configuration at a glance:
per-pair delay, width, channel references and IO levels, plus the trigger and
burst configuration.
"""
pair_channels: dict[DelayChannelNames, tuple[LiteralChannels, LiteralChannels]] = {
"ab": ("A", "B"),
"cd": ("C", "D"),
"ef": ("E", "F"),
"gh": ("G", "H"),
}
def _reference_name(value: int) -> str:
try:
return CHANNELREFERENCE(value).name
except ValueError:
return str(value)
def _polarity_name(value: int) -> str:
try:
return OUTPUTPOLARITY(value).name
except ValueError:
return str(value)
pairs = PrettyTable()
pairs.title = f"{self.name} delay pair settings"
pairs.field_names = [
"Pair",
"Ref (ch1)",
"Delay [s]",
"Ref (ch2)",
"Width [s]",
"Polarity",
"Amplitude [V]",
"Offset [V]",
]
pairs.add_row(
[
"T0",
"-",
"-",
"-",
"-",
_polarity_name(self.t0.polarity.get()),
self.t0.amplitude.get(),
self.t0.offset.get(),
]
)
for pair_name in pair_channels:
pair = getattr(self, pair_name)
pairs.add_row(
[
pair_name.upper(),
_reference_name(pair.ch1.reference.get()),
pair.delay.get(),
_reference_name(pair.ch2.reference.get()),
pair.width.get(),
_polarity_name(pair.io.polarity.get()),
pair.io.amplitude.get(),
pair.io.offset.get(),
]
)
print(pairs)
try:
trigger_source = TRIGGERSOURCE(self.trigger_source.get()).name
except ValueError:
trigger_source = str(self.trigger_source.get())
try:
burst_config = BURSTCONFIG(self.burst_config.get()).name
except ValueError:
burst_config = str(self.burst_config.get())
trigger = PrettyTable()
trigger.title = f"{self.name} trigger/burst settings"
trigger.field_names = [
"Trigger Source",
"Trigger Level [V]",
"Trigger Rate [Hz]",
"Trigger Holdoff [s]",
"Burst Mode",
"Burst Count",
"Burst Delay [s]",
"Burst Period [s]",
"Burst Config",
]
trigger.add_row(
[
trigger_source,
self.trigger_level.get(),
self.trigger_rate.get(),
self.trigger_holdoff.get(),
bool(self.burst_mode.get()),
self.burst_count.get(),
self.burst_delay.get(),
self.burst_period.get(),
burst_config,
]
)
print(trigger)
if __name__ == "__main__":
ddg = DelayGeneratorCSAXS(name="ddg", prefix="X12SA-CPCL-DDG1:")
@@ -149,6 +149,28 @@ def test_ddg_set_io_values(mock_ddg: DelayGeneratorCSAXS):
assert attr.nim_mode.get() == 1
def test_ddg_show_all(mock_ddg: DelayGeneratorCSAXS, capsys):
"""Smoke test: show_all() must run against a fresh device and print both tables."""
mock_ddg.set_delay_pairs(channel="ab", delay=0.1, width=0.2)
mock_ddg.set_channel_reference("A", CHANNELREFERENCE.T0)
mock_ddg.show_all()
out = capsys.readouterr().out
assert "delay pair settings" in out
assert "trigger/burst settings" in out
assert "AB" in out
assert "T0" in out
def test_ddg1_show_all_is_reachable_from_the_client():
"""USER_ACCESS is how BEC exposes a method over RPC as dev.ddg1.show_all()."""
assert "show_all" in DDG1.USER_ACCESS
def test_ddg2_show_all_is_reachable_from_the_client():
"""USER_ACCESS is how BEC exposes a method over RPC as dev.ddg2.show_all()."""
assert "show_all" in DDG2.USER_ACCESS
def test_ddg_set_delay_pairs(mock_ddg: DelayGeneratorCSAXS):
"""Test setting delay pairs."""
mock_ddg.set_delay_pairs(channel="ab", delay=0.1, width=0.2)