From fa2dcaccc22cfa360b4603023e7fda2d787c89a3 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Tue, 25 Aug 2026 13:50:20 +0000 Subject: [PATCH 1/7] Add configurable movement simulation to MockSuperConductingMagnetController --- .../beamlines/i06_1/magnet/__init__.py | 2 + .../i06_1/magnet/superconducting_magnet.py | 69 +++++++++++++++++-- .../magnet/test_superconducting_magnet.py | 57 ++++++++++++++- 3 files changed, 119 insertions(+), 9 deletions(-) diff --git a/src/dodal/devices/beamlines/i06_1/magnet/__init__.py b/src/dodal/devices/beamlines/i06_1/magnet/__init__.py index f81cc14696b..34a91107dc7 100644 --- a/src/dodal/devices/beamlines/i06_1/magnet/__init__.py +++ b/src/dodal/devices/beamlines/i06_1/magnet/__init__.py @@ -13,6 +13,7 @@ MagnetLimitStatus, MagnetMode, MagnetRampStatus, + MockSuperConductingMagnetController, SuperConductingMagnetController, ) @@ -30,5 +31,6 @@ "MagnetLimitStatus", "MagnetMode", "MagnetRampStatus", + "MockSuperConductingMagnetController", "SuperConductingMagnetController", ] diff --git a/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py b/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py index da154c500df..e4af281af8f 100644 --- a/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py +++ b/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py @@ -243,8 +243,35 @@ async def _set_phi(self, phi: float): class MockSuperConductingMagnetController( DeviceMock["SuperConductingMagnetController"] ): - """Add additional callback logic to our device to get the mock behaviour to simulate - the hardware as best we can. + """Mock controller that simulates the behaviour of the + SuperConductingMagnetController hardware. + + The mock reproduces additional IOC behaviour that is not provided by the + standard device mock, including: + + - Updating readback positions when a ramp is triggered. + - Resetting demand positions and configuring PSU limits when the mode + changes. + - Updating ramp and limit status signals. + - Simulating the movement of readback positions over time. + + Movements are simulated over multiple steps by default so that beamline + operation in mock mode behaves similarly to the real hardware. This also + allows fly scans to be exercised in mock mode, with detector events + occurring while the magnet is moving. + + Unit tests that do not require simulated movement can disable it by + setting ``steps`` to zero:: + + MockSuperConductingMagnetController(steps=0) + + Args: + name: Name of the mock device. + parent: Parent mock device, if any. + steps: Number of intermediate positions used to simulate a movement. + A value less than or equal to zero makes movements instantaneous. + ramp_time: Total time in seconds over which a simulated movement takes + place. The time is divided equally between ``steps``. """ # Pulled directly from live IOC so can replicate behaviour in mock mode. @@ -258,8 +285,18 @@ class MockSuperConductingMagnetController( MagnetMode.SPHERICAL: (2, 2, 2), } - async def connect(self, device: "SuperConductingMagnetController"): + def __init__( + self, + name: str = "", + parent: DeviceMock | None = None, + steps: int = 10, + ramp_time: float = 1.0, + ): + super().__init__(name, parent) + self.steps = steps + self.ramp_time = ramp_time + async def connect(self, device: "SuperConductingMagnetController"): async def _trigger_start_ramp(): # Whenever ramp is triggered for the ioc, readback values move to the # demand values. Simulate this behaviour here. @@ -268,10 +305,30 @@ async def _trigger_start_ramp(): device.cart.y.demand.get_value(), device.cart.z.demand.get_value(), ) + + x_r, y_r, z_r = await asyncio.gather( + device.cart.x.readback.get_value(), + device.cart.y.readback.get_value(), + device.cart.z.readback.get_value(), + ) + set_mock_value(device.ramp_status, MagnetRampStatus.RAMPING) - set_mock_value(device.cart.x.readback, x_d) - set_mock_value(device.cart.y.readback, y_d) - set_mock_value(device.cart.z.readback, z_d) + + if self.steps <= 0: + set_mock_value(device.cart.x.readback, x_d) + set_mock_value(device.cart.y.readback, y_d) + set_mock_value(device.cart.z.readback, z_d) + else: + for step in range(1, self.steps + 1): + fraction = step / self.steps + + set_mock_value(device.cart.x.readback, x_r + (x_d - x_r) * fraction) + set_mock_value(device.cart.y.readback, y_r + (y_d - y_r) * fraction) + set_mock_value(device.cart.z.readback, z_r + (z_d - z_r) * fraction) + + if self.ramp_time: + await asyncio.sleep(self.ramp_time / self.steps) + set_mock_value(device.ramp_status, MagnetRampStatus.RAMP_MADE) callback_on_mock_execute(device._start_ramp, _trigger_start_ramp) # noqa: SLF001 diff --git a/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py b/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py index 6310d585918..49fb5708cd5 100644 --- a/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py +++ b/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py @@ -20,6 +20,7 @@ MagnetRampStatus, MagnetRequest, MagnetSphericalPosition, + MockSuperConductingMagnetController, SuperConductingMagnetController, ThreeMagnetAxisPowerSupply, movement, @@ -38,11 +39,12 @@ def scmc_psu() -> ThreeMagnetAxisPowerSupply: @pytest.fixture -def scmc( +async def scmc( scmc_psu: ThreeMagnetAxisPowerSupply, ) -> SuperConductingMagnetController: - with init_devices(mock=True): - scmc = SuperConductingMagnetController("TEST:", scmc_psu) + # Optimise tests by making movement of readback to setpoint instant. + scmc = SuperConductingMagnetController("TEST:", scmc_psu, name="scmc") + await scmc.connect(mock=MockSuperConductingMagnetController(steps=0)) return scmc @@ -541,3 +543,52 @@ async def test_scmc_set_within_boundary_timeout_set_correctly( target_request, timeout=expected_timeout, ) + + +@pytest.mark.parametrize( + "steps, ramp_time", + [ + pytest.param(0, 0.0, id="instant"), + pytest.param(4, 0.04, id="stepped"), + ], +) +@pytest.mark.parametrize( + "axis, mode, value", + [ + pytest.param("x", MagnetMode.UNIAXIAL_X, 1.0, id="x"), + pytest.param("y", MagnetMode.UNIAXIAL_Y, 1.0, id="y"), + pytest.param("z", MagnetMode.UNIAXIAL_Z, 1.0, id="z"), + ], +) +async def test_mock_scmc_ramps_to_demand( + scmc_psu: ThreeMagnetAxisPowerSupply, + steps: int, + ramp_time: float, + axis: str, + mode: MagnetMode, + value: float, +): + scmc = SuperConductingMagnetController("TEST", scmc_psu, name="scmc") + await scmc.connect( + mock=MockSuperConductingMagnetController(steps=steps, ramp_time=ramp_time) + ) + await scmc.mode.set(mode) + + values = [] + + readback = getattr(scmc.cart, axis).readback + + def callback(value: dict[str, Reading[float]]): + values.append(value[readback.name]["value"]) + + readback.subscribe(callback) + + await getattr(scmc.cart, axis).set(value) + + if steps == 0: + expected_values = [0.0, value] + else: + expected_values = [value * step / steps for step in range(steps + 1)] + + assert values == expected_values + assert await scmc.ramp_status.get_value() == MagnetRampStatus.RAMP_MADE From 1417da99361eded174df410598affbeb97593811 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Tue, 25 Aug 2026 13:58:47 +0000 Subject: [PATCH 2/7] Correct name for psu --- .../beamlines/i06_1/magnet/test_superconducting_magnet.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py b/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py index 49fb5708cd5..0f01c267123 100644 --- a/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py +++ b/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py @@ -34,8 +34,8 @@ @pytest.fixture def scmc_psu() -> ThreeMagnetAxisPowerSupply: with init_devices(mock=True): - ramp_rate = ThreeMagnetAxisPowerSupply("TEST:") - return ramp_rate + scmc_psu = ThreeMagnetAxisPowerSupply("TEST:") + return scmc_psu @pytest.fixture @@ -575,7 +575,6 @@ async def test_mock_scmc_ramps_to_demand( await scmc.mode.set(mode) values = [] - readback = getattr(scmc.cart, axis).readback def callback(value: dict[str, Reading[float]]): @@ -591,4 +590,3 @@ def callback(value: dict[str, Reading[float]]): expected_values = [value * step / steps for step in range(steps + 1)] assert values == expected_values - assert await scmc.ramp_status.get_value() == MagnetRampStatus.RAMP_MADE From dee06756eade5b49622fba4be0094c746471c906 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Tue, 25 Aug 2026 14:02:20 +0000 Subject: [PATCH 3/7] Update doc string --- .../devices/beamlines/i06_1/magnet/superconducting_magnet.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py b/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py index e4af281af8f..8cb45bf188c 100644 --- a/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py +++ b/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py @@ -263,7 +263,8 @@ class MockSuperConductingMagnetController( Unit tests that do not require simulated movement can disable it by setting ``steps`` to zero:: - MockSuperConductingMagnetController(steps=0) + scmc = SuperConductingMagnetController(...) + await scmc.connect(mock=MockSuperConductingMagnetController(steps=0)) Args: name: Name of the mock device. From aed9381b51fd7ec09ea1cbde34940f8d580bb93e Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Tue, 25 Aug 2026 14:06:28 +0000 Subject: [PATCH 4/7] Make attributes private --- .../i06_1/magnet/superconducting_magnet.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py b/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py index 8cb45bf188c..cf6e6670630 100644 --- a/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py +++ b/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py @@ -294,8 +294,8 @@ def __init__( ramp_time: float = 1.0, ): super().__init__(name, parent) - self.steps = steps - self.ramp_time = ramp_time + self._steps = steps + self._ramp_time = ramp_time async def connect(self, device: "SuperConductingMagnetController"): async def _trigger_start_ramp(): @@ -315,20 +315,20 @@ async def _trigger_start_ramp(): set_mock_value(device.ramp_status, MagnetRampStatus.RAMPING) - if self.steps <= 0: + if self._steps <= 0: set_mock_value(device.cart.x.readback, x_d) set_mock_value(device.cart.y.readback, y_d) set_mock_value(device.cart.z.readback, z_d) else: - for step in range(1, self.steps + 1): - fraction = step / self.steps + for step in range(1, self._steps + 1): + fraction = step / self._steps set_mock_value(device.cart.x.readback, x_r + (x_d - x_r) * fraction) set_mock_value(device.cart.y.readback, y_r + (y_d - y_r) * fraction) set_mock_value(device.cart.z.readback, z_r + (z_d - z_r) * fraction) - if self.ramp_time: - await asyncio.sleep(self.ramp_time / self.steps) + if self._ramp_time: + await asyncio.sleep(self._ramp_time / self._steps) set_mock_value(device.ramp_status, MagnetRampStatus.RAMP_MADE) From d8480d17900393b2749103b2c009eb2055038eae Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Tue, 25 Aug 2026 14:44:14 +0000 Subject: [PATCH 5/7] Fix leaky test state --- .../magnet/test_superconducting_magnet.py | 47 ++++++++++--------- 1 file changed, 25 insertions(+), 22 deletions(-) diff --git a/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py b/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py index 0f01c267123..ac0aafb292f 100644 --- a/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py +++ b/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py @@ -353,27 +353,30 @@ async def test_scmc_executes_movement_strategy_and_ramp_at_each_step( call(movement.MagnetRequest(x=0.5), timeout=DEFAULT_TIMEOUT), call(movement.MagnetRequest(x=1.2), timeout=DEFAULT_TIMEOUT), ] - scmc._MODE_MOVEMENT_STRATEGY[MagnetMode.UNIAXIAL_X] = movement_strategy - scmc._trigger_ramp = AsyncMock() - - with patch.object( - scmc, - "_apply_step", - wraps=scmc._apply_step, - ) as mock_apply_step: - # Configures PSU limits to X=2, Y=0, Z=0 - await scmc.mode.set(MagnetMode.UNIAXIAL_X) - - # Target is within the X axis limit - await scmc.cart.x.set(1.2) - - movement_strategy.move_steps.assert_called_once_with( - ANY, - movement.MagnetRequest(x=1.2), - ) + with patch.dict( + scmc._MODE_MOVEMENT_STRATEGY, + {MagnetMode.UNIAXIAL_X: movement_strategy}, + ): + scmc._trigger_ramp = AsyncMock() + + with patch.object( + scmc, + "_apply_step", + wraps=scmc._apply_step, + ) as mock_apply_step: + # Configures PSU limits to X=2, Y=0, Z=0 + await scmc.mode.set(MagnetMode.UNIAXIAL_X) + + # Target is within the X axis limit + await scmc.cart.x.set(1.2) + + movement_strategy.move_steps.assert_called_once_with( + ANY, + movement.MagnetRequest(x=1.2), + ) - assert mock_apply_step.call_args_list == expected_apply_step_calls - assert scmc._trigger_ramp.call_count == len(move_steps) + assert mock_apply_step.call_args_list == expected_apply_step_calls + assert scmc._trigger_ramp.call_count == len(move_steps) async def test_external_parallel_moves_for_scmc_raise_error( @@ -568,14 +571,14 @@ async def test_mock_scmc_ramps_to_demand( mode: MagnetMode, value: float, ): - scmc = SuperConductingMagnetController("TEST", scmc_psu, name="scmc") + scmc = SuperConductingMagnetController("PV:", scmc_psu, name="scmc") await scmc.connect( mock=MockSuperConductingMagnetController(steps=steps, ramp_time=ramp_time) ) await scmc.mode.set(mode) + readback = getattr(scmc.cart, axis).readback values = [] - readback = getattr(scmc.cart, axis).readback def callback(value: dict[str, Reading[float]]): values.append(value[readback.name]["value"]) From 092b184e0c3f9b393a1cf193a4684cd667361862 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 26 Aug 2026 12:46:52 +0000 Subject: [PATCH 6/7] Simplify logic --- .../i06_1/magnet/superconducting_magnet.py | 48 ++++++------------- 1 file changed, 15 insertions(+), 33 deletions(-) diff --git a/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py b/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py index cf6e6670630..08560f98cd3 100644 --- a/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py +++ b/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py @@ -250,9 +250,8 @@ class MockSuperConductingMagnetController( standard device mock, including: - Updating readback positions when a ramp is triggered. - - Resetting demand positions and configuring PSU limits when the mode - changes. - - Updating ramp and limit status signals. + - Resetting demand positions back to zero, configuring PSU limits, and triggering a + ramp when the mode changes. - Simulating the movement of readback positions over time. Movements are simulated over multiple steps by default so that beamline @@ -263,16 +262,8 @@ class MockSuperConductingMagnetController( Unit tests that do not require simulated movement can disable it by setting ``steps`` to zero:: - scmc = SuperConductingMagnetController(...) + scmc = SuperConductingMagnetController(..., name="scmc") await scmc.connect(mock=MockSuperConductingMagnetController(steps=0)) - - Args: - name: Name of the mock device. - parent: Parent mock device, if any. - steps: Number of intermediate positions used to simulate a movement. - A value less than or equal to zero makes movements instantaneous. - ramp_time: Total time in seconds over which a simulated movement takes - place. The time is divided equally between ``steps``. """ # Pulled directly from live IOC so can replicate behaviour in mock mode. @@ -294,41 +285,32 @@ def __init__( ramp_time: float = 1.0, ): super().__init__(name, parent) - self._steps = steps - self._ramp_time = ramp_time + self.steps = steps + self.ramp_time = ramp_time async def connect(self, device: "SuperConductingMagnetController"): async def _trigger_start_ramp(): # Whenever ramp is triggered for the ioc, readback values move to the # demand values. Simulate this behaviour here. - x_d, y_d, z_d = await asyncio.gather( + x_d, y_d, z_d, x_r, y_r, z_r = await asyncio.gather( device.cart.x.demand.get_value(), device.cart.y.demand.get_value(), device.cart.z.demand.get_value(), - ) - - x_r, y_r, z_r = await asyncio.gather( device.cart.x.readback.get_value(), device.cart.y.readback.get_value(), device.cart.z.readback.get_value(), ) - set_mock_value(device.ramp_status, MagnetRampStatus.RAMPING) - if self._steps <= 0: - set_mock_value(device.cart.x.readback, x_d) - set_mock_value(device.cart.y.readback, y_d) - set_mock_value(device.cart.z.readback, z_d) - else: - for step in range(1, self._steps + 1): - fraction = step / self._steps - - set_mock_value(device.cart.x.readback, x_r + (x_d - x_r) * fraction) - set_mock_value(device.cart.y.readback, y_r + (y_d - y_r) * fraction) - set_mock_value(device.cart.z.readback, z_r + (z_d - z_r) * fraction) - - if self._ramp_time: - await asyncio.sleep(self._ramp_time / self._steps) + # Use configured number of steps or use a single step, whichever is larger + steps = max(self.steps, 1) + step_time = self.ramp_time / steps if steps > 1 else 0 + for step in range(1, steps + 1): + fraction = step / steps + set_mock_value(device.cart.x.readback, x_r + (x_d - x_r) * fraction) + set_mock_value(device.cart.y.readback, y_r + (y_d - y_r) * fraction) + set_mock_value(device.cart.z.readback, z_r + (z_d - z_r) * fraction) + await asyncio.sleep(step_time) set_mock_value(device.ramp_status, MagnetRampStatus.RAMP_MADE) From 4aa87b0eb6e5bb1ee9dbce7b7d0ca3f6e8d5fa13 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 26 Aug 2026 13:23:11 +0000 Subject: [PATCH 7/7] Update to only update axis which has new value --- .../i06_1/magnet/superconducting_magnet.py | 25 ++++++---- .../magnet/test_superconducting_magnet.py | 47 ++++++++++++------- 2 files changed, 46 insertions(+), 26 deletions(-) diff --git a/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py b/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py index 08560f98cd3..c46538fd311 100644 --- a/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py +++ b/src/dodal/devices/beamlines/i06_1/magnet/superconducting_magnet.py @@ -290,7 +290,7 @@ def __init__( async def connect(self, device: "SuperConductingMagnetController"): async def _trigger_start_ramp(): - # Whenever ramp is triggered for the ioc, readback values move to the + # Whenever ramp is triggered for the IOC, readback values move to the # demand values. Simulate this behaviour here. x_d, y_d, z_d, x_r, y_r, z_r = await asyncio.gather( device.cart.x.demand.get_value(), @@ -300,18 +300,27 @@ async def _trigger_start_ramp(): device.cart.y.readback.get_value(), device.cart.z.readback.get_value(), ) - set_mock_value(device.ramp_status, MagnetRampStatus.RAMPING) - + axes = ( + (device.cart.x.readback, x_r, x_d), + (device.cart.y.readback, y_r, y_d), + (device.cart.z.readback, z_r, z_d), + ) + # Only move the axis that has changed + axes_to_move = [ + (rb, rb_val, demand) for rb, rb_val, demand in axes if rb_val != demand + ] # Use configured number of steps or use a single step, whichever is larger steps = max(self.steps, 1) step_time = self.ramp_time / steps if steps > 1 else 0 + + set_mock_value(device.ramp_status, MagnetRampStatus.RAMPING) for step in range(1, steps + 1): fraction = step / steps - set_mock_value(device.cart.x.readback, x_r + (x_d - x_r) * fraction) - set_mock_value(device.cart.y.readback, y_r + (y_d - y_r) * fraction) - set_mock_value(device.cart.z.readback, z_r + (z_d - z_r) * fraction) - await asyncio.sleep(step_time) - + for readback, readback_value, demand in axes_to_move: + set_mock_value( + readback, readback_value + (demand - readback_value) * fraction + ) + await asyncio.sleep(step_time) set_mock_value(device.ramp_status, MagnetRampStatus.RAMP_MADE) callback_on_mock_execute(device._start_ramp, _trigger_start_ramp) # noqa: SLF001 diff --git a/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py b/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py index ac0aafb292f..118dea79495 100644 --- a/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py +++ b/tests/devices/beamlines/i06_1/magnet/test_superconducting_magnet.py @@ -6,7 +6,7 @@ from bluesky import FailedStatus, RunEngine from bluesky.plan_stubs import mv from bluesky.protocols import Reading -from ophyd_async.core import DEFAULT_TIMEOUT, init_devices, set_mock_value +from ophyd_async.core import DEFAULT_TIMEOUT, SignalR, init_devices, set_mock_value from ophyd_async.testing import assert_configuration, assert_reading, partial_reading from dodal.devices.beamlines.i06_1.magnet import ( @@ -563,7 +563,7 @@ async def test_scmc_set_within_boundary_timeout_set_correctly( pytest.param("z", MagnetMode.UNIAXIAL_Z, 1.0, id="z"), ], ) -async def test_mock_scmc_ramps_to_demand( +async def test_mock_scmc_only_ramps_target_axis( scmc_psu: ThreeMagnetAxisPowerSupply, steps: int, ramp_time: float, @@ -573,23 +573,34 @@ async def test_mock_scmc_ramps_to_demand( ): scmc = SuperConductingMagnetController("PV:", scmc_psu, name="scmc") await scmc.connect( - mock=MockSuperConductingMagnetController(steps=steps, ramp_time=ramp_time) + mock=MockSuperConductingMagnetController( + steps=steps, + ramp_time=ramp_time, + ) ) await scmc.mode.set(mode) - readback = getattr(scmc.cart, axis).readback - - values = [] - - def callback(value: dict[str, Reading[float]]): - values.append(value[readback.name]["value"]) - - readback.subscribe(callback) + readbacks: dict[str, SignalR[float]] = { + axis: getattr(scmc.cart, axis).readback for axis in ("x", "y", "z") + } + values: dict[str, list[float]] = {axis: [] for axis in readbacks} + + for axis_name, readback in readbacks.items(): + readback_name = readback.name + readback.subscribe( + lambda value, axis_name=axis_name, readback_name=readback_name: values[ + axis_name + ].append(value[readback_name]["value"]) + ) await getattr(scmc.cart, axis).set(value) - - if steps == 0: - expected_values = [0.0, value] - else: - expected_values = [value * step / steps for step in range(steps + 1)] - - assert values == expected_values + # The initial 0.0 is emitted when the readback subscription is created, + # followed by each value produced during the ramp. + assert values[axis] == [ + 0.0, + *(value * step / max(steps, 1) for step in range(1, max(steps, 1) + 1)), + ] + # Non-target axes should only emit their initial readback value and should not + # be updated by the ramp as value not changed. + for other_axis in readbacks: + if other_axis != axis: + assert values[other_axis] == [0.0]