diff --git a/debye_bec/devices/nidaq/nidaq.py b/debye_bec/devices/nidaq/nidaq.py index cea6fac..0707d58 100644 --- a/debye_bec/devices/nidaq/nidaq.py +++ b/debye_bec/devices/nidaq/nidaq.py @@ -184,49 +184,49 @@ class Nidaq(PSIDeviceBase, NidaqControl): # fmt: off ### Per-channel async groups: emitted names are channel-first (nidaq_ai0_mean) ### - ai0 = Cpt(DynamicSignal, name="ai0", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ai1 = Cpt(DynamicSignal, name="ai1", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ai2 = Cpt(DynamicSignal, name="ai2", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ai3 = Cpt(DynamicSignal, name="ai3", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ai4 = Cpt(DynamicSignal, name="ai4", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ai5 = Cpt(DynamicSignal, name="ai5", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ai6 = Cpt(DynamicSignal, name="ai6", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ai7 = Cpt(DynamicSignal, name="ai7", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci0 = Cpt(DynamicSignal, name="ci0", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci1 = Cpt(DynamicSignal, name="ci1", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci2 = Cpt(DynamicSignal, name="ci2", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci3 = Cpt(DynamicSignal, name="ci3", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci4 = Cpt(DynamicSignal, name="ci4", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci5 = Cpt(DynamicSignal, name="ci5", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci6 = Cpt(DynamicSignal, name="ci6", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci7 = Cpt(DynamicSignal, name="ci7", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci8 = Cpt(DynamicSignal, name="ci8", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci9 = Cpt(DynamicSignal, name="ci9", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci10 = Cpt(DynamicSignal, name="ci10", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci11 = Cpt(DynamicSignal, name="ci11", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci12 = Cpt(DynamicSignal, name="ci12", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci13 = Cpt(DynamicSignal, name="ci13", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci14 = Cpt(DynamicSignal, name="ci14", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci15 = Cpt(DynamicSignal, name="ci15", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci16 = Cpt(DynamicSignal, name="ci16", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - ci17 = Cpt(DynamicSignal, name="ci17", ndim=1, max_size=1000, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) - di0_max = Cpt(DynamicSignal, name="di0_max", ndim=1, max_size=1000, signals=["max"], async_update={"type": "add", "max_shape": [None]}) - di1_max = Cpt(DynamicSignal, name="di1_max", ndim=1, max_size=1000, signals=["max"], async_update={"type": "add", "max_shape": [None]}) - di2_max = Cpt(DynamicSignal, name="di2_max", ndim=1, max_size=1000, signals=["max"], async_update={"type": "add", "max_shape": [None]}) - di3_max = Cpt(DynamicSignal, name="di3_max", ndim=1, max_size=1000, signals=["max"], async_update={"type": "add", "max_shape": [None]}) - di4_max = Cpt(DynamicSignal, name="di4_max", ndim=1, max_size=1000, signals=["max"], async_update={"type": "add", "max_shape": [None]}) - energy = Cpt(DynamicSignal, name="energy", ndim=1, max_size=1000, signals=["energy"], async_update={"type": "add", "max_shape": [None]}) - enc = Cpt(DynamicSignal, name="enc", ndim=1, max_size=1000, signals=["enc"], async_update={"type": "add", "max_shape": [None]}) - smpl_abs = Cpt(DynamicSignal, name="smpl_abs", ndim=1, max_size=1000, signals=["smpl_abs"], async_update={"type": "add", "max_shape": [None]}) - smpl_fluo = Cpt(DynamicSignal, name="smpl_fluo", ndim=1, max_size=1000, signals=["smpl_fluo"], async_update={"type": "add", "max_shape": [None]}) - ref_abs = Cpt(DynamicSignal, name="ref_abs", ndim=1, max_size=1000, signals=["ref_abs"], async_update={"type": "add", "max_shape": [None]}) - cisum = Cpt(DynamicSignal, name="cisum", ndim=1, max_size=1000, signals=["cisum"], async_update={"type": "add", "max_shape": [None]}) - rle = Cpt(DynamicSignal, name="rle", ndim=1, max_size=1000, signals=["rle"], async_update={"type": "add", "max_shape": [None]}) - xas_timestamp = Cpt(DynamicSignal, name="xas_timestamp", ndim=1, max_size=1000, signals=["xas_timestamp"], async_update={"type": "add", "max_shape": [None]}) - xrd_timestamp = Cpt(DynamicSignal, name="xrd_timestamp", ndim=1, max_size=1000, signals=["xrd_timestamp"], async_update={"type": "add", "max_shape": [None]}) - xrd_angle = Cpt(DynamicSignal, name="xrd_angle", ndim=1, max_size=1000, signals=["xrd_angle"], async_update={"type": "add", "max_shape": [None]}) - xrd_energy = Cpt(DynamicSignal, name="xrd_energy", ndim=1, max_size=1000, signals=["xrd_energy"], async_update={"type": "add", "max_shape": [None]}) - xrd_ai0 = Cpt(DynamicSignal, name="xrd_ai0", ndim=1, max_size=1000, signals=["xrd_ai0"], async_update={"type": "add", "max_shape": [None]}) + ai0 = Cpt(DynamicSignal, name="ai0", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ai1 = Cpt(DynamicSignal, name="ai1", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ai2 = Cpt(DynamicSignal, name="ai2", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ai3 = Cpt(DynamicSignal, name="ai3", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ai4 = Cpt(DynamicSignal, name="ai4", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ai5 = Cpt(DynamicSignal, name="ai5", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ai6 = Cpt(DynamicSignal, name="ai6", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ai7 = Cpt(DynamicSignal, name="ai7", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci0 = Cpt(DynamicSignal, name="ci0", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci1 = Cpt(DynamicSignal, name="ci1", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci2 = Cpt(DynamicSignal, name="ci2", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci3 = Cpt(DynamicSignal, name="ci3", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci4 = Cpt(DynamicSignal, name="ci4", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci5 = Cpt(DynamicSignal, name="ci5", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci6 = Cpt(DynamicSignal, name="ci6", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci7 = Cpt(DynamicSignal, name="ci7", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci8 = Cpt(DynamicSignal, name="ci8", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci9 = Cpt(DynamicSignal, name="ci9", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci10 = Cpt(DynamicSignal, name="ci10", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci11 = Cpt(DynamicSignal, name="ci11", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci12 = Cpt(DynamicSignal, name="ci12", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci13 = Cpt(DynamicSignal, name="ci13", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci14 = Cpt(DynamicSignal, name="ci14", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci15 = Cpt(DynamicSignal, name="ci15", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci16 = Cpt(DynamicSignal, name="ci16", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + ci17 = Cpt(DynamicSignal, name="ci17", ndim=1, max_size=1000, saved=False, signals=["mean", "std_dev"], async_update={"type": "add", "max_shape": [None]}) + di0_max = Cpt(DynamicSignal, name="di0_max", ndim=1, max_size=1000, saved=False, signals=["max"], async_update={"type": "add", "max_shape": [None]}) + di1_max = Cpt(DynamicSignal, name="di1_max", ndim=1, max_size=1000, saved=False, signals=["max"], async_update={"type": "add", "max_shape": [None]}) + di2_max = Cpt(DynamicSignal, name="di2_max", ndim=1, max_size=1000, saved=False, signals=["max"], async_update={"type": "add", "max_shape": [None]}) + di3_max = Cpt(DynamicSignal, name="di3_max", ndim=1, max_size=1000, saved=False, signals=["max"], async_update={"type": "add", "max_shape": [None]}) + di4_max = Cpt(DynamicSignal, name="di4_max", ndim=1, max_size=1000, saved=False, signals=["max"], async_update={"type": "add", "max_shape": [None]}) + energy = Cpt(DynamicSignal, name="energy", ndim=1, max_size=1000, saved=False, signals=["energy"], async_update={"type": "add", "max_shape": [None]}) + enc = Cpt(DynamicSignal, name="enc", ndim=1, max_size=1000, saved=False, signals=["enc"], async_update={"type": "add", "max_shape": [None]}) + smpl_abs = Cpt(DynamicSignal, name="smpl_abs", ndim=1, max_size=1000, saved=False, signals=["smpl_abs"], async_update={"type": "add", "max_shape": [None]}) + smpl_fluo = Cpt(DynamicSignal, name="smpl_fluo", ndim=1, max_size=1000, saved=False, signals=["smpl_fluo"], async_update={"type": "add", "max_shape": [None]}) + ref_abs = Cpt(DynamicSignal, name="ref_abs", ndim=1, max_size=1000, saved=False, signals=["ref_abs"], async_update={"type": "add", "max_shape": [None]}) + cisum = Cpt(DynamicSignal, name="cisum", ndim=1, max_size=1000, saved=False, signals=["cisum"], async_update={"type": "add", "max_shape": [None]}) + rle = Cpt(DynamicSignal, name="rle", ndim=1, max_size=1000, saved=False, signals=["rle"], async_update={"type": "add", "max_shape": [None]}) + xas_timestamp = Cpt(DynamicSignal, name="xas_timestamp", ndim=1, max_size=1000, saved=False, signals=["xas_timestamp"], async_update={"type": "add", "max_shape": [None]}) + xrd_timestamp = Cpt(DynamicSignal, name="xrd_timestamp", ndim=1, max_size=1000, saved=False, signals=["xrd_timestamp"], async_update={"type": "add", "max_shape": [None]}) + xrd_angle = Cpt(DynamicSignal, name="xrd_angle", ndim=1, max_size=1000, saved=False, signals=["xrd_angle"], async_update={"type": "add", "max_shape": [None]}) + xrd_energy = Cpt(DynamicSignal, name="xrd_energy", ndim=1, max_size=1000, saved=False, signals=["xrd_energy"], async_update={"type": "add", "max_shape": [None]}) + xrd_ai0 = Cpt(DynamicSignal, name="xrd_ai0", ndim=1, max_size=1000, saved=False, signals=["xrd_ai0"], async_update={"type": "add", "max_shape": [None]}) # fmt: on USER_ACCESS = ["set_config"] @@ -319,6 +319,39 @@ class Nidaq(PSIDeviceBase, NidaqControl): # Annotate the alignment: these rows are in sync with the # monitored readouts (cf. MCSCardCSAXS acquisition_group logic). self._guarded_group_put(group, batch, acquisition_group="monitored") + self._send_sw_file_stream(buckets, now) + + def _send_sw_file_stream(self, buckets: dict[str, dict], timestamp: float) -> None: + """Mirror the software readings onto device_async_readback for the writer. + + The groups are declared saved=False: their device_async_signal streams + serve the GUI only, otherwise the writer would consume the same + channel-first names twice during hardware scans (backend push + bridge). + The file is therefore fed exclusively from device_async_readback — by + the backend in hardware mode and by this mirror in software mode. + """ + connector = getattr(self.device_manager, "connector", None) + if connector is None: + return + from bec_lib import messages + from bec_lib.endpoints import MessageEndpoints + + signals = {} + for group, batch in buckets.items(): + for key, data in batch.items(): + name = key if key.startswith(f"{self.name}_") else f"{self.name}_{group}_{key}" + signals[name] = {"value": data["value"], "timestamp": timestamp} + msg = messages.DeviceMessage( + signals=signals, metadata={"async_update": {"type": "add", "max_shape": [None]}} + ) + connector.xadd( + MessageEndpoints.device_async_readback( + scan_id=getattr(self.scan_info.msg, "scan_id", None), device=self.name + ), + {"data": msg}, + max_size=1000, + expire=600, + ) def _guarded_group_put( self, diff --git a/debye_bec/scans/nidaq_continuous_scan.py b/debye_bec/scans/nidaq_continuous_scan.py index d1afd99..6a5da9d 100644 --- a/debye_bec/scans/nidaq_continuous_scan.py +++ b/debye_bec/scans/nidaq_continuous_scan.py @@ -71,11 +71,7 @@ class NidaqContinuousScan(ScanBase): self.motors = [self.daq] self.update_scan_info(scan_duration=scan_duration, compression=compression) - # NOTE: the daq is deliberately NOT promoted to async readout priority: - # the file writer would then ALSO consume the backend's legacy - # device_async_readback stream, whose signal names are now identical to - # the bridged ones, double-writing every dataset. The unified-bridge - # emission on device_async_signal is stored for every scan regardless. + self.actions.set_device_readout_priority([self.daq], priority="async") @scan_hook def prepare_scan(self): diff --git a/debye_bec/scans/xas_simple_scan.py b/debye_bec/scans/xas_simple_scan.py index dad2973..d6a313f 100644 --- a/debye_bec/scans/xas_simple_scan.py +++ b/debye_bec/scans/xas_simple_scan.py @@ -80,11 +80,7 @@ class XasSimpleScan(ScanBase): scan_duration=scan_duration, monitored_readout_cycle=monitored_readout_cycle, ) - # NOTE: the daq is deliberately NOT promoted to async readout priority: - # the file writer would then ALSO consume the backend's legacy - # device_async_readback stream, whose signal names are now identical to - # the bridged ones, double-writing every dataset. The unified-bridge - # emission on device_async_signal is stored for every scan regardless. + self.actions.set_device_readout_priority([self.daq], priority="async") @scan_hook def prepare_scan(self): diff --git a/tests/tests_devices/test_nidaq.py b/tests/tests_devices/test_nidaq.py index 03f3ee2..9b83187 100644 --- a/tests/tests_devices/test_nidaq.py +++ b/tests/tests_devices/test_nidaq.py @@ -284,3 +284,30 @@ def test_device_manager_is_an_explicit_init_parameter(): import inspect assert "device_manager" in inspect.signature(Nidaq).parameters + + +def test_groups_are_gui_only_and_sw_mode_mirrors_to_writer_stream(mock_nidaq, scan_info_mock): + """saved=False keeps the writer off the group streams; software mode mirrors + the readings onto device_async_readback so the file has ONE source with the + channel-first names in both trigger modes.""" + dev = mock_nidaq + assert dev.ai0.saved is False + assert dev.energy.saved is False + + dev.device_manager = mock.MagicMock() + dev.ai0_epics._read_pv.mock_data = 5.0 + scan_info_mock.scan_name = "line_scan" + dev.scan_parameters = scan_info_mock + from types import SimpleNamespace as _NS + + with mock.patch.object(dev.scan_info, "msg", _NS(status="open", scan_id="swscan")): + dev._staged = Staged.yes + dev.read() + dev._staged = Staged.no + call = dev.device_manager.connector.xadd.call_args + assert "async_readback" in str(call.args[0]) + payload = call.args[1]["data"] + assert payload.signals["nidaq_ai0_mean"]["value"] == 5.0 + assert payload.signals["nidaq_di2_max"]["value"] is not None + assert payload.signals["nidaq_energy"]["value"] is not None + assert payload.metadata["async_update"] == {"type": "add", "max_shape": [None]} diff --git a/tests/tests_scans/test_mono_bragg_scans_v4.py b/tests/tests_scans/test_mono_bragg_scans_v4.py index fab5f6f..e33fe92 100644 --- a/tests/tests_scans/test_mono_bragg_scans_v4.py +++ b/tests/tests_scans/test_mono_bragg_scans_v4.py @@ -52,10 +52,7 @@ def test_xas_simple_scan_v4_prepare_scan_updates_metadata(v4_scan_assembler): np.testing.assert_array_equal(scan.scan_info.positions, np.array([8000.0, 9000.0])) assert scan.scan_info.additional_scan_parameters["scan_time"] == 1.0 assert scan.scan_info.additional_scan_parameters["scan_duration"] == 10.0 - # the daq is no longer promoted to async: with unified channel-first - # names the writer must consume ONLY the bridged device_async_signal - # streams, never the backend's legacy per-device stream in parallel - assert scan.scan_info.readout_priority_modification["async"] == [] + assert scan.scan_info.readout_priority_modification["async"] == ["nidaq"] scan.actions.add_scan_report_instruction_device_progress.assert_called_once_with(scan.motor) scan.actions.read_baseline_devices.assert_called_once_with(wait=False) assert scan._baseline_readout_status is baseline_status diff --git a/tests/tests_scans/test_nidaq_continous_scan.py b/tests/tests_scans/test_nidaq_continous_scan.py index 6679939..f2c10ec 100644 --- a/tests/tests_scans/test_nidaq_continous_scan.py +++ b/tests/tests_scans/test_nidaq_continous_scan.py @@ -41,10 +41,7 @@ def test_nidaq_continuous_scan_v4_prepare_scan_updates_metadata(v4_scan_assemble assert scan.scan_info.additional_scan_parameters["scan_duration"] == 10.0 assert scan.scan_info.additional_scan_parameters["compression"] is False - # the daq is no longer promoted to async: with unified channel-first - # names the writer must consume ONLY the bridged device_async_signal - # streams, never the backend's legacy per-device stream in parallel - assert scan.scan_info.readout_priority_modification["async"] == [] + assert scan.scan_info.readout_priority_modification["async"] == ["nidaq"] scan.actions.add_scan_report_instruction_device_progress.assert_called_once_with(scan.daq) scan.actions.read_baseline_devices.assert_called_once_with(wait=False) assert scan._baseline_readout_status is baseline_status