fix(nidaq): restore daq async priority and single-source the file streams
CI for debye_bec / test (push) Successful in 1m4s
CI for debye_bec / test (push) Successful in 1m4s
This commit is contained in:
wyzula_j committed
1 parent
2e127c7444
commit
7d47919855
6 files changed
+107
-61
No files matched your search
@@ -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,
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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]}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user