bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2026-01-30 20:54:55 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix(simulation): fix simulated devices

Browse Source
This commit is contained in:
appel_c
2025-10-15 15:24:25 +02:00
committed by Christian Appel
parent 8a964083dc
commit 051d665490
2 changed files with 12 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
ophyd_devices/sim/sim_test_devices.py
View File

@@ -388,31 +388,13 @@ class SimCameraWithPSIComponents(SimCamera):
}
progress = messages.ProgressMessage(**progress)
self.progress.put(progress)
self._set_async_signal()
self.async_signal.put(
np.sum(data, 1), async_update={"type": "add", "max_shape": [None]}
)
status = self.task_handler.submit_task(trigger_cam)
return status
def _set_async_signal(self, update_all: bool = False):
"""Set the async signal values.
Args:
update_all (bool): If True, update all signals. If False, update only hinted and normal signals.
"""
# pylint: disable=protected-access
for sig in self.async_1d._signals.values():
if update_all is True:
sig.put(np.random.random())
continue
if sig.kind in [Kind.hinted, Kind.normal]:
sig.put(np.random.random())
for sig in self.async_2d._signals.values():
if update_all is True:
sig.put(np.random.random())
continue
if sig.kind in [Kind.hinted, Kind.normal]:
sig.put(np.random.random())
def on_unstage(self):
"""Unstage device"""
self._triggers_received = 0

25
ophyd_devices/sim/sim_waveform.py
View File

@@ -207,33 +207,25 @@ class SimWaveform(Device):
waveform_shape = self.waveform_shape.get()
if async_update_type == "add_slice":
if index is not None:
metadata = {
"async_update": {
"type": "add_slice",
"index": index,
"max_shape": [None, waveform_shape],
}
async_update = {
"type": "add_slice",
"index": index,
"max_shape": [None, waveform_shape],
}
else:
metadata = {"async_update": {"type": "add", "max_shape": [None, waveform_shape]}}
async_update = {"type": "add", "max_shape": [None, waveform_shape]}
elif async_update_type == "add":
metadata = {"async_update": {"type": "add", "max_shape": [None]}}
async_update = {"type": "add", "max_shape": [None]}
else:
# Again, this should never happen -> check_value,
# but just in case we raise an exception
raise ValueError(
f"Invalid async_update type: {async_update_type} for device {self.name}"
)
# TODO remove once BEC e2e test async data is updated to use AsyncSignal 'data'
msg = messages.DeviceMessage(
signals={self.waveform.name: {"value": value, "timestamp": time.time()}},
metadata=metadata,
metadata={"async_update": async_update},
)
self.data.put(
{self.data.name: {"value": value, "timestamp": time.time()}}, metadata=metadata
)
# Send the message to BEC
self.connector.xadd(
MessageEndpoints.device_async_readback(
@@ -242,6 +234,7 @@ class SimWaveform(Device):
{"data": msg},
expire=self._stream_ttl,
)
self.data.put(value, async_update=async_update)
def stage(self) -> list[object]:
"""Stage the camera for upcoming scan