diff --git a/ophyd_devices/configs/ophyd_devices_simulation.yaml b/ophyd_devices/configs/ophyd_devices_simulation.yaml index 97fc756..1653c44 100644 --- a/ophyd_devices/configs/ophyd_devices_simulation.yaml +++ b/ophyd_devices/configs/ophyd_devices_simulation.yaml @@ -100,8 +100,8 @@ bpm5i: # Monitored device that ALSO exposes asynchronous signals (bec_widgets issue #1185). # Its readout priority is 'monitored', yet async_counts / async_spectrum / async_channels -# stream asynchronously, and 'morphing' is delivered both ways. softwareTrigger drives the -# async streams at every scan point. +# stream asynchronously (the signals push their own data via put()). softwareTrigger +# drives the async streams at every scan point. mixed_mon: readoutPriority: monitored deviceClass: ophyd_devices.sim.SimMonitorMixedSignals diff --git a/ophyd_devices/sim/sim_monitor.py b/ophyd_devices/sim/sim_monitor.py index f413139..482b1b6 100644 --- a/ophyd_devices/sim/sim_monitor.py +++ b/ophyd_devices/sim/sim_monitor.py @@ -1,6 +1,5 @@ """Module for simulated monitor devices.""" -import time from dataclasses import dataclass import numpy as np @@ -289,10 +288,11 @@ class SimMonitorMixedSignalsControl(Device): # Its serialized signal_class is 'ReadOnlySignal', so it is classified as sync. readback = Cpt(ReadOnlySignal, value=BIT_DEPTH(0), kind=Kind.hinted, compute_readback=True) - # --- Asynchronous signals (modern AsyncSignal family) ------------------- + # --- Asynchronous signals (AsyncSignal family) -------------------------- # These carry signal_class 'AsyncSignal' / 'AsyncMultiSignal' in the device info, # so a signal-aware classifier marks them async even though the parent device sits # in the 'monitored' readout-priority group (the core of bec_widgets issue #1185). + # As in SimWaveform, the signals push their own data to BEC via ``.put()``. async_counts = Cpt( AsyncSignal, ndim=0, max_size=1000, doc="Scalar counts streamed asynchronously." ) @@ -307,22 +307,12 @@ class SimMonitorMixedSignalsControl(Device): doc="Two scalar channels streamed asynchronously as one multi-signal.", ) - # --- Late-typed signal -------------------------------------------------- - # Declared as a plain (synchronous) SetableSignal, so its serialized signal_class - # is 'SetableSignal'. Outside a scan it behaves synchronously, but once a scan - # starts the device also streams it on the async readback endpoint. Its effective - # data-delivery type is therefore only known at the first reported value / - # beginning of the scan -- the case static, class-based classification cannot see. - morphing = Cpt(SetableSignal, value=BIT_DEPTH(0), kind=Kind.normal) - # --- Non-curve signal --------------------------------------------------- # role='progress' -> not curve data; a signal-aware classifier must ignore it. progress = Cpt(ProgressSignal, doc="Scan progress; not plotted as a curve.") # --- Config ------------------------------------------------------------- spectrum_size = Cpt(SetableSignal, value=200, kind=Kind.config) - # Whether the late-typed 'morphing' signal is streamed asynchronously during a scan. - stream_morphing_async = Cpt(SetableSignal, value=1, kind=Kind.config) SUB_READBACK = "readback" SUB_PROGRESS = "progress" @@ -333,7 +323,6 @@ class SimMonitorMixedSignalsControl(Device): self.device_manager = device_manager else: self.device_manager = bec_utils.DMMock() - self.connector = self.device_manager.connector self.sim_init = sim_init self.sim = self.sim_cls(parent=self, **kwargs) self._registered_proxies = {} @@ -354,23 +343,22 @@ class SimMonitorMixedSignalsControl(Device): class SimMonitorMixedSignals(PSIDeviceBase, SimMonitorMixedSignalsControl): """A simulated *monitored* device that mixes synchronous and asynchronous signals. - This device exists to reproduce and exercise bec_widgets issue #1185: a device whose - readout priority is ``monitored`` can still expose asynchronous signals. Curves must - therefore be classified per-signal (sync vs async), not by the parent device's - readout-priority group. + Reproduces and exercises bec_widgets issue #1185: a device whose readout priority is + ``monitored`` can still expose asynchronous signals, so curves must be classified + per-signal (sync vs async), not by the parent device's readout-priority group. + + Following the practice in :class:`~ophyd_devices.sim.sim_waveform.SimWaveform`, the + asynchronous signals push their own data through ``.put(..., async_update=...)``; the + device never assembles device messages by hand. The device server publishes those + ``BECMessageSignal`` puts to the async endpoint. Signals exposed: - * ``readback`` - synchronous, hinted; read at every scan point. (sync) - * ``async_counts`` - ``AsyncSignal`` (scalar); appended on every trigger. (async) - * ``async_spectrum`` - ``AsyncSignal`` (1D); the latest spectrum on every trigger. (async) - * ``async_channels`` - ``AsyncMultiSignal`` (ch1/ch2); appended on every trigger. (async) - * ``morphing`` - plain ``SetableSignal`` that is *also* streamed on the async - endpoint during a scan. Its data-delivery type is only revealed at the first - reported value / beginning of the scan -- a case static class-based classification - cannot resolve, and which class-based classification will (incorrectly) treat as - synchronous. Useful for validating any runtime/first-value reclassification. - * ``progress`` - ``ProgressSignal`` (role 'progress'); never a curve. + * ``readback`` - synchronous, hinted; read at every scan point. (sync) + * ``async_counts`` - ``AsyncSignal`` (scalar); a value appended on every trigger. (async) + * ``async_spectrum`` - ``AsyncSignal`` (1D); the latest spectrum on every trigger. (async) + * ``async_channels`` - ``AsyncMultiSignal`` (ch1/ch2); a value per channel each trigger. (async) + * ``progress`` - ``ProgressSignal`` (role 'progress'); never a curve. Intended config: ``readoutPriority: monitored`` and ``softwareTrigger: true`` so that the device is both read at every point (sync) and triggered to stream async data. @@ -394,16 +382,7 @@ class SimMonitorMixedSignals(PSIDeviceBase, SimMonitorMixedSignalsControl): sim_init=sim_init, **kwargs, ) - self._stream_ttl = 1800 # 30 min max self._counter = 0 - self._morphing_send_interval = 1 # flush the morphing buffer every trigger - self._morphing_buffer: dict[str, list] = {"value": [], "timestamp": []} - - # ------------------------------------------------------------------ helpers - def _clear_morphing_buffer(self) -> None: - """Clear the buffer used for the late-typed ``morphing`` signal.""" - self._morphing_buffer["value"].clear() - self._morphing_buffer["timestamp"].clear() def _generate_spectrum(self) -> np.ndarray: """Generate a noisy 1D spectrum whose peak drifts with the trigger counter.""" @@ -416,39 +395,19 @@ class SimMonitorMixedSignals(PSIDeviceBase, SimMonitorMixedSignalsControl): spectrum = spectrum + np.random.normal(0, 2, size) return self.BIT_DEPTH(np.clip(spectrum, 0, None)) - def _send_morphing_to_bec(self) -> None: - """Stream the buffered ``morphing`` values on the async readback endpoint. - - This emulates a legacy/late-typed async signal: a plain signal whose values are - delivered asynchronously even though its static signal_class looks synchronous. - """ - msg = messages.DeviceMessage( - signals={self.morphing.name: dict(self._morphing_buffer)}, - metadata={"async_update": {"type": "add", "max_shape": [None]}}, - ) - self.connector.xadd( - MessageEndpoints.device_async_readback( - scan_id=self.scan_info.msg.scan_id, device=self.name - ), - {"data": msg}, - expire=self._stream_ttl, - ) - self._clear_morphing_buffer() - - # -------------------------------------------------------------- scan hooks def on_stage(self) -> None: - """Reset counters and buffers for a fresh scan.""" + """Reset the trigger counter for a fresh scan.""" self._counter = 0 - self._clear_morphing_buffer() def on_trigger(self) -> StatusBase: - """Read the sync readback and stream all async signals for one scan point.""" + """Read the sync readback and let each async signal push its data for one point.""" def _acquire(): self._counter += 1 counts = int(self.readback.get()) # synchronous, computed readback - # Modern AsyncSignal-family streams (classified async by signal_class). + # The AsyncSignal-family signals push their own data via put(); the device + # server publishes these BECMessageSignal puts to the async endpoint. self.async_counts.put(counts, async_update={"type": "add", "max_shape": [None]}) self.async_spectrum.put(self._generate_spectrum(), async_update={"type": "replace"}) self.async_channels.put( @@ -456,15 +415,6 @@ class SimMonitorMixedSignals(PSIDeviceBase, SimMonitorMixedSignalsControl): async_update={"type": "add", "max_shape": [None]}, ) - # Late-typed signal: also stream the plain 'morphing' signal asynchronously. - if self.stream_morphing_async.get(): - val = self.BIT_DEPTH(counts) - self.morphing.put(val) - self._morphing_buffer["value"].append(val) - self._morphing_buffer["timestamp"].append(self.morphing.timestamp) - if self._counter % self._morphing_send_interval == 0: - self._send_morphing_to_bec() - # Progress (role 'progress' -> never a curve). num_points = getattr(self.scan_info.msg, "num_points", 0) or 0 self.progress.put( @@ -475,15 +425,6 @@ class SimMonitorMixedSignals(PSIDeviceBase, SimMonitorMixedSignalsControl): return self.task_handler.submit_task(_acquire) - def on_complete(self) -> StatusBase: - """Flush any buffered late-typed data at the end of the scan.""" - - def _complete(): - if self._morphing_buffer["value"]: - self._send_morphing_to_bec() - - return self.task_handler.submit_task(_complete) - def on_stop(self) -> None: """Stop the device and shut down background tasks.""" self.task_handler.shutdown() diff --git a/tests/test_simulation.py b/tests/test_simulation.py index aacb50f..f98a128 100644 --- a/tests/test_simulation.py +++ b/tests/test_simulation.py @@ -820,14 +820,13 @@ def test_mixed_mon_signal_classes(mixed_mon): assert type(mixed_mon.async_counts).__name__ == "AsyncSignal" # async assert type(mixed_mon.async_spectrum).__name__ == "AsyncSignal" # async assert type(mixed_mon.async_channels).__name__ == "AsyncMultiSignal" # async - assert type(mixed_mon.morphing).__name__ == "SetableSignal" # late-typed (sync class) assert type(mixed_mon.progress).__name__ == "ProgressSignal" # not a curve def test_mixed_mon_describe_roles(mixed_mon): """Async signals embed a 'main' signal_info; progress embeds 'progress'. - Plain synchronous signals embed no signal_info block at all. + The synchronous readback embeds no signal_info block at all. """ def role(comp): @@ -838,74 +837,45 @@ def test_mixed_mon_describe_roles(mixed_mon): assert role("async_spectrum") == "main" assert role("async_channels") == "main" assert role("progress") == "progress" - # Synchronous signals carry no signal_info block. - for comp in ("readback", "morphing"): - sig = getattr(mixed_mon, comp) - assert "signal_info" not in sig.describe().get(sig.name, {}) + # The synchronous readback carries no signal_info block. + assert "signal_info" not in mixed_mon.readback.describe().get(mixed_mon.readback.name, {}) def test_mixed_mon_stage_resets(mixed_mon): - """Staging resets the counter and the late-typed buffer.""" + """Staging resets the trigger counter.""" mixed_mon._counter = 5 - mixed_mon._morphing_buffer["value"].append(1) mixed_mon.stage() assert mixed_mon._counter == 0 - assert mixed_mon._morphing_buffer["value"] == [] -def test_mixed_mon_trigger_streams_async_signals(mixed_mon): - """One trigger updates every async signal and streams the late-typed signal.""" - with mock.patch.object(mixed_mon.connector, "xadd") as mock_xadd: - mixed_mon.stage() - status = mixed_mon.trigger() - status_wait(status) - assert status.success is True +def test_mixed_mon_trigger_pushes_async_signals(mixed_mon): + """One trigger lets each async signal push its own data (no hand-built messages). - # Modern AsyncSignal-family signals hold their last DeviceMessage. - counts_msg = mixed_mon.async_counts.get() - assert mixed_mon.async_counts.name in counts_msg.signals - assert counts_msg.metadata["async_update"] == {"type": "add", "max_shape": [None]} + Like SimWaveform, the device calls ``signal.put(..., async_update=...)``; it does + not assemble DeviceMessages or call ``connector.xadd`` itself. + """ + mixed_mon.stage() + status = mixed_mon.trigger() + status_wait(status) + assert status.success is True - spectrum_msg = mixed_mon.async_spectrum.get() - assert spectrum_msg.metadata["async_update"] == {"type": "replace"} + # Each AsyncSignal-family signal holds the DeviceMessage it pushed via put(). + counts_msg = mixed_mon.async_counts.get() + assert mixed_mon.async_counts.name in counts_msg.signals + assert counts_msg.metadata["async_update"] == {"type": "add", "max_shape": [None]} - channels_msg = mixed_mon.async_channels.get() - assert { - f"{mixed_mon.name}_async_channels_ch1", - f"{mixed_mon.name}_async_channels_ch2", - } == set(channels_msg.signals) + spectrum_msg = mixed_mon.async_spectrum.get() + assert mixed_mon.async_spectrum.name in spectrum_msg.signals + assert spectrum_msg.metadata["async_update"] == {"type": "replace"} - # Late-typed signal streamed on the async readback endpoint (interval == 1). - assert mock_xadd.call_count == 1 - assert mock_xadd.call_args[0][0] == MessageEndpoints.device_async_readback( - scan_id=mixed_mon.scan_info.msg.scan_id, device=mixed_mon.name - ) - assert mixed_mon._morphing_buffer["value"] == [] # flushed on send + channels_msg = mixed_mon.async_channels.get() + assert {f"{mixed_mon.name}_async_channels_ch1", f"{mixed_mon.name}_async_channels_ch2"} == set( + channels_msg.signals + ) - -def test_mixed_mon_morphing_can_be_disabled(mixed_mon): - """Disabling stream_morphing_async stops the late-typed async streaming.""" - mixed_mon.stream_morphing_async.put(0) - with mock.patch.object(mixed_mon.connector, "xadd") as mock_xadd: - mixed_mon.stage() - status_wait(mixed_mon.trigger()) - assert mock_xadd.call_count == 0 - assert mixed_mon._morphing_buffer["value"] == [] - - -def test_mixed_mon_complete_flushes_buffer(mixed_mon): - """on_complete flushes any buffered late-typed data not yet sent.""" - mixed_mon._morphing_send_interval = 100 # don't flush on trigger - with mock.patch.object(mixed_mon.connector, "xadd") as mock_xadd: - mixed_mon.stage() - status_wait(mixed_mon.trigger()) - assert mock_xadd.call_count == 0 # buffered, not sent yet - assert mixed_mon._morphing_buffer["value"] != [] - status = mixed_mon.complete() - status_wait(status) - assert status.success is True - assert mock_xadd.call_count == 1 - assert mixed_mon._morphing_buffer["value"] == [] + # Progress was reported (role 'progress' -> not a curve). + assert mixed_mon._counter == 1 + assert mixed_mon.progress.get() is not None def test_positioner_updated_timestamp(positioner):