feat(ddg): add show_all() status command for the delay generators #320
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user