Merge pull request #139 from ISISComputingGroup/redefine

add plan stub to set a motor's current position (redefine)

Author: esmith1729

doc/plan_stubs/redefining.md

@@ -0,0 +1,37 @@
+# Redefinition
+
+The plan stubs in this module implement 'redefinition' of a parameter, where supported. That is, no physical movement
+occurs, but the reported position changes to the given position.
+
+## `redefine_motor`
+
+The {py:obj}`ibex_bluesky_core.plan_stubs.redefine_motor` plan stub can be used to redefine the current
+position of a motor (for example a {py:obj}`ibex_bluesky_core.devices.block.BlockMot`) to a new value.
+
+The motor does not physically move, but after this plan stub executes, the current position will be defined
+as `value`.
+
+```python
+from ibex_bluesky_core.plan_stubs import redefine_motor
+from ibex_bluesky_core.devices.block import block_mot, BlockMot
+import bluesky.plan_stubs as bps
+
+
+def my_plan():
+    motor: BlockMot = block_mot("my_motor")
+    optimimum_value: float = ...
+    
+    # Physically move the motor to it's optimum position
+    yield from bps.mv(motor, optimimum_value)
+    
+    # Redefine the current position as zero
+    yield from redefine_motor(motor, 0.)
+```
+
+## `redefine_refl_parameter`
+
+The {py:obj}`ibex_bluesky_core.plan_stubs.redefine_refl_parameter` plan stub can be used to redefine the current
+position of a {py:obj}`ibex_bluesky_core.devices.reflectometry.ReflParameter` to a new value.
+
+This plan stub has an identical API to that of the {py:obj}`ibex_bluesky_core.plan_stubs.redefine_motor` plan stub
+described above, but operates on a reflectometry parameter rather than a motor.

pyproject.toml

@@ -42,7 +42,9 @@ classifiers = [
 
 dependencies = [
   "bluesky",  # Bluesky framework
-  "ophyd-async[ca] >= 0.8.0",  # Device abstraction
+  # Pinned for extra motor fields
+  # Unpin when a version >= 0.9.1 is available
+  "ophyd-async[ca,pva] @ git+https://github.com/bluesky/ophyd-async@bf6b71f045bf297e2bbbfd3ea4de7f368643a948",  # Device abstraction
   "matplotlib",  # Plotting
   "lmfit",  # Fitting
   "scipy",  # Definitions of erf/erfc functions

src/ibex_bluesky_core/plan_stubs/__init__.py

@@ -1,10 +1,17 @@
 """Core plan stubs."""
 
+import logging
 from collections.abc import Callable, Generator
 from typing import ParamSpec, TypeVar, cast
 
 import bluesky.plan_stubs as bps
+from bluesky.preprocessors import finalize_wrapper
 from bluesky.utils import Msg
+from ophyd_async.epics.motor import Motor, UseSetMode
+
+from ibex_bluesky_core.devices.reflectometry import ReflParameter
+
+logger = logging.getLogger(__name__)
 
 P = ParamSpec("P")
 T = TypeVar("T")
@@ -14,7 +21,13 @@
 CALL_QT_AWARE_MSG_KEY = "ibex_bluesky_core_call_qt_aware"
 
 
-__all__ = ["call_qt_aware", "call_sync", "prompt_user_for_choice"]
+__all__ = [
+    "call_qt_aware",
+    "call_sync",
+    "prompt_user_for_choice",
+    "redefine_motor",
+    "redefine_refl_parameter",
+]
 
 
 def call_sync(func: Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> Generator[Msg, None, T]:
@@ -79,6 +92,46 @@ def call_qt_aware(
     return cast(T, (yield Msg(CALL_QT_AWARE_MSG_KEY, func, *args, **kwargs)))
 
 
+def redefine_motor(motor: Motor, position: float) -> Generator[Msg, None, None]:
+    """Redefines the current positions of a motor.
+
+    Note:
+        This does not move the motor, it just redefines its position to be the given value.
+
+    Args:
+        motor: The motor to set a position on.
+        position: The position to set.
+
+    """
+    logger.info("Redefining motor %s to %s", motor.name, position)
+
+    def make_motor_usable() -> Generator[Msg, None, None]:
+        yield from bps.abs_set(motor.set_use_switch, UseSetMode.USE)
+
+    def inner() -> Generator[Msg, None, None]:
+        yield from bps.abs_set(motor.set_use_switch, UseSetMode.SET)
+        yield from bps.abs_set(motor.user_setpoint, position)
+
+    return (yield from finalize_wrapper(inner(), make_motor_usable()))
+
+
+def redefine_refl_parameter(
+    parameter: ReflParameter, position: float
+) -> Generator[Msg, None, None]:
+    """Redefines the current positions of a reflectometry parameter.
+
+    Note:
+        This does not move the parameter, it just redefines its position to be the given value.
+
+    Args:
+        parameter: The reflectometry parameter to set a position on.
+        position: The position to set.
+
+    """
+    logger.info("Redefining refl parameter %s to %s", parameter.name, position)
+    yield from bps.mv(parameter.redefine, position)
+
+
 def prompt_user_for_choice(*, prompt: str, choices: list[str]) -> Generator[Msg, None, str]:
     """Prompt the user to choose between a limited set of options.
 

tests/plans/test_init.py

@@ -271,7 +271,7 @@ async def test_polling_plan_drops_readable_updates_if_no_new_motor_position(RE):
     motor.user_readback.set_name("motor1")
     await motor.velocity.set(2)
     block_readable = BlockR(prefix="UNITTEST:", block_name="READABLE", datatype=int)
-    initial_pos = 0
+    initial_pos = 0.1
     destination = 2
     initial_reading = 10
     RE(ensure_connected(motor, block_readable, mock=True))

tests/test_plan_stubs.py

@@ -1,17 +1,24 @@
 # pyright: reportMissingParameterType=false
 import time
 from asyncio import CancelledError
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, call, patch
 
 import matplotlib.pyplot as plt
 import pytest
 from bluesky.utils import Msg
+from ophyd_async.epics.motor import UseSetMode
+from ophyd_async.plan_stubs import ensure_connected
+from ophyd_async.testing import callback_on_mock_put, get_mock_put, set_mock_value
 
+from ibex_bluesky_core.devices.block import BlockMot
+from ibex_bluesky_core.devices.reflectometry import ReflParameter
 from ibex_bluesky_core.plan_stubs import (
     CALL_QT_AWARE_MSG_KEY,
     call_qt_aware,
     call_sync,
     prompt_user_for_choice,
+    redefine_motor,
+    redefine_refl_parameter,
 )
 from ibex_bluesky_core.run_engine._msg_handlers import call_sync_handler
 
@@ -126,6 +133,35 @@ def plan():
     mock.assert_not_called()
 
 
+def test_redefine_motor(RE):
+    motor = BlockMot(prefix="", block_name="some_motor")
+
+    def plan():
+        yield from ensure_connected(motor, mock=True)
+        yield from redefine_motor(motor, 42.0)
+
+    RE(plan())
+
+    get_mock_put(motor.set_use_switch).assert_has_calls(
+        [call(UseSetMode.SET, wait=True), call(UseSetMode.USE, wait=True)]
+    )
+
+    get_mock_put(motor.user_setpoint).assert_called_once_with(42.0, wait=True)
+
+
+async def test_redefine_refl_parameter(RE):
+    param = ReflParameter(prefix="", name="some_refl_parameter", changing_timeout_s=60)
+    await param.connect(mock=True)
+
+    callback_on_mock_put(
+        param.redefine.define_pos_sp, lambda *a, **k: set_mock_value(param.redefine.changed, True)
+    )
+
+    RE(redefine_refl_parameter(param, 42.0))
+
+    get_mock_put(param.redefine.define_pos_sp).assert_called_once_with(42.0, wait=True)
+
+
 def test_get_user_input(RE):
     with patch("ibex_bluesky_core.plan_stubs.input") as mock_input:
         mock_input.__name__ = "mock"