diff --git a/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_1.py b/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_1.py index 6bf5f793..bf1151d1 100644 --- a/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_1.py +++ b/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_1.py @@ -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. diff --git a/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_2.py b/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_2.py index 533aecd6..7b19dda8 100644 --- a/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_2.py +++ b/csaxs_bec/devices/epics/delay_generator_csaxs/ddg_2.py @@ -120,7 +120,12 @@ class DDG2(PSIDeviceBase, DelayGeneratorCSAXS): """ # Methods exposed to the BEC client, i.e. callable as dev.ddg2.() - 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, diff --git a/csaxs_bec/devices/epics/delay_generator_csaxs/delay_generator_csaxs.py b/csaxs_bec/devices/epics/delay_generator_csaxs/delay_generator_csaxs.py index a5cdeffc..5efcb5f7 100644 --- a/csaxs_bec/devices/epics/delay_generator_csaxs/delay_generator_csaxs.py +++ b/csaxs_bec/devices/epics/delay_generator_csaxs/delay_generator_csaxs.py @@ -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:") diff --git a/tests/tests_devices/test_delay_generator_csaxs.py b/tests/tests_devices/test_delay_generator_csaxs.py index 57d64be1..f06f290f 100644 --- a/tests/tests_devices/test_delay_generator_csaxs.py +++ b/tests/tests_devices/test_delay_generator_csaxs.py @@ -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)