bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2025-06-06 03:40:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor(waveform): added second waveform signal to the async waveform sim device

Browse Source
This commit is contained in:
wyzula_j 2025-05-05 12:54:14 +02:00
parent 673be87a2d
commit 47ae7d61a9
2 changed files with 19 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
ophyd_devices/sim/sim_waveform.py
View File

@ -82,9 +82,16 @@ class SimWaveform(Device):
burst = Cpt(SetableSignal, name="burst", value=1, kind=Kind.config)
waveform_shape = Cpt(SetableSignal, name="waveform_shape", value=SHAPE, kind=Kind.config)
waveform = Cpt(
waveform_1 = Cpt(
ReadOnlySignal,
name="waveform",
name="waveform_1",
value=np.empty(SHAPE, dtype=BIT_DEPTH),
compute_readback=True,
kind=Kind.hinted,
)
waveform_2 = Cpt(
ReadOnlySignal,
name="waveform_2",
value=np.empty(SHAPE, dtype=BIT_DEPTH),
compute_readback=True,
kind=Kind.hinted,
@ -124,6 +131,7 @@ class SimWaveform(Device):
if self.sim_init:
self.sim.set_init(self.sim_init)
self._slice_index = 0
self.signals = [self.waveform_1, self.waveform_2]
@property
def delay_slice_update(self) -> bool:
@ -154,7 +162,7 @@ class SimWaveform(Device):
try:
for _ in range(self.burst.get()):
# values of the Waveform
values = self.waveform.get()
values = [signal.get() for signal in self.signals]
# add_slice option
if self.async_update.get() == "add_slice":
size = self.slice_size.get()
@ -224,7 +232,10 @@ class SimWaveform(Device):
)
msg = messages.DeviceMessage(
signals={self.waveform.name: {"value": value, "timestamp": time.time()}},
signals={
self.waveform_1.name: {"value": value[0], "timestamp": time.time()},
self.waveform_2.name: {"value": value[1], "timestamp": time.time()},
},
metadata=metadata,
)
# Send the message to BEC

8
tests/test_simulation.py
View File

@ -741,15 +741,15 @@ def test_waveform(waveform):
"""Test the SimWaveform class"""
waveform.sim.select_model("GaussianModel")
waveform.sim.params = {"amplitude": 500, "center": 500, "sigma": 10}
data = waveform.waveform.get()
data = waveform.waveform_1.get()
assert isinstance(data, np.ndarray)
assert data.shape == waveform.SHAPE
assert np.isclose(np.argmax(data), 500, atol=5)
waveform.waveform_shape.put(50)
data = waveform.waveform.get()
data = waveform.waveform_1.get()
for model in waveform.sim.get_all_sim_models():
waveform.sim.select_model(model)
waveform.waveform.get()
waveform.waveform_1.get()
# Now also test the async readback
mock_connector = waveform.connector = mock.MagicMock()
mock_run_subs = waveform._run_subs = mock.MagicMock()
@ -788,7 +788,7 @@ def test_waveform_update_modes(waveform, mode, mock_data, expected_calls):
with (
mock.patch.object(waveform, "_run_subs") as mock_run_subs,
mock.patch.object(waveform, "_send_async_update") as mock_send_async_update,
mock.patch.object(waveform.waveform, "get", return_value=mock_data),
mock.patch.object(waveform.waveform_1, "get", return_value=mock_data),
):
status = waveform.trigger()