Merge pull request #197 from ISISComputingGroup/add_run_plan

Add run_plan

Author: rerpha

doc/dev/troubleshooting.md

@@ -185,8 +185,9 @@ This is not specific to bluesky - it's an IPython feature where `_` is bound to
 `__` is bound to the result of the second-to-last expression, and so on.
 ```
 
-[Do not try to "hide" the `RunEngine` in a script/function](https://blueskyproject.io/bluesky/main/tutorial.html#plans-in-series).
-The `RE(...)` call should always be typed by the user, at the terminal.
+[It is not advised to "hide" the `RunEngine` in a script/function](https://blueskyproject.io/bluesky/main/tutorial.html#plans-in-series).
+The `RE(...)` call should be typed by the user, at the terminal. If you *really* need to run a plan as part of a
+larger script, see {py:obj}`ibex_bluesky_core.run_engine.run_plan`.
 
 ### Connect a device
 

doc/tutorial/overview.md

@@ -81,6 +81,9 @@ bluesky-native functionality - i.e. plans using `yield from`.
 
 Carefully review [calling external code](../plan_stubs/external_code.md) if you do need to call 
 external code in a plan.
+
+If, on the other hand, you need to run a plan as part of a larger script, see
+{py:obj}`ibex_bluesky_core.run_engine.run_plan`.
 ```
 
 For more details about plan stubs (plan fragments like `mv` and `read`), see 

src/ibex_bluesky_core/run_engine/__init__.py

@@ -3,17 +3,19 @@
 import asyncio
 import functools
 import logging
+from collections.abc import Generator
 from functools import cache
-from threading import Event
+from threading import Event, Lock
+from typing import Any, cast
 
 import bluesky.preprocessors as bpp
-from bluesky.run_engine import RunEngine
-from bluesky.utils import DuringTask
+from bluesky.run_engine import RunEngine, RunEngineResult
+from bluesky.utils import DuringTask, Msg, RunEngineControlException, RunEngineInterrupted
 
 from ibex_bluesky_core.callbacks import DocLoggingCallback
 from ibex_bluesky_core.preprocessors import add_rb_number_processor
 
-__all__ = ["get_run_engine"]
+__all__ = ["get_run_engine", "run_plan"]
 
 
 from ibex_bluesky_core.plan_stubs import CALL_QT_AWARE_MSG_KEY, CALL_SYNC_MSG_KEY
@@ -106,3 +108,81 @@ def get_run_engine() -> RunEngine:
     RE.preprocessors.append(functools.partial(bpp.plan_mutator, msg_proc=add_rb_number_processor))
 
     return RE
+
+
+_RUN_PLAN_LOCK = Lock()  # Explicitly *not* an RLock - RunEngine is not reentrant.
+
+
+def run_plan(
+    plan: Generator[Msg, Any, Any],
+    **metadata_kw: Any,  # noqa ANN401 - this really does accept anything serializable
+) -> RunEngineResult:
+    """Run a plan.
+
+    .. Warning::
+
+        The usual way to run a plan in bluesky is by calling ``RE(plan(...))`` interactively.
+        An ``RE`` object is already available in recent versions of the IBEX user interface,
+        or can be acquired by calling ``get_run_engine()``.
+
+        Use of this function is **not recommended**, but it is nevertheless provided as an escape
+        hatch for workflows which would otherwise be difficult to express or where parts of scanning
+        scripts have not, or cannot, be migrated to bluesky.
+
+    Args:
+        plan (positional-only): The plan to run. This is typically a generator instance.
+        metadata_kw (optional, keyword-only): Keyword arguments (metadata) to pass to the bluesky
+            run engine.
+
+    Returns:
+        A ``RunEngineResult`` instance. The return value of the plan can then be accessed using
+        the ``plan_result`` attribute.
+
+    Raises:
+        RuntimeError: if the run engine was not idle at the start of this call.
+        RuntimeError: if a reentrant call to the run engine is detected.
+        :py:obj:`bluesky.utils.RunEngineInterrupted`: if the user, or the plan itself, explicitly
+            requests an interruption.
+
+    Calling a plan using this function means that keyboard-interrupt handling will be
+    degraded: all keyboard interrupts will now force an immediate abort of the plan, using
+    ``RE.abort()``, rather than giving the possibility of gracefully resuming. Cleanup handlers
+    will execute during the ``RE.abort()``.
+
+    The bluesky run engine is not reentrant. It is a programming error to attempt to run a plan
+    using this function, from within a plan. To call a sub plan from within an outer plan, use::
+
+        def outer_plan():
+            ...
+            yield from subplan(...)
+
+    """
+    RE = get_run_engine()
+
+    if not _RUN_PLAN_LOCK.acquire(blocking=False):
+        raise RuntimeError(
+            "reentrant run_plan call attempted; this cannot be supported.\n"
+            "It is a programming error to attempt to run a plan using run_plan "
+            "from within a plan.\n"
+            "To call a sub plan from within an outer plan, "
+            "use 'yield from subplan(...)' instead.\n"
+        )
+    try:
+        if RE.state != "idle":
+            raise RuntimeError(
+                "Cannot run plan; RunEngine is not idle at start of run_plan call. "
+                "You may need to call RE.abort() to abort after a previous plan."
+            )
+        try:
+            return cast(RunEngineResult, RE(plan, **metadata_kw))
+        except (RunEngineInterrupted, RunEngineControlException):
+            raise RunEngineInterrupted(
+                "bluesky RunEngine interrupted; not resumable as running via run_plan"
+            ) from None
+        finally:
+            if RE.state != "idle":
+                # Safest reasonable default? Don't want to halt()
+                # as that wouldn't do cleanup e.g. dae settings
+                RE.abort()
+    finally:
+        _RUN_PLAN_LOCK.release()

tests/test_run_engine.py

@@ -6,11 +6,12 @@
 from unittest.mock import MagicMock
 
 import bluesky.plan_stubs as bps
+import bluesky.preprocessors as bpp
 import pytest
 from bluesky.run_engine import RunEngineResult
 from bluesky.utils import Msg, RequestAbort, RunEngineInterrupted
 
-from ibex_bluesky_core.run_engine import _DuringTask, get_run_engine
+from ibex_bluesky_core.run_engine import _DuringTask, get_run_engine, run_plan
 from ibex_bluesky_core.version import version
 
 
@@ -87,3 +88,61 @@ def test_during_task_does_wait_with_small_timeout():
 
 def test_runengine_has_version_number_as_metadata(RE):
     assert RE.md["versions"]["ibex_bluesky_core"] == version
+
+
+def test_run_plan_rejects_reentrant_call(RE):
+    def _null():
+        yield from bps.null()
+
+    def plan():
+        yield from _null()
+        run_plan(_null())
+
+    with pytest.raises(
+        RuntimeError, match="reentrant run_plan call attempted; this cannot be supported"
+    ):
+        run_plan(plan())
+
+
+def test_run_plan_rejects_call_if_re_already_busy(RE):
+    def _null():
+        yield from bps.null()
+
+    with pytest.raises(RunEngineInterrupted):
+        RE(bps.pause())
+    assert RE.state == "paused"
+    with pytest.raises(RuntimeError):
+        run_plan(_null())
+
+    RE.halt()
+
+
+def test_run_plan_runs_cleanup_on_interruption(RE):
+    cleaned_up = False
+
+    def plan():
+        def cleanup():
+            yield from bps.null()
+            nonlocal cleaned_up
+            cleaned_up = True
+
+        yield from bpp.finalize_wrapper(bps.pause(), cleanup())
+
+    with pytest.raises(
+        RunEngineInterrupted,
+        match="bluesky RunEngine interrupted; not resumable as running via run_plan",
+    ):
+        run_plan(plan())
+
+    assert RE.state == "idle"
+    assert cleaned_up
+
+
+def test_run_plan_happy_path(RE):
+    def plan():
+        yield from bps.null()
+        return "happy_path_result"
+
+    result = run_plan(plan())
+    assert result.plan_result == "happy_path_result"
+    assert result.exit_status == "success"