Repository: bluesky/bluesky (python)

Issue:
Add warning when plan not iterated
## Expected Behavior

At the moment, it is very easy to forget to `yield from` in the middle of the plan, and there don't appear to be any warnings that appear when a plan stub is created but not iterated. To contrast, `asyncio` will warn on garbage collect of any coroutine that is not awaited. It would be nice to do the same.

## Current Behavior

```python
from bluesky import RunEngine
import bluesky.plan_stubs as bps

RE = RunEngine()

def my_plan():
    yield from bps.null()
    bps.sleep(10)

RE(my_plan())
```

Nothing is printed, vs the asyncio:
```python
import asyncio

async def my_func():
    asyncio.sleep(1)

asyncio.run(my_func())
```
which prints
```
RuntimeWarning: coroutine 'sleep' was never awaited
  asyncio.sleep(1)
```

## Possible Solution

We could decorate the function with something that tracks iteration and raises on `__del__`:

```python
# utils.py
import traceback

class EnsureIterated:
    def __init__(self, f, *args, **kwargs):
        self._iter = f(*args, **kwargs)
        self._stack = traceback.format_stack()
        self._stack += [f"RuntimeWarning: plan `{f.__name__}` was never iterated"]

    def __iter__(self):
        self._stack = None
        yield from self._iter

    def __del__(self):
        if self._stack:
            print(*self._stack)

def ensure_plan_iterated(f):
    # TODO: wrap with functools.wrap and sort out type hints
    def wrapper(*args, **kwargs):
        return EnsureIterated(f, *args, **kwargs)
    
    return wrapper
    

# plan_stubs.py
from bluesky import Msg

@ensure_plan_iterated
def sleep(time):
    return (yield Msg('sleep', None, time))


# my_plan.py
from bluesky import RunEngine

RE = RunEngine()

def my_plan():
    yield from sleep(0.1)
    sleep(10)

RE(my_plan())
```
will output
```
  File "/usr/lib/python3.10/threading.py", line 973, in _bootstrap
    self._bootstrap_inner()
   File "/usr/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
    self.run()
   File "/usr/lib/python3.10/threading.py", line 953, in run
    self._target(*self._args, **self._kwargs)
   File "/usr/lib/python3.10/asyncio/base_events.py", line 600, in run_forever
    self._run_once()
   File "/usr/lib/python3.10/asyncio/base_events.py", line 1896, in _run_once
    handle._run()
   File "/usr/lib/python3.10/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
   File "/dls/science/users/tmc43/repos/bluesky/bluesky/run_engine.py", line 1532, in _run
    msg = self._plan_stack[-1].send(resp)
   File "/repos/bluesky/my_plan.py", line 41, in my_plan
    sleep(10)
   File "/repos/bluesky/my_plan.py", line 21, in wrapper
    return EnsureIterated(f, *args, **kwargs)
   File "/repos/bluesky/my_plan.py", line 7, in __init__
    self._stack = traceback.format_stack()
 RuntimeWarning: plan `sleep` was never iterated
```
This is a bit wordy, but demonstrates the principle...

@callumforrester @rosesyrett we talked about this issue a few days ago, maybe something like this would help?

Pull request description:
Added class and decorator to warn users if plan is not iterated over
Adds a decorator to be used on plans that warns users if the `yield from` statement is not used

## Motivation and Context
Fixes #1632 

## How Has This Been Tested?
Tested locally using the example described in the issue



Patch:
diff --git a/src/bluesky/callbacks/olog.py b/src/bluesky/callbacks/olog.py
index 26664f42..d996c3f6 100644
--- a/src/bluesky/callbacks/olog.py
+++ b/src/bluesky/callbacks/olog.py
@@ -173,5 +173,5 @@ class OlogCallback(CallbackBase):
             pformat(doc),
         )
         olog_status = self.client.log(document_content, logbooks=self.logbook)
-        logger.debug("client.log returned %s", olog_status)
+        logger.debug("client.log returned %s", olog_status)  # noqa: UP031
         super().start(doc)
diff --git a/src/bluesky/plan_stubs.py b/src/bluesky/plan_stubs.py
index d087151c..8361ba60 100644
--- a/src/bluesky/plan_stubs.py
+++ b/src/bluesky/plan_stubs.py
@@ -22,6 +22,7 @@ from .utils import (
     ensure_generator,
     get_hinted_fields,
     merge_cycler,
+    plan,
     separate_devices,
 )
 from .utils import (
@@ -29,6 +30,7 @@ from .utils import (
 )
 
 
+@plan
 def declare_stream(*objs, name: str, collect=False):
     """
     Bundle future readings into a new Event document.
@@ -53,6 +55,7 @@ def declare_stream(*objs, name: str, collect=False):
     return (yield Msg("declare_stream", None, *separate_devices(objs), name=name, collect=collect))
 
 
+@plan
 def create(name="primary"):
     """
     Bundle future readings into a new Event document.
@@ -75,6 +78,7 @@ def create(name="primary"):
     return (yield Msg("create", name=name))
 
 
+@plan
 def save():
     """
     Close a bundle of readings and emit a completed Event document.
@@ -91,6 +95,7 @@ def save():
     return (yield Msg("save"))
 
 
+@plan
 def drop():
     """
     Drop a bundle of readings without emitting a completed Event document.
@@ -108,6 +113,7 @@ def drop():
     return (yield Msg("drop"))
 
 
+@plan
 def read(obj):
     """
     Take a reading and add it to the current bundle of readings.
@@ -124,6 +130,7 @@ def read(obj):
     return (yield Msg("read", obj))
 
 
+@plan
 def monitor(obj, *, name=None, **kwargs):
     """
     Asynchronously monitor for new values and emit Event documents.
@@ -150,6 +157,7 @@ def monitor(obj, *, name=None, **kwargs):
     return (yield Msg("monitor", obj, name=name, **kwargs))
 
 
+@plan
 def unmonitor(obj):
     """
     Stop monitoring.
@@ -170,6 +178,7 @@ def unmonitor(obj):
     return (yield Msg("unmonitor", obj))
 
 
+@plan
 def null():
     """
     Yield a no-op Message. (Primarily for debugging and testing.)
@@ -182,6 +191,7 @@ def null():
     return (yield Msg("null"))
 
 
+@plan
 def abs_set(obj, *args, group=None, wait=False, **kwargs):
     """
     Set a value. Optionally, wait for it to complete before continuing.
@@ -217,6 +227,7 @@ def abs_set(obj, *args, group=None, wait=False, **kwargs):
     return ret
 
 
+@plan
 def rel_set(obj, *args, group=None, wait=False, **kwargs):
     """
     Set a value relative to current value. Optionally, wait before continuing.
@@ -248,6 +259,7 @@ def rel_set(obj, *args, group=None, wait=False, **kwargs):
     return (yield from relative_set_wrapper(abs_set(obj, *args, group=group, wait=wait, **kwargs)))
 
 
+@plan
 def mv(*args, group=None, **kwargs):
     """
     Move one or more devices to a setpoint. Wait for all to complete.
@@ -287,6 +299,7 @@ def mv(*args, group=None, **kwargs):
 mov = mv  # synonym
 
 
+@plan
 def mvr(*args, group=None, **kwargs):
     """
     Move one or more devices to a relative setpoint. Wait for all to complete.
@@ -327,6 +340,7 @@ def mvr(*args, group=None, **kwargs):
 movr = mvr  # synonym
 
 
+@plan
 def rd(obj, *, default_value=0):
     """Reads a single-value non-triggered object
 
@@ -429,6 +443,7 @@ def rd(obj, *, default_value=0):
         return data["value"]
 
 
+@plan
 def stop(obj):
     """
     Stop a device.
@@ -444,6 +459,7 @@ def stop(obj):
     return (yield Msg("stop", obj))
 
 
+@plan
 def trigger(obj, *, group=None, wait=False):
     """
     Trigger and acquisition. Optionally, wait for it to complete.
@@ -467,6 +483,7 @@ def trigger(obj, *, group=None, wait=False):
     return ret
 
 
+@plan
 def sleep(time):
     """
     Tell the RunEngine to sleep, while asynchronously doing other processing.
@@ -487,6 +504,7 @@ def sleep(time):
     return (yield Msg("sleep", None, time))
 
 
+@plan
 def wait(group=None, *, timeout=None):
     """
     Wait for all statuses in a group to report being finished.
@@ -507,6 +525,7 @@ def wait(group=None, *, timeout=None):
 _wait = wait  # for internal references to avoid collision with 'wait' kwarg
 
 
+@plan
 def checkpoint():
     """
     If interrupted, rewind to this point.
@@ -523,6 +542,7 @@ def checkpoint():
     return (yield Msg("checkpoint"))
 
 
+@plan
 def clear_checkpoint():
     """
     Designate that it is not safe to resume. If interrupted or paused, abort.
@@ -539,6 +559,7 @@ def clear_checkpoint():
     return (yield Msg("clear_checkpoint"))
 
 
+@plan
 def pause():
     """
     Pause and wait for the user to resume.
@@ -556,6 +577,7 @@ def pause():
     return (yield Msg("pause", None, defer=False))
 
 
+@plan
 def deferred_pause():
     """
     Pause at the next checkpoint.
@@ -573,6 +595,7 @@ def deferred_pause():
     return (yield Msg("pause", None, defer=True))
 
 
+@plan
 def input_plan(prompt=""):
     """
     Prompt the user for text input.
@@ -590,6 +613,7 @@ def input_plan(prompt=""):
     return (yield Msg("input", prompt=prompt))
 
 
+@plan
 def prepare(obj, *args, group=None, wait=False, **kwargs):
     """
     Prepare a device.
@@ -623,6 +647,7 @@ def prepare(obj, *args, group=None, wait=False, **kwargs):
     return ret
 
 
+@plan
 def kickoff(obj, *, group=None, wait=False, **kwargs):
     """
     Kickoff one fly-scanning device.
@@ -655,6 +680,7 @@ def kickoff(obj, *, group=None, wait=False, **kwargs):
     return ret
 
 
+@plan
 def kickoff_all(*args, group=None, wait=True, **kwargs):
     """
     Kickoff one or more fly-scanning devices.
@@ -695,6 +721,7 @@ def kickoff_all(*args, group=None, wait=True, **kwargs):
     return tuple(statuses)
 
 
+@plan
 def complete(obj, *, group=None, wait=False, **kwargs):
     """
     Tell a flyable, 'stop collecting, whenever you are ready'.
@@ -734,6 +761,7 @@ def complete(obj, *, group=None, wait=False, **kwargs):
     return ret
 
 
+@plan
 def complete_all(*args, group=None, wait=False, **kwargs):
     """
     Tell one or more flyable objects, 'stop collecting, whenever you are ready'.
@@ -780,6 +808,7 @@ def complete_all(*args, group=None, wait=False, **kwargs):
     return tuple(statuses)
 
 
+@plan
 def collect(obj, *args, stream=False, return_payload=True, name=None):
     """
     Collect data cached by one or more fly-scanning devices and emit documents.
@@ -813,6 +842,7 @@ def collect(obj, *args, stream=False, return_payload=True, name=None):
     return (yield Msg("collect", obj, *args, stream=stream, return_payload=return_payload, name=name))
 
 
+@plan
 def configure(obj, *args, **kwargs):
     """
     Change Device configuration and emit an updated Event Descriptor document.
@@ -833,6 +863,7 @@ def configure(obj, *args, **kwargs):
     return (yield Msg("configure", obj, *args, **kwargs))
 
 
+@plan
 def stage(obj, *, group=None, wait=None):
     """
     'Stage' a device (i.e., prepare it for use, 'arm' it).
@@ -870,6 +901,7 @@ def stage(obj, *, group=None, wait=None):
     return ret
 
 
+@plan
 def stage_all(*args, group=None):
     """
     'Stage' one or more devices (i.e., prepare them for use, 'arm' them).
@@ -902,6 +934,7 @@ def stage_all(*args, group=None):
         yield Msg("wait", None, group=group)
 
 
+@plan
 def unstage(obj, *, group=None, wait=None):
     """
     'Unstage' a device (i.e., put it in standby, 'disarm' it).
@@ -939,6 +972,7 @@ def unstage(obj, *, group=None, wait=None):
     return ret
 
 
+@plan
 def unstage_all(*args, group=None):
     """
     'Unstage' one or more devices (i.e., put them in standby, 'disarm' them).
@@ -971,6 +1005,7 @@ def unstage_all(*args, group=None):
         yield Msg("wait", None, group=group)
 
 
+@plan
 def subscribe(name, func):
     """
     Subscribe the stream of emitted documents.
@@ -994,6 +1029,7 @@ def subscribe(name, func):
     return (yield Msg("subscribe", None, func, name))
 
 
+@plan
 def unsubscribe(token):
     """
     Remove a subscription.
@@ -1015,6 +1051,7 @@ def unsubscribe(token):
     return (yield Msg("unsubscribe", token=token))
 
 
+@plan
 def install_suspender(suspender):
     """
     Install a suspender during a plan.
@@ -1036,6 +1073,7 @@ def install_suspender(suspender):
     return (yield Msg("install_suspender", None, suspender))
 
 
+@plan
 def remove_suspender(suspender):
     """
     Remove a suspender during a plan.
@@ -1057,6 +1095,7 @@ def remove_suspender(suspender):
     return (yield Msg("remove_suspender", None, suspender))
 
 
+@plan
 def open_run(md=None):
     """
     Mark the beginning of a new 'run'. Emit a RunStart document.
@@ -1078,6 +1117,7 @@ def open_run(md=None):
     return (yield Msg("open_run", **(md or {})))
 
 
+@plan
 def close_run(exit_status=None, reason=None):
     """
     Mark the end of the current 'run'. Emit a RunStop document.
@@ -1101,6 +1141,7 @@ def close_run(exit_status=None, reason=None):
     return (yield Msg("close_run", exit_status=exit_status, reason=reason))
 
 
+@plan
 def wait_for(futures, **kwargs):
     """
     Low-level: wait for a list of ``asyncio.Future`` objects to set (complete).
@@ -1124,6 +1165,7 @@ def wait_for(futures, **kwargs):
     return (yield Msg("wait_for", None, futures, **kwargs))
 
 
+@plan
 def trigger_and_read(devices, name="primary"):
     """
     Trigger and read a list of detectors and bundle readings into one Event.
@@ -1184,6 +1226,7 @@ def trigger_and_read(devices, name="primary"):
     return (yield from rewindable_wrapper(inner_trigger_and_read(), rewindable))
 
 
+@plan
 def broadcast_msg(command, objs, *args, **kwargs):
     """
     Generate many copies of a message, applying it to a list of devices.
@@ -1209,6 +1252,7 @@ def broadcast_msg(command, objs, *args, **kwargs):
     return return_vals
 
 
+@plan
 def repeater(n, gen_func, *args, **kwargs):
     """
     Generate n chained copies of the messages from gen_func
@@ -1241,6 +1285,7 @@ def repeater(n, gen_func, *args, **kwargs):
         yield from gen_func(*args, **kwargs)
 
 
+@plan
 def caching_repeater(n, plan):
     """
     Generate n chained copies of the messages in a plan.
@@ -1273,6 +1318,7 @@ def caching_repeater(n, plan):
         yield from (m for m in lst_plan)
 
 
+@plan
 def one_shot(detectors, take_reading=None):
     """Inner loop of a count.
 
@@ -1298,6 +1344,7 @@ def one_shot(detectors, take_reading=None):
     yield from take_reading(list(detectors))
 
 
+@plan
 def one_1d_step(detectors, motor, step, take_reading=None):
     """
     Inner loop of a 1D step scan
@@ -1334,6 +1381,7 @@ def one_1d_step(detectors, motor, step, take_reading=None):
     return (yield from take_reading(list(detectors) + [motor]))
 
 
+@plan
 def move_per_step(step, pos_cache):
     """
     Inner loop of an N-dimensional step scan without any readings
@@ -1358,6 +1406,7 @@ def move_per_step(step, pos_cache):
     yield Msg("wait", None, group=grp)
 
 
+@plan
 def one_nd_step(detectors, step, pos_cache, take_reading=None):
     """
     Inner loop of an N-dimensional step scan
@@ -1388,6 +1437,7 @@ def one_nd_step(detectors, step, pos_cache, take_reading=None):
     yield from take_reading(list(detectors) + list(motors))
 
 
+@plan
 def repeat(plan, num=1, delay=None):
     """
     Repeat a plan num times with delay and checkpoint between each repeat.
diff --git a/src/bluesky/utils/__init__.py b/src/bluesky/utils/__init__.py
index 8046a408..bdcd13c3 100644
--- a/src/bluesky/utils/__init__.py
+++ b/src/bluesky/utils/__init__.py
@@ -10,6 +10,7 @@ import signal
 import sys
 import threading
 import time
+import traceback
 import types
 import uuid
 import warnings
@@ -1914,3 +1915,52 @@ async def maybe_await(ret: SyncOrAsync[T]) -> T:
         # Mypy does not understand how to narrow type to non-awaitable in this
         # instance, see https://github.com/python/mypy/issues/15520
         return ret  # type: ignore
+
+
+class Plan:
+    def __init__(self, f, *args, **kwargs) -> None:
+        self._iter = f(*args, **kwargs)
+        self._stack = traceback.format_stack()
+        self._stack = self._stack[:-2]
+        self._stack += [
+            f"RuntimeWarning: plan `{f.__name__}` was never iterated" ", did you mean to use `yield from`?"
+        ]
+
+    def __iter__(self):
+        self._stack = None
+        return (yield from self._iter)
+
+    def __next__(self):
+        return next(self._iter)
+
+    def __del__(self):
+        if self._stack:
+            warning_message = "\n" + "".join(self._stack)
+            warnings.warn(warning_message, RuntimeWarning, stacklevel=1)
+
+    def send(self, value):
+        return self._iter.send(value)
+
+    def throw(self, typ, val=None, tb=None):
+        return self._iter.throw(typ, val, tb)
+
+
+def plan(plan):
+    """Decorator that warns user if a `yield from` is not called
+
+    Parameters
+    ----------
+    plan : Generator
+        Generator coroutine usually a Bluesky plan
+
+    Returns
+    -------
+    Plan
+        Wrapped plans
+    """
+
+    @wraps(plan)
+    def wrapper(*args, **kwargs) -> Plan:
+        return Plan(plan, *args, **kwargs)
+
+    return wrapper


Test patch:
diff --git a/src/bluesky/tests/test_utils.py b/src/bluesky/tests/test_utils.py
index 05072708..52d2a81a 100644
--- a/src/bluesky/tests/test_utils.py
+++ b/src/bluesky/tests/test_utils.py
@@ -1,16 +1,19 @@
 import operator
+import warnings
 from functools import reduce
 
 import numpy as np
 import pytest
 from cycler import cycler
 
+from bluesky import RunEngine
 from bluesky.utils import (
     CallbackRegistry,
     Msg,
     ensure_generator,
     is_movable,
     merge_cycler,
+    plan,
     warn_if_msg_args_or_kwargs,
 )
 
@@ -505,3 +508,31 @@ https://github.com/bluesky/bluesky/issues"""
     # Called without kwargs doesn't warn
     warn_if_msg_args_or_kwargs(msg, device.kickoff, (), {})
     assert len(recwarn) == 0
+
+
+@plan
+def sample_plan():
+    return (yield Msg("null"))
+
+
+def iterating_plan():
+    yield from sample_plan()
+
+
+def non_iterating_plan():
+    yield from sample_plan()
+    sample_plan()
+
+
+@pytest.mark.parametrize("gen_func, iterated", [(iterating_plan, True), (non_iterating_plan, False)])
+def test_warning_behavior(gen_func, iterated):
+    """Test that warnings are issued correctly based on iteration."""
+    RE = RunEngine()
+    if iterated:
+        with warnings.catch_warnings(record=True) as record:
+            warnings.simplefilter("always")
+            RE(gen_func())
+            assert not record, "There should be no warnings if fully iterated"
+    else:
+        with pytest.warns(RuntimeWarning, match=r".*was never iterated.*"):
+            RE(gen_func())