Repository: bluesky/bluesky
Pull request #1522 (2022-05-26 14:42:10)

Add async capabilities to protocols
To better document how Bluesky interfaces with hardware and make the asyncio compatible.
Supersedes #1502 

## Description
- Add async capabilities to protocols
- Make bundler and run engine use them
- Cache describe_* for both read and subscribe
- PendingDeprecationWarning for args and kwargs passed to collect_asset_docs, read, kickoff, complete or trigger
- Allow subscriptions to callback with Readings, rather than having to call read to get them
- Add helpers for calling in bluesky event loop
- Remove Movable ABC

## Motivation and Context
So I can write Ophyd.v2

## How Has This Been Tested?
Unit tests pass

ksunden: I tend to agree with @untzag, though the `float` annotation does not capture the fullness of what can be passed there... the point is that to my understanding, timeout is not a required kwarg, it is an implementation detail and to type check actual implementations, all that is required is a single positional argument. 

(I personally tend towards devices where that _is_ a float, but various devices use either arrays, str, or dict for various reasons, and not sure we can narrow that type)

I will note that something specified as a single positional argument does type check against an implementation which allows variadic args and kwargs just fine, but that is an implementation detail of the particular library.
coretl: I didn't know there were those tests, I've put the original definition back as adding anything to the type signature would require all implementers to add it to the implementation
coretl: BTW the docs build error should go away once #1524 is merged

Patch:
diff --git a/bluesky/bundlers.py b/bluesky/bundlers.py
index 5fe94ce9..a3850ea1 100644
--- a/bluesky/bundlers.py
+++ b/bluesky/bundlers.py
@@ -1,16 +1,27 @@
 from collections import deque
+import inspect
 from itertools import count, tee
 import time as ttime
+from typing import Any, Deque, Dict, FrozenSet, List, Tuple
 from event_model import DocumentNames
 from .log import doc_logger
+from .protocols import (
+    T, Callback, Configurable, PartialEvent, Descriptor, Flyable,
+    HasName, Readable, Reading, Subscribable, check_supports
+)
 from .utils import (
     new_uid,
     IllegalMessageSequence,
     _rearrange_into_parallel_dicts,
     short_uid,
     Msg,
+    maybe_await,
+    maybe_collect_asset_docs,
+    maybe_update_hints,
 )
 
+ObjDict = Dict[Any, Dict[str, T]]
+
 
 class RunBundler:
     def __init__(self, md, record_interruptions, emit, emit_sync, log):
@@ -18,17 +29,19 @@ class RunBundler:
         self.bundling = False  # if we are in the middle of bundling readings
         self._bundle_name = None  # name given to event descriptor
         self._run_start_uid = None  # The (future) runstart uid
-        self._objs_read = deque()  # objects read in one Event
-        self._read_cache = deque()  # cache of obj.read() in one Event
+        self._objs_read: Deque[HasName] = deque()  # objects read in one Event
+        self._read_cache: Deque[Dict[str, Reading]] = deque()  # cache of obj.read() in one Event
         self._asset_docs_cache = deque()  # cache of obj.collect_asset_docs()
-        self._describe_cache = dict()  # cache of all obj.describe() output
-        self._config_desc_cache = dict()  # " obj.describe_configuration()
-        self._config_values_cache = dict()  # " obj.read_configuration() values
-        self._config_ts_cache = dict()  # " obj.read_configuration() timestamps
+        self._describe_cache: ObjDict[Descriptor] = dict()  # cache of all obj.describe() output
+        self._config_desc_cache: ObjDict[Descriptor] = dict()  # " obj.describe_configuration()
+        self._config_values_cache: ObjDict[Any] = dict()  # " obj.read_configuration() values
+        self._config_ts_cache: ObjDict[Any] = dict()  # " obj.read_configuration() timestamps
         self._descriptors = dict()  # cache of {name: (objs_frozen_set, doc)}
+        # cache of {obj: {objs_frozen_set: (name, uid)}}
+        self._local_descriptors: Dict[Any, Dict[FrozenSet[str], Tuple[str, str]]] = dict()
         self._sequence_counters = dict()  # a seq_num counter per stream
         self._teed_sequence_counters = dict()  # for if we redo data-points
-        self._monitor_params = dict()  # cache of {obj: (cb, kwargs)}
+        self._monitor_params: Dict[Subscribable, Tuple[Callback, Dict]] = dict()  # cache of {obj: (cb, kwargs)}
         self.run_is_open = False
         self._uncollected = set()  # objects after kickoff(), before collect()
         # we expect the RE to take care of the composition
@@ -167,12 +180,16 @@ class RunBundler:
         if self.bundling:
             obj = msg.obj
             # if the object is not in the _describe_cache, cache it
+            # Note: there is a race condition between the code here
+            # and in monitor() and collect(), so if you do them concurrently
+            # on the same device you make obj.describe() calls multiple times.
+            # As this is harmless and not an expected use case, we don't guard
+            # against it. Reading multiple devices concurrently works fine.
             if obj not in self._describe_cache:
-                # Validate that there is no data key name collision.
-                data_keys = obj.describe()
-                self._describe_cache[obj] = data_keys
-                self._config_desc_cache[obj] = obj.describe_configuration()
-                self._cache_config(obj)
+                await self._cache_describe(obj)
+            if obj not in self._config_desc_cache:
+                await self._cache_describe_config(obj)
+                await self._cache_read_config(obj)
 
             # check that current read collides with nothing else in
             # current event
@@ -196,18 +213,34 @@ class RunBundler:
             # Ask the object for any resource or datum documents is has cached
             # and cache them as well. Likewise, these will be emitted if and
             # when _save is called.
-            if hasattr(obj, "collect_asset_docs"):
-                self._asset_docs_cache.extend(
-                    obj.collect_asset_docs(*msg.args, **msg.kwargs)
-                )
+            self._asset_docs_cache.extend(
+                maybe_collect_asset_docs(msg, obj, *msg.args, **msg.kwargs)
+            )
 
         return reading
 
-    def _cache_config(self, obj):
+    async def _cache_describe(self, obj):
+        "Read the object's describe and cache it."
+        obj = check_supports(obj, Readable)
+        self._describe_cache[obj] = await maybe_await(obj.describe())
+
+    async def _cache_describe_config(self, obj):
+        "Read the object's describe_configuration and cache it."
+        if isinstance(obj, Configurable):
+            conf_keys = await maybe_await(obj.describe_configuration())
+        else:
+            conf_keys = {}
+        self._config_desc_cache[obj] = conf_keys
+
+    async def _cache_read_config(self, obj):
         "Read the object's configuration and cache it."
+        if isinstance(obj, Configurable):
+            conf = await maybe_await(obj.read_configuration())
+        else:
+            conf = {}
         config_values = {}
         config_ts = {}
-        for key, val in obj.read_configuration().items():
+        for key, val in conf.items():
             config_values[key] = val["value"]
             config_ts[key] = val["timestamp"]
         self._config_values_cache[obj] = config_values
@@ -229,7 +262,7 @@ class RunBundler:
 
         where kwargs are passed through to ``obj.subscribe()``
         """
-        obj = msg.obj
+        obj = check_supports(msg.obj, Subscribable)
         if msg.args:
             raise ValueError(
                 "The 'monitor' Msg does not accept positional " "arguments."
@@ -242,16 +275,20 @@ class RunBundler:
                 "which is already monitored".format(obj)
             )
         descriptor_uid = new_uid()
-        data_keys = obj.describe()
-        config = {obj.name: {"data": {}, "timestamps": {}}}
-        config[obj.name]["data_keys"] = obj.describe_configuration()
-        for key, val in obj.read_configuration().items():
-            config[obj.name]["data"][key] = val["value"]
-            config[obj.name]["timestamps"][key] = val["timestamp"]
+        if obj not in self._describe_cache:
+            await self._cache_describe(obj)
+        if obj not in self._config_desc_cache:
+            await self._cache_describe_config(obj)
+            await self._cache_read_config(obj)
+        data_keys = self._describe_cache[obj]
+        config = {obj.name: {
+            "data": self._config_values_cache[obj],
+            "timestamps": self._config_ts_cache[obj],
+            "data_keys": self._config_desc_cache[obj]
+        }}
         object_keys = {obj.name: list(data_keys)}
         hints = {}
-        if hasattr(obj, "hints"):
-            hints.update({obj.name: obj.hints})
+        maybe_update_hints(hints, obj)
         desc_doc = dict(
             run_start=self._run_start_uid,
             time=ttime.time(),
@@ -270,10 +307,21 @@ class RunBundler:
                                 'data_keys': data_keys.keys()})
         seq_num_counter = count(1)
 
-        def emit_event(*args, **kwargs):
-            # Ignore the inputs. Use this call as a signal to call read on the
-            # object, a crude way to be sure we get all the info we need.
-            data, timestamps = _rearrange_into_parallel_dicts(obj.read())
+        def emit_event(readings: Dict[str, Reading] = None, *args, **kwargs):
+            if readings is not None:
+                # We were passed something we can use, but check no args or kwargs
+                assert not args and not kwargs, \
+                    "If subscribe callback called with readings, " \
+                    "args and kwargs are not supported."
+            else:
+                # Ignore the inputs. Use this call as a signal to call read on the
+                # object, a crude way to be sure we get all the info we need.
+                readable_obj = check_supports(obj, Readable)
+                readings = readable_obj.read()
+                assert not inspect.isawaitable(readings), \
+                    f"{readable_obj} has async read() method and the callback " \
+                    "passed to subscribe() was not called with Dict[str, Reading]"
+            data, timestamps = _rearrange_into_parallel_dicts(readings)
             doc = dict(
                 descriptor=descriptor_uid,
                 time=ttime.time(),
@@ -286,6 +334,7 @@ class RunBundler:
 
         self._monitor_params[obj] = emit_event, kwargs
         await self.emit(DocumentNames.descriptor, desc_doc)
+        # TODO: deprecate **kwargs when Ophyd.v2 is available
         obj.subscribe(emit_event, **kwargs)
 
     def record_interruption(self, content):
@@ -323,7 +372,7 @@ class RunBundler:
 
             Msg('unmonitor', obj)
         """
-        obj = msg.obj
+        obj = check_supports(msg.obj, Subscribable)
         if obj not in self._monitor_params:
             raise IllegalMessageSequence(
                 f"Cannot 'unmonitor' {obj}; it is not " "being monitored."
@@ -392,7 +441,7 @@ class RunBundler:
                 dks = self._describe_cache[obj]
                 obj_name = obj.name
                 # dks is an OrderedDict. Record that order as a list.
-                object_keys[obj.name] = list(dks)
+                object_keys[obj_name] = list(dks)
                 for field, dk in dks.items():
                     dk["object_name"] = obj_name
                 data_keys.update(dks)
@@ -400,8 +449,7 @@ class RunBundler:
                 config[obj_name]["data"] = self._config_values_cache[obj]
                 config[obj_name]["timestamps"] = self._config_ts_cache[obj]
                 config[obj_name]["data_keys"] = self._config_desc_cache[obj]
-                if hasattr(obj, "hints"):
-                    hints[obj_name] = obj.hints
+                maybe_update_hints(hints, obj)
             descriptor_uid = new_uid()
             descriptor_doc = dict(
                 run_start=self._run_start_uid,
@@ -573,63 +621,30 @@ class RunBundler:
         """
         ...
 
-    async def collect(self, msg):
-        """
-        Collect data cached by a flyer and emit documents.
-
-        Expect message object is
-
-            Msg('collect', collect_obj)
-            Msg('collect', flyer_object, stream=True, return_payload=False)
-        """
-        collect_obj = msg.obj
-
-        if not self.run_is_open:
-            # sanity check -- 'kickoff' should catch this and make this
-            # code path impossible
-            raise IllegalMessageSequence(
-                "A 'collect' message was sent but no run is open."
-            )
-        self._uncollected.discard(collect_obj)
-
-        if hasattr(collect_obj, "collect_asset_docs"):
-            # Resource and Datum documents
-            for name, doc in collect_obj.collect_asset_docs():
-                # Add a 'run_start' field to the resource document on its way out.
-                if name == "resource":
-                    doc["run_start"] = self._run_start_uid
-                await self.emit(DocumentNames(name), doc)
-
-        collect_obj_config = {}
-        bulk_data = {}
-        local_descriptors = {}  # hashed on objs_read, not (name, objs_read)
+    async def _cache_describe_collect(self, collect_obj: Flyable):
+        "Read the object's describe_collect and cache it."
+        describe_collect = await maybe_await(collect_obj.describe_collect())
+        collect_obj_config: Dict[str, Dict[str, Any]] = {}
+        local_descriptors: Dict[FrozenSet[str], Tuple[str, str]] = {}
         # collect_obj.describe_collect() returns a dictionary like this:
         #     {name_for_desc1: data_keys_for_desc1,
         #      name_for_desc2: data_keys_for_desc2, ...}
-        for stream_name, stream_data_keys in collect_obj.describe_collect().items():
+        for stream_name, stream_data_keys in describe_collect.items():
             if stream_name not in self._descriptors:
                 # We do not have an Event Descriptor for this set.
                 descriptor_uid = new_uid()
                 # if we have not yet read the configuration, do so
                 if collect_obj.name not in collect_obj_config:
-                    _config = collect_obj_config[collect_obj.name] = {
-                        "data": {},
-                        "timestamps": {},
-                        "data_keys": {}
+                    if collect_obj not in self._config_desc_cache:
+                        await self._cache_describe_config(collect_obj)
+                        await self._cache_read_config(collect_obj)
+                    collect_obj_config[collect_obj.name] = {
+                        "data": self._config_values_cache[collect_obj],
+                        "timestamps": self._config_ts_cache[collect_obj],
+                        "data_keys": self._config_desc_cache[collect_obj]
                     }
-                    # but read_configuration is optional
-                    if hasattr(collect_obj, "read_configuration"):
-                        doc_logger.debug("reading configuration from %s", collect_obj)
-                        _config['data_keys'].update(collect_obj.describe_configuration())
-                        for config_key, config in collect_obj.read_configuration().items():
-                            _config["data"][config_key] = config["value"]
-                            _config["timestamps"][config_key] = config["timestamp"]
-                    else:
-                        doc_logger.debug("%s has no read_configuration method", collect_obj)
-
                 hints = {}
-                if hasattr(collect_obj, "hints"):
-                    hints.update({collect_obj.name: collect_obj.hints})
+                maybe_update_hints(hints, collect_obj)
                 doc = dict(
                     run_start=self._run_start_uid,
                     time=ttime.time(),
@@ -660,8 +675,39 @@ class RunBundler:
 
             descriptor_uid = doc["uid"]
             local_descriptors[frozenset(stream_data_keys)] = (stream_name, descriptor_uid)
+        self._local_descriptors[collect_obj] = local_descriptors
+
+    async def collect(self, msg):
+        """
+        Collect data cached by a flyer and emit documents.
+
+        Expect message object is
+
+            Msg('collect', collect_obj)
+            Msg('collect', flyer_object, stream=True, return_payload=False)
+        """
+        collect_obj = check_supports(msg.obj, Flyable)
+
+        if not self.run_is_open:
+            # sanity check -- 'kickoff' should catch this and make this
+            # code path impossible
+            raise IllegalMessageSequence(
+                "A 'collect' message was sent but no run is open."
+            )
+        self._uncollected.discard(collect_obj)
 
-            bulk_data[descriptor_uid] = []
+        # Resource and Datum documents
+        for name, doc in maybe_collect_asset_docs(msg, collect_obj):
+            # Add a 'run_start' field to the resource document on its way out.
+            if name == "resource":
+                doc["run_start"] = self._run_start_uid
+            await self.emit(DocumentNames(name), doc)
+
+        # Populate descriptors and local descriptors from describe_collect
+        # if not already called
+        if collect_obj not in self._local_descriptors:
+            await self._cache_describe_collect(collect_obj)
+        local_descriptors = self._local_descriptors[collect_obj]
 
         # If stream is True, run 'event' subscription per document.
         # If stream is False, run 'bulk_events' subscription once.
@@ -671,6 +717,7 @@ class RunBundler:
         # accumulate, and return None.
         return_payload = msg.kwargs.get('return_payload', True)
         payload = []
+        bulk_data: Dict[str, List[PartialEvent]] = {}
 
         for ev in collect_obj.collect():
             if return_payload:
@@ -679,13 +726,7 @@ class RunBundler:
             objs_read = frozenset(ev["data"])
             stream_name, descriptor_uid = local_descriptors[objs_read]
             seq_num = next(self._sequence_counters[stream_name])
-
             event_uid = new_uid()
-
-            reading = ev["data"]
-            for key in ev["data"]:
-                reading[key] = reading[key]
-            ev["data"] = reading
             ev["descriptor"] = descriptor_uid
             ev["seq_num"] = seq_num
             ev["uid"] = event_uid
@@ -699,7 +740,7 @@ class RunBundler:
                                         'data_keys': ev['data'].keys()})
                 await self.emit(DocumentNames.event, ev)
             else:
-                bulk_data[descriptor_uid].append(ev)
+                bulk_data.setdefault(descriptor_uid, []).append(ev)
 
         if not stream:
             await self.emit(DocumentNames.bulk_events, bulk_data)
@@ -736,4 +777,4 @@ class RunBundler:
             obj_set, _ = self._descriptors[name]
             if obj in obj_set:
                 del self._descriptors[name]
-        self._cache_config(obj)
+        await self._cache_read_config(obj)
diff --git a/bluesky/plan_stubs.py b/bluesky/plan_stubs.py
index 57683d4d..eef58a96 100644
--- a/bluesky/plan_stubs.py
+++ b/bluesky/plan_stubs.py
@@ -1,7 +1,6 @@
 import itertools
 import uuid
 from cycler import cycler
-from . import utils
 import operator
 from functools import reduce
 from collections.abc import Iterable
@@ -14,8 +13,10 @@ try:
 except ImportError:
     from toolz import partition
 
-
+from .protocols import Triggerable
 from .utils import (
+    get_hinted_fields,
+    merge_cycler,
     separate_devices,
     all_safe_rewind,
     Msg,
@@ -51,7 +52,7 @@ def save():
     Close a bundle of readings and emit a completed Event document.
 
     Yields
-    -------
+    ------
     msg : Msg
         Msg('save')
 
@@ -251,7 +252,7 @@ def mv(*args, group=None, **kwargs):
     status_objects = []
 
     cyl = reduce(operator.add, [cycler(obj, [val]) for obj, val in partition(2, args)])
-    (step,) = utils.merge_cycler(cyl)
+    (step,) = merge_cycler(cyl)
     for obj, val in step.items():
         ret = yield Msg('set', obj, val, group=group, **kwargs)
         status_objects.append(ret)
@@ -341,7 +342,7 @@ def rd(obj, *, default_value=0):
         The "single" value of the device
 
     """
-    hints = getattr(obj, 'hints', {}).get("fields", [])
+    hints = get_hinted_fields(obj)
     if len(hints) > 1:
         msg = (
             f"Your object {obj} ({obj.name}.{getattr(obj, 'dotted_name', '')}) "
@@ -896,7 +897,7 @@ def trigger_and_read(devices, name='primary'):
         grp = _short_uid('trigger')
         no_wait = True
         for obj in devices:
-            if hasattr(obj, 'trigger'):
+            if isinstance(obj, Triggerable):
                 no_wait = False
                 yield from trigger(obj, group=grp)
         # Skip 'wait' if none of the devices implemented a trigger method.
diff --git a/bluesky/plans.py b/bluesky/plans.py
index 4940a629..9c060e6e 100644
--- a/bluesky/plans.py
+++ b/bluesky/plans.py
@@ -16,7 +16,7 @@ except ImportError:
 from . import plan_patterns
 
 from . import utils
-from .utils import Msg
+from .utils import Msg, get_hinted_fields
 
 from . import preprocessors as bpp
 from . import plan_stubs as bps
@@ -158,7 +158,7 @@ def list_scan(detectors, *args, per_step=None, md=None):
 
     x_fields = []
     for motor in motors:
-        x_fields.extend(getattr(motor, 'hints', {}).get('fields', []))
+        x_fields.extend(get_hinted_fields(motor))
 
     default_dimensions = [(x_fields, 'primary')]
 
@@ -1087,7 +1087,7 @@ def scan(detectors, *args, num=None, per_step=None, md=None):
     # call x_fields because these are meant to be the x (independent) axis
     x_fields = []
     for motor in motors:
-        x_fields.extend(getattr(motor, 'hints', {}).get('fields', []))
+        x_fields.extend(get_hinted_fields(motor))
 
     default_dimensions = [(x_fields, 'primary')]
 
diff --git a/bluesky/preprocessors.py b/bluesky/preprocessors.py
index 1c667daa..3630bf20 100644
--- a/bluesky/preprocessors.py
+++ b/bluesky/preprocessors.py
@@ -3,7 +3,7 @@ from __future__ import generator_stop
 from collections import OrderedDict, deque, ChainMap
 from collections.abc import Iterable
 import uuid
-from .utils import (normalize_subs_input, root_ancestor,
+from .utils import (get_hinted_fields, normalize_subs_input, root_ancestor,
                     separate_devices,
                     Msg, ensure_generator, single_gen,
                     short_uid as _short_uid, make_decorator,
@@ -996,7 +996,7 @@ def __read_and_stash_a_motor(obj, initial_positions, coupled_parents):
                   "relative to current position ***".format(m=obj))
             cur_pos = 0
         else:
-            fields = getattr(obj, 'hints', {}).get('fields', [])
+            fields = get_hinted_fields(obj)
             if len(fields) == 1:
                 k, = fields
                 cur_pos = reading[k]['value']
diff --git a/bluesky/protocols.py b/bluesky/protocols.py
index ef3b8376..6c310c9f 100644
--- a/bluesky/protocols.py
+++ b/bluesky/protocols.py
@@ -1,17 +1,124 @@
 try:
-    from typing import Protocol, runtime_checkable
+    from typing_extensions import Literal, Protocol, TypedDict, runtime_checkable
 except ImportError:
-    from typing_extensions import Protocol, runtime_checkable
-
-from typing import Dict, Any, Optional, Callable, Generator, List, Union, NoReturn
-
-
-Configuration = Dict[str, Dict[str, Any]]
+    from typing import Literal, Protocol, TypedDict, runtime_checkable
+
+from typing import Any, Awaitable, Callable, Dict, Iterator, List, Optional, Sequence, Tuple, Type, TypeVar, Union
+
+
+# TODO: these are not placed in Events by RE yet
+class ReadingOptional(TypedDict, total=False):
+    """A dictionary containing the optional per-reading metadata of a piece of scan data"""
+    #: * -ve: alarm unknown, e.g. device disconnected
+    #: * 0: ok, no alarm
+    #: * +ve: there is an alarm
+    #:
+    #: The exact numbers are transport specific
+    alarm_severity: int
+    #: A descriptive message if there is an alarm
+    message: str
+
+
+class Reading(ReadingOptional):
+    """A dictionary containing the value and timestamp of a piece of scan data"""
+    #: The current value, as a JSON encodable type or numpy array
+    value: Any
+    #: Timestamp in seconds since the UNIX epoch
+    timestamp: float
+
+
+Dtype = Literal["string", "number", "array", "boolean", "integer"]
+
+
+# https://github.com/bluesky/event-model/blob/master/event_model/schemas/event_descriptor.json
+# Just the data_key definition
+class DescriptorOptional(TypedDict, total=False):
+    """A dictionary containing optional per-scan metadata of a sseries of Readings"""
+    #: Where the data is stored if it is stored external to the events
+    external: str
+    #: The names for dimensions of the data. Empty list if scalar data
+    dims: Sequence[str]
+    #: Numpy dtype representation of the data. E.g. <u2
+    dtype_str: str
+    #: Number of digits after decimal place if a floating point number
+    precision: int
+    #: Engineering units of the value
+    units: str
+
+
+class Descriptor(DescriptorOptional):
+    """A dictionary containing the source and datatype of a series of Readings"""
+    #: The source of the data, e.g. an EPICS Process Variable
+    source: str
+    #: The JSON type of the data in the event. Deprecated for dtype_str
+    dtype: Dtype
+    #: The shape of the data, e.g. ``[5,5 ]`` for a 5x5 array.
+    #: An empty list ``[]`` indicates scalar data.
+    shape: Sequence[int]
+
+
+# https://github.com/bluesky/event-model/blob/master/event_model/schemas/event.json
+# But exclude descriptor, seq_num, uid added by RE
+class PartialEventOptional(TypedDict, total=False):
+    """A dictionary containing optional Event data"""
+    #: If any data key is in an external asset it should be present in this
+    #: dictionary with value False
+    filled: Dict[str, bool]
+
+
+class PartialEvent(PartialEventOptional):
+    """A dictionary containing Event data"""
+    #: The event data for each data key
+    data: Dict[str, Any]
+    #: The timestamps for each data key
+    timestamps: Dict[str, float]
+    #: The timestamp that the event was taken at
+    time: float
+
+
+# https://github.com/bluesky/event-model/blob/master/event_model/schemas/resource.json
+# But exclude run_start added by RE
+class PartialResourceOptional(TypedDict, total=False):
+    """A dictionary containing optional information needed to load an external resource"""
+    #: Whether the path is a posix or windows path. If not given default to posix
+    path_semantics: Literal["posix", "windows"]
+
+
+class PartialResource(TypedDict):
+    """A dictionary containing information needed to load an external resource"""
+    #: Hint about the format of the resource, and how it should be loaded
+    spec: str
+    #: Relative path, should not change over lifecycle of the resource
+    resource_path: str
+    #: Context-dependent root (which may change based on where data is accessed from)
+    #: that resource_path is relative to
+    root: str
+    #: Additional parameters for reading the Resource
+    resource_kwargs: Dict[str, Any]
+    #: UID that can be referenced in a Datum
+    uid: str
+
+
+# https://github.com/bluesky/event-model/blob/master/event_model/schemas/datum.json
+class Datum(TypedDict):
+    """A dictionary containing information to load event data from a Resource"""
+    #: The UID of a Resource
+    resource: str
+    #: UID that can be referenced in an Event
+    datum_id: str
+    #: Additional parameters for reading the Datum
+    datum_kwargs: Dict[str, Any]
+
+
+Asset = Union[Tuple[Literal["resource"], PartialResource], Tuple[Literal["datum"], Datum]]
+
+T = TypeVar("T")
+SyncOrAsync = Union[T, Awaitable[T]]
 
 
 @runtime_checkable
 class Status(Protocol):
-    def add_callback(self, callback: Callable[["Status"], NoReturn]) -> NoReturn:
+    def add_callback(self, callback: Callable[["Status"], None]) -> None:
         """Add a callback function to be called upon completion.
 
         The function must take the status as an argument.
@@ -23,34 +130,96 @@ class Status(Protocol):
 
     @property
     def done(self) -> bool:
+        """If done return True, otherwise return False."""
         ...
 
     @property
     def success(self) -> bool:
+        """If done return whether the operation was successful."""
         ...
 
 
 @runtime_checkable
-class Readable(Protocol):
+class HasName(Protocol):
+    @property
+    def name(self) -> str:
+        """Used to populate object_keys in the Event Descriptor
+
+        https://blueskyproject.io/event-model/event-descriptors.html#object-keys"""
+        ...
+
+
+@runtime_checkable
+class HasParent(Protocol):
     @property
     def parent(self) -> Optional[Any]:
-        """``None``, or a reference to a parent device."""
+        """``None``, or a reference to a parent device.
+
+        Used by the RE to stop duplicate stages.
+        """
         ...
 
+
+@runtime_checkable
+class WritesExternalAssets(Protocol):
+    def collect_asset_docs(self) -> Iterator[Asset]:
+        """Create the resource and datum documents describing data in external source.
+
+        Example yielded values:
+
+        .. code-block:: python
+
+            ('resource', {
+                'path_semantics': 'posix',
+                'resource_kwargs': {'frame_per_point': 1},
+                'resource_path': 'det.h5',
+                'root': '/tmp/tmpcvxbqctr/',
+                'spec': 'AD_HDF5',
+                'uid': '9123df61-a09f-49ae-9d23-41d4d6c6d788'
+            })
+            # or
+            ('datum', {
+                'datum_id': '9123df61-a09f-49ae-9d23-41d4d6c6d788/0',
+                'datum_kwargs': {'point_number': 0},
+                'resource': '9123df61-a09f-49ae-9d23-41d4d6c6d788'}
+            })
+        """
+
+
+@runtime_checkable
+class Configurable(Protocol):
+    def read_configuration(self) -> SyncOrAsync[Dict[str, Reading]]:
+        """Same API as ``read`` but for slow-changing fields related to configuration.
+        e.g., exposure time. These will typically be read only once per run.
+
+        This can be a standard function or an ``async`` function.
+        """
+        ...
+
+    def describe_configuration(self) -> SyncOrAsync[Dict[str, Descriptor]]:
+        """Same API as ``describe``, but corresponding to the keys in
+        ``read_configuration``.
+
+        This can be a standard function or an ``async`` function.
+        """
+        ...
+
+
+@runtime_checkable
+class Triggerable(Protocol):
     def trigger(self) -> Status:
         """Return a ``Status`` that is marked done when the device is done triggering.
-
-        If the device does not need to be triggered, simply return a ``Status``
-        that is marked done immediately.
         """
         ...
 
-    def read(self) -> Configuration:
-        """Return an OrderedDict mapping field name(s) to values and timestamps.
 
-        The field names must be strings. The values can be any JSON-encodable
-        type or a numpy array, which the RunEngine will convert to (nested)
-        lists. The timestamps should be UNIX time (seconds since 1970).
+@runtime_checkable
+class Readable(HasName, Protocol):
+    def read(self) -> SyncOrAsync[Dict[str, Reading]]:
+        """Return an OrderedDict mapping string field name(s) to dictionaries
+        of values and timestamps and optional per-point metadata.
+
+        This can be a standard function or an ``async`` function.
 
         Example return value:
 
@@ -63,9 +232,11 @@ class Readable(Protocol):
         """
         ...
 
-    def describe(self) -> Configuration:
+    def describe(self) -> SyncOrAsync[Dict[str, Descriptor]]:
         """Return an OrderedDict with exactly the same keys as the ``read``
-        method, here mapped to metadata about each field.
+        method, here mapped to per-scan metadata about each field.
+
+        This can be a standard function or an ``async`` function.
 
         Example return value:
 
@@ -79,59 +250,19 @@ class Readable(Protocol):
                          {'source': 'XF23-ID:SOME_PV_NAME',
                           'dtype': 'number',
                           'shape': []}))
-
-        We refer to each entry as a "data key." These fields are required:
-
-        * source (a descriptive string --- e.g., an EPICS Process Variable)
-        * dtype: one of the JSON data types: {'number', 'string', 'array'}
-        * shape: list of integers (dimension sizes) --- e.g., ``[5, 5]`` for a
-          5x5 array. Use empty list ``[]`` to indicate a scalar.
-
-        Optional additional fields (precision, units, etc.) are allowed.
-        The optional field ``external`` should be used to provide information
-        about references to externally-stored data, such as large image arrays.
         """
         ...
 
-    def read_configuration(self) -> Configuration:
-        """Same API as ``read`` but for slow-changing fields related to configuration.
-
-        (e.g., exposure time)
-
-        These will typically be read only once per run.
-
-        Of course, for simple cases, you can effectively omit this complexity
-        by returning an empty dictionary.
-        """
-        ...
-
-    def describe_configuration(self) -> Configuration:
-        """Same API as ``describe``, but corresponding to the keys in ``read_configuration``."""
-        ...
-
-    @property
-    def name(self) -> str:
-        ...
-
 
 @runtime_checkable
-class Movable(Readable, Protocol):
+class Movable(Protocol):
     def set(self, value) -> Status:
         """Return a ``Status`` that is marked done when the device is done moving."""
         ...
 
 
 @runtime_checkable
-class Flyable(Protocol):
-    @property
-    def name(self) -> str:
-        ...
-
-    @property
-    def parent(self) -> Optional[Any]:
-        """``None``, or a reference to a parent device."""
-        ...
-
+class Flyable(HasName, Protocol):
     def kickoff(self) -> Status:
         """Begin acculumating data.
 
@@ -143,7 +274,7 @@ class Flyable(Protocol):
         """Return a ``Status`` and mark it done when acquisition has completed."""
         ...
 
-    def collect(self) -> Generator[Dict[str, Any], None, None]:
+    def collect(self) -> Iterator[PartialEvent]:
         """Yield dictionaries that are partial Event documents.
 
         They should contain the keys 'time', 'data', and 'timestamps'.
@@ -151,26 +282,21 @@ class Flyable(Protocol):
         """
         ...
 
-    def describe_collect(self) -> Dict[str, Configuration]:
+    def describe_collect(self) -> SyncOrAsync[Dict[str, Dict[str, Descriptor]]]:
         """This is like ``describe()`` on readable devices, but with an extra layer of nesting.
 
         Since a flyer can potentially return more than one event stream, this is a dict
         of stream names (strings) mapped to a ``describe()``-type output for each.
-        """
-        ...
 
-    # Remaining same as Readable
-    def read_configuration(self) -> Configuration:
-        """Same as for a Readable device."""
-        ...
-
-    def describe_configuration(self) -> Configuration:
-        """Same as for a Readable device."""
+        This can be a standard function or an ``async`` function.
+        """
         ...
 
 
 @runtime_checkable
 class Stageable(Protocol):
+    # TODO: we were going to extend these to be able to return plans, what
+    # signature should they have?
     def stage(self) -> List[Any]:
         """An optional hook for "setting up" the device for acquisition.
 
@@ -192,18 +318,24 @@ class Stageable(Protocol):
 
 @runtime_checkable
 class Pausable(Protocol):
-    def pause(self) -> Union[NoReturn, None]:
-        """Perform device-specific work when the RunEngine pauses."""
+    def pause(self) -> SyncOrAsync[None]:
+        """Perform device-specific work when the RunEngine pauses.
+
+        This can be a standard function or an ``async`` function.
+        """
         ...
 
-    def resume(self) -> Union[NoReturn, None]:
-        """Perform device-specific work when the RunEngine resumes after a pause."""
+    def resume(self) -> SyncOrAsync[None]:
+        """Perform device-specific work when the RunEngine resumes after a pause.
+
+        This can be a standard function or an ``async`` function.
+        """
         ...
 
 
 @runtime_checkable
 class Stoppable(Protocol):
-    def stop(self, sucess=True) -> Union[NoReturn, None]:
+    def stop(self, success=True) -> SyncOrAsync[None]:
         """Safely stop a device that may or may not be in motion.
 
         The argument ``success`` is a boolean.
@@ -211,30 +343,35 @@ class Stoppable(Protocol):
         and the device should stop "normally".
         When ``success`` is false, something has gone wrong and the device
         may wish to take defensive action to make itself safe.
+
+        This can be a standard function or an ``async`` function.
         """
         ...
 
 
+Callback = Callable[[Dict[str, Reading]], None]
+
+
 @runtime_checkable
-class Subscribable(Protocol):
-    def subscribe(self, function: Callable) -> Union[NoReturn, None]:
+class Subscribable(HasName, Protocol):
+    def subscribe(self, function: Callback) -> None:
         """Subscribe to updates in value of a device.
 
         When the device has a new value ready, it should call ``function``
-        asynchronously in a separate thread.
+        with something that looks like the output of ``read()``.
 
         Needed for :doc:`monitored <async>`.
         """
         ...
 
-    def clear_sub(self, function: Callable) -> Union[NoReturn, None]:
+    def clear_sub(self, function: Callback) -> None:
         """Remove a subscription."""
         ...
 
 
 @runtime_checkable
 class Checkable(Protocol):
-    def check_value(self, value: Any) -> Union[NoReturn, None]:
+    def check_value(self, value: Any) -> SyncOrAsync[None]:
         """Test for a valid setpoint without actually moving.
 
         This should accept the same arguments as ``set``. It should raise an
@@ -244,14 +381,30 @@ class Checkable(Protocol):
 
         This method is used by simulators that check limits. If not implemented
         those simulators should assume all values are valid, but may warn.
+
+        This method may be used during a scan, so should not write to any Signals
+
+        This can be a standard function or an ``async`` function.
         """
         ...
 
 
+class Hints(TypedDict, total=False):
+    """A dictionary of optional hints for visualization"""
+    #: A list of the interesting fields to plot
+    fields: List[str]
+    #: Partition fields (and their stream name) into dimensions for plotting
+    #:
+    #: ``'dimensions': [(fields, stream_name), (fields, stream_name), ...]``
+    dimensions: List[Tuple[List[str], str]]
+    #: Include this if scan data is sampled on a regular rectangular grid
+    gridding: Literal["rectilinear", "rectilinear_nonsequential"]
+
+
 @runtime_checkable
-class Hinted(Protocol):
+class HasHints(HasName, Protocol):
     @property
-    def hints(self) -> Dict:
+    def hints(self) -> Hints:
         """A dictionary of suggestions for best-effort visualization and processing.
 
         This does not affect what data is read or saved; it is only
@@ -259,3 +412,19 @@ class Hinted(Protocol):
         with minimal guidance from the user. See :ref:`hints`.
         """
         ...
+
+
+def check_supports(obj, protocol: Type[T]) -> T:
+    """Check that an object supports a protocol
+
+    This exists so that multiple protocol checks can be run in a mypy
+    compatible way, e.g.::
+
+        triggerable = check_supports(obj, Triggerable)
+        triggerable.trigger()
+        readable = check_supports(obj, Readable)
+        readable.read()
+    """
+    assert isinstance(obj, protocol), \
+        "%s does not implement all %s methods" % (obj, protocol.__name__)
+    return obj
diff --git a/bluesky/run_engine.py b/bluesky/run_engine.py
index 87e0d75d..6013ea0d 100644
--- a/bluesky/run_engine.py
+++ b/bluesky/run_engine.py
@@ -2,7 +2,7 @@ import asyncio
 from datetime import datetime
 import sys
 from warnings import warn
-from inspect import Parameter, Signature
+from inspect import Parameter, Signature, iscoroutine
 from itertools import count
 from collections import deque, defaultdict, ChainMap
 from enum import Enum
@@ -13,8 +13,9 @@ import threading
 import weakref
 from dataclasses import dataclass
 import typing
-from .bundlers import RunBundler
-from . import protocols as bs_protocols
+from .bundlers import RunBundler, maybe_await
+from .protocols import (Flyable, Movable, Pausable, Readable, Stageable,
+                        Stoppable, Triggerable, check_supports)
 
 import concurrent
 
@@ -38,7 +39,7 @@ from .utils import (CallbackRegistry, SigintHandler,
                     RunEngineInterrupted, IllegalMessageSequence,
                     FailedPause, FailedStatus, InvalidCommand,
                     PlanHalt, Msg, ensure_generator, single_gen,
-                    DefaultDuringTask)
+                    DefaultDuringTask, warn_if_msg_args_or_kwargs)
 
 
 class _RunEnginePanic(Exception):
@@ -353,7 +354,8 @@ class RunEngine:
                  scan_id_source=default_scan_id_source,
                  during_task=None, call_returns_result=False):
         if loop is None:
-            loop = get_bluesky_event_loop()
+            loop = asyncio.new_event_loop()
+            set_bluesky_event_loop(loop)
         self._th = _ensure_event_loop_running(loop)
         self._state_lock = threading.RLock()
         self._loop = loop
@@ -422,7 +424,7 @@ class RunEngine:
         # The RunEngine keeps track of a *lot* of state.
         # All flags and caches are defined here with a comment. Good luck.
         self._call_returns_result = call_returns_result  # should __call__ return UIDs or plan value
-        self._run_bundlers = {}  # a mapping of open run -> bundlers
+        self._run_bundlers: typing.Dict[typing.Any, RunBundler] = {}  # a mapping of open run -> bundlers
         self._metadata_per_call = {}  # for all runs generated by one __call__
         self._deferred_pause_requested = False  # pause at next 'checkpoint'
         self._exception = None  # stored and then raised in the _run loop
@@ -933,8 +935,9 @@ class RunEngine:
         self._response_stack.append(None)
         # Notify Devices of the resume in case they want to clean up.
         for obj in self._objs_seen:
-            if hasattr(obj, 'resume'):
-                obj.resume()
+            if isinstance(obj, Pausable):
+                fut = asyncio.run_coroutine_threadsafe(maybe_await(obj.resume()), self._loop)
+                fut.result()
         plan_return = self._resume_task()
         if self._interrupted:
             raise RunEngineInterrupted(self.pause_msg) from None
@@ -1180,12 +1183,12 @@ class RunEngine:
             )
         # During suspend, all motors should be stopped. Call stop() on
         # every object we ever set().
-        self._stop_movable_objects(success=True)
+        await self._stop_movable_objects(success=True)
         # Notify Devices of the pause in case they want to clean up.
         for obj in self._objs_seen:
             if hasattr(obj, 'pause'):
                 try:
-                    obj.pause()
+                    await maybe_await(obj.pause())
                 except NoReplayAllowed:
                     self._reset_checkpoint_state_meth()
         # rewind to the last checkpoint
@@ -1325,6 +1328,7 @@ class RunEngine:
 
     def __interrupter_helper(self, coro):
         if self.state == 'panicked':
+            coro.close()
             raise RuntimeError("The RunEngine is panicked and "
                                "cannot be recovered. "
                                "You must restart bluesky.")
@@ -1370,18 +1374,16 @@ class RunEngine:
         else:
             return tuple(self._run_start_uids)
 
-    def _stop_movable_objects(self, *, success=True):
+    async def _stop_movable_objects(self, *, success=True):
         "Call obj.stop() for all objects we have moved. Log any exceptions."
         for obj in self._movable_objs_touched:
-            try:
-                stop = obj.stop
-            except AttributeError:
-                self.log.debug("No 'stop' method available on %r", obj)
-            else:
+            if isinstance(obj, Stoppable):
                 try:
-                    stop(success=success)
+                    await maybe_await(obj.stop(success=success))
                 except Exception:
                     self.log.exception("Failed to stop %r.", obj)
+            else:
+                self.log.debug("No 'stop' method available on %r", obj)
 
     async def _run(self):
         """Pull messages from the plan, process them, send results back.
@@ -1433,13 +1435,13 @@ class RunEngine:
                         await current_run.suspend_monitors()
                     # During pause, all motors should be stopped. Call stop()
                     # on every object we ever set().
-                    self._stop_movable_objects(success=True)
+                    await self._stop_movable_objects(success=True)
                     # Notify Devices of the pause in case they want to
                     # clean up.
                     for obj in self._objs_seen:
-                        if hasattr(obj, 'pause'):
+                        if isinstance(obj, Pausable):
                             try:
-                                obj.pause()
+                                await maybe_await(obj.pause())
                             except NoReplayAllowed:
                                 self._reset_checkpoint_state_meth()
                     self._state = 'paused'
@@ -1670,7 +1672,7 @@ class RunEngine:
             # Block them from creating new 'failed status' tasks on the loop.
             self._pardon_failures.set()
             # call stop() on every movable object we ever set()
-            self._stop_movable_objects(success=True)
+            await self._stop_movable_objects(success=True)
             for current_run in self._run_bundlers.values():
                 # Clear any uncleared monitoring callbacks.
                 current_run.clear_monitors()
@@ -1831,9 +1833,10 @@ class RunEngine:
 
             Msg('read', obj)
         """
-        obj = msg.obj
+        obj = check_supports(msg.obj, Readable)
         # actually _read_ the object
-        ret = obj.read(*msg.args, **msg.kwargs)
+        warn_if_msg_args_or_kwargs(msg, obj.read, msg.args, msg.kwargs)
+        ret = await maybe_await(obj.read(*msg.args, **msg.kwargs))
 
         if ret is None:
             raise RuntimeError(
@@ -1957,9 +1960,10 @@ class RunEngine:
                                          "run is open.") from ke
 
         _, obj, args, kwargs, _ = msg
+        obj = check_supports(obj, Flyable)
         kwargs = dict(msg.kwargs)
         group = kwargs.pop("group", None)
-
+        warn_if_msg_args_or_kwargs(msg, obj.kickoff, msg.args, kwargs)
         ret = obj.kickoff(*msg.args, **kwargs)
         p_event = asyncio.Event(**self._loop_for_kwargs)
         pardon_failures = self._pardon_failures
@@ -1969,7 +1973,7 @@ class RunEngine:
         def done_callback(status=None):
             self.log.debug(
                 "The object %r reports 'kickoff' is done " "with status %r",
-                msg.obj,
+                obj,
                 ret.success,
             )
             task = self._loop.call_soon_threadsafe(
@@ -2013,7 +2017,9 @@ class RunEngine:
         await current_run.complete(msg)
         kwargs = dict(msg.kwargs)
         group = kwargs.pop("group", None)
-        ret = msg.obj.complete(*msg.args, **kwargs)
+        obj = check_supports(msg.obj, Flyable)
+        warn_if_msg_args_or_kwargs(msg, obj.complete, msg.args, kwargs)
+        ret = obj.complete(*msg.args, **kwargs)
 
         p_event = asyncio.Event(**self._loop_for_kwargs)
         pardon_failures = self._pardon_failures
@@ -2021,7 +2027,7 @@ class RunEngine:
         def done_callback(status=None):
             self.log.debug(
                 "The object %r reports 'complete' is done " "with status %r",
-                msg.obj,
+                obj,
                 ret.success,
             )
             task = self._loop.call_soon_threadsafe(
@@ -2081,16 +2087,17 @@ class RunEngine:
 
         where arguments are passed through to `obj.set(*args, **kwargs)`.
         """
+        obj = check_supports(msg.obj, Movable)
         kwargs = dict(msg.kwargs)
         group = kwargs.pop('group', None)
-        self._movable_objs_touched.add(msg.obj)
-        ret = msg.obj.set(*msg.args, **kwargs)
+        self._movable_objs_touched.add(obj)
+        ret = obj.set(*msg.args, **kwargs)
         p_event = asyncio.Event(**self._loop_for_kwargs)
         pardon_failures = self._pardon_failures
 
         def done_callback(status=None):
             self.log.debug("The object %r reports set is done "
-                           "with status %r", msg.obj, ret.success)
+                           "with status %r", obj, ret.success)
             task = self._loop.call_soon_threadsafe(
                 self._status_object_completed, ret, p_event, pardon_failures)
             self._status_tasks.append(task)
@@ -2113,15 +2120,17 @@ class RunEngine:
 
             Msg('trigger', obj)
         """
+        obj = check_supports(msg.obj, Triggerable)
         kwargs = dict(msg.kwargs)
         group = kwargs.pop('group', None)
-        ret = msg.obj.trigger(*msg.args, **kwargs)
+        warn_if_msg_args_or_kwargs(msg, obj.trigger, msg.args, kwargs)
+        ret = obj.trigger(*msg.args, **kwargs)
         p_event = asyncio.Event(**self._loop_for_kwargs)
         pardon_failures = self._pardon_failures
 
         def done_callback(status=None):
             self.log.debug("The object %r reports trigger is "
-                           "done with status %r.", msg.obj, ret.success)
+                           "done with status %r.", obj, ret.success)
             task = self._loop.call_soon_threadsafe(
                 self._status_object_completed, ret, p_event, pardon_failures)
             self._status_tasks.append(task)
@@ -2226,8 +2235,8 @@ class RunEngine:
             await current_run.restore_monitors()
         # Notify Devices of the resume in case they want to clean up.
         for obj in self._objs_seen:
-            if hasattr(obj, 'resume'):
-                obj.resume()
+            if isinstance(obj, Pausable):
+                await maybe_await(obj.resume())
 
     async def _checkpoint(self, msg):
         """Instruct the RunEngine to create a checkpoint so that we can rewind
@@ -2329,7 +2338,7 @@ class RunEngine:
         """
         _, obj, args, kwargs, _ = msg
         # If an object has no 'stage' method, assume there is nothing to do.
-        if not isinstance(obj, bs_protocols.Stageable):
+        if not isinstance(obj, Stageable):
             return []
         result = obj.stage()
         self._staged.add(obj)  # add first in case of failure below
@@ -2345,7 +2354,7 @@ class RunEngine:
         """
         _, obj, args, kwargs, _ = msg
         # If an object has no 'unstage' method, assume there is nothing to do.
-        if not isinstance(obj, bs_protocols.Stageable):
+        if not isinstance(obj, Stageable):
             return []
         result = obj.unstage()
         # use `discard()` to ignore objects that are not in the staged set.
@@ -2361,7 +2370,8 @@ class RunEngine:
 
             Msg('stop', obj)
         """
-        return msg.obj.stop()  # nominally, this returns None
+        obj = check_supports(msg.obj, Stoppable)
+        return await maybe_await(obj.stop())  # nominally, this returns None
 
     async def _subscribe(self, msg):
         """
@@ -2610,12 +2620,37 @@ _bluesky_event_loop = None
 
 
 def get_bluesky_event_loop():
-    global _bluesky_event_loop
-    if _bluesky_event_loop is None:
-        _bluesky_event_loop = asyncio.new_event_loop()
     return _bluesky_event_loop
 
 
 def set_bluesky_event_loop(loop):
     global _bluesky_event_loop
     _bluesky_event_loop = loop
+
+
+def in_bluesky_event_loop() -> bool:
+    try:
+        # TODO: change to asyncio.get_running_loop() when we drop py3.6
+        loop = asyncio.get_event_loop()
+    except RuntimeError:
+        # Ok, no running loop
+        return False
+    else:
+        # Check if running loop is bluesky event loop
+        return loop is _bluesky_event_loop
+
+
+T = typing.TypeVar("T")
+
+
+def call_in_bluesky_event_loop(coro: typing.Awaitable[T], timeout: float = None) -> T:
+    if _bluesky_event_loop is None or not _bluesky_event_loop.is_running():
+        # Quell "coroutine never awaited" warnings
+        if iscoroutine(coro):
+            coro.close()
+        raise RuntimeError("Bluesky event loop not running")
+    fut = asyncio.run_coroutine_threadsafe(
+        coro,
+        loop=_bluesky_event_loop,
+    )
+    return fut.result(timeout=timeout)
diff --git a/bluesky/simulators.py b/bluesky/simulators.py
index 7a70e640..f7383622 100644
--- a/bluesky/simulators.py
+++ b/bluesky/simulators.py
@@ -1,5 +1,8 @@
 from warnings import warn
+from bluesky.utils import maybe_await
 from bluesky.preprocessors import print_summary_wrapper
+from bluesky.run_engine import call_in_bluesky_event_loop, in_bluesky_event_loop
+from .protocols import Checkable
 
 
 def plot_raster_path(plan, x_motor, y_motor, ax=None, probe_size=None, lw=2):
@@ -77,6 +80,13 @@ print_summary = summarize_plan  # back-compat
 
 
 def check_limits(plan):
+    """Run check_limits_async in the RE"""
+    if in_bluesky_event_loop():
+        raise RuntimeError("Can't call check_limits() from within RE, use await check_limits_async() instead")
+    call_in_bluesky_event_loop(check_limits_async(plan))
+
+
+async def check_limits_async(plan):
     """
     Check that a plan will not move devices outside of their limits.
 
@@ -87,10 +97,11 @@ def check_limits(plan):
     """
     ignore = []
     for msg in plan:
-        if msg.command == 'set' and msg.obj not in ignore:
-            if hasattr(msg.obj, "check_value"):
-                msg.obj.check_value(msg.args[0])
+        obj = msg.obj
+        if msg.command == 'set' and obj not in ignore:
+            if isinstance(obj, Checkable):
+                await maybe_await(obj.check_value(msg.args[0]))
             else:
-                warn(f"{msg.obj.name} has no check_value() method"
+                warn(f"{obj.name} has no check_value() method"
                      f" to check if {msg.args[0]} is within its limits.")
-                ignore.append(msg.obj)
+                ignore.append(obj)
diff --git a/bluesky/utils/__init__.py b/bluesky/utils/__init__.py
index c54bce82..8feffd25 100644
--- a/bluesky/utils/__init__.py
+++ b/bluesky/utils/__init__.py
@@ -7,7 +7,7 @@ import signal
 import operator
 import uuid
 from functools import reduce
-from typing import Any, Optional, Callable
+from typing import Any, Dict, Iterator, List, Optional, Callable
 from weakref import ref, WeakKeyDictionary
 import types
 import inspect
@@ -29,6 +29,11 @@ import msgpack
 import msgpack_numpy
 import zict
 
+from bluesky.protocols import (
+    T, Asset, HasParent, HasHints, Hints, Movable, Readable,
+    SyncOrAsync, WritesExternalAssets, check_supports
+)
+
 try:
     # cytools is a drop-in replacement for toolz, implemented in Cython
     from cytools import groupby
@@ -1389,7 +1394,7 @@ def merge_axis(objs):
         Mapping of interdependent axis passed in.
     '''
     def get_parent(o):
-        return getattr(o, 'parent')
+        return check_supports(o, HasParent).parent
 
     independent_objs = set()
     maybe_coupled = set()
@@ -1719,7 +1724,7 @@ def _rearrange_into_parallel_dicts(readings):
 
 
 def is_movable(obj):
-    """Check if object satisfies bluesky 'movable' interface.
+    """Check if object satisfies bluesky 'Movable' and `Readable` interfaces.
 
     Parameters
     ----------
@@ -1731,48 +1736,44 @@ def is_movable(obj):
     boolean
         True if movable, False otherwise.
     """
-    EXPECTED_ATTRS = (
-        'name',
-        'parent',
-        'read',
-        'describe',
-        'read_configuration',
-        'describe_configuration',
-        'set',
-    )
-    return all(hasattr(obj, attr) for attr in EXPECTED_ATTRS)
-
-
-class Movable(metaclass=abc.ABCMeta):
-    """
-    Abstract base class for objects that satisfy the bluesky 'movable' interface.
+    return isinstance(obj, Movable) and isinstance(obj, Readable)
 
-    Examples
-    --------
 
-    .. code-block:: python
+def get_hinted_fields(obj) -> List[str]:
+    if isinstance(obj, HasHints):
+        return obj.hints.get("fields", [])
+    else:
+        return []
 
-        m = hw.motor
-        # We need to detect if 'm' is a motor
-        if isinstance(m, Movable):
-            print(f"The object {m.name} is a motor")
-    """
-    @classmethod
-    def __subclasshook__(cls, C):
-        # If the following condition is True, the object C is recognized
-        # to have Movable interface (e.g. a motor)
-        msg = """The Movable abstract base class is deprecated and will be removed in a future
-                 version of bluesky. Please use bluesky.utils.is_movable(obj) to test if an object
-                 satisfies the movable interface."""
-        warnings.warn(msg, DeprecationWarning)
-        EXPECTED_ATTRS = (
-            'name',
-            'parent',
-            'read',
-            'describe',
-            'read_configuration',
-            'describe_configuration',
-            'set',
-            'stop',
-        )
-        return all(hasattr(C, attr) for attr in EXPECTED_ATTRS)
+
+already_warned = {}
+
+
+def warn_if_msg_args_or_kwargs(msg, meth, args, kwargs):
+    if args or kwargs and not already_warned.get(msg.command):
+        already_warned[msg.command] = True
+        error_msg = f"""\
+About to call {meth.__name__}() with args {args} and kwargs {kwargs}.
+In the future the passing of Msg.args and Msg.kwargs down to hardware from
+Msg("{msg.command}") may be deprecated. If you have a use case for these,
+we would like to know about it, so please open an issue at
+https://github.com/bluesky/bluesky/issues"""
+        warnings.warn(error_msg, PendingDeprecationWarning)
+
+
+def maybe_update_hints(hints: Dict[str, Hints], obj):
+    if isinstance(obj, HasHints):
+        hints[obj.name] = obj.hints
+
+
+def maybe_collect_asset_docs(msg, obj, *args, **kwargs) -> Iterator[Asset]:
+    if isinstance(obj, WritesExternalAssets):
+        warn_if_msg_args_or_kwargs(msg, obj.collect_asset_docs, args, kwargs)
+        yield from obj.collect_asset_docs(*args, **kwargs)
+
+
+async def maybe_await(ret: SyncOrAsync[T]) -> T:
+    if inspect.isawaitable(ret):
+        return await ret
+    else:
+        return ret
diff --git a/docs/source/api_changes.rst b/docs/source/api_changes.rst
index 2e46a7b0..f4ab7572 100644
--- a/docs/source/api_changes.rst
+++ b/docs/source/api_changes.rst
@@ -7,6 +7,8 @@ v1.9.0 (2022-XX-YY)
 
 * the `"resume"` message which can only be used internally has been renamed to
   `"_resume_from_suspender"`.
+* ``Movable`` (which has long been deprecated for ``bluesky.utils.is_movable``)
+  has been deleted
 
 v1.8.3 (2022-04-08)
 ===================
@@ -23,7 +25,6 @@ Documentation
 
 * Document pycertifspec as hardware interface.
 
-
 v1.8.2 (2021-12-20)
 ===================
 
diff --git a/docs/source/hardware.rst b/docs/source/hardware.rst
index d895adba..7e6efd90 100644
--- a/docs/source/hardware.rst
+++ b/docs/source/hardware.rst
@@ -17,6 +17,9 @@ methods with certain established names. We have taken pains to make this
 interface as slim as possible, while still being general enough to address
 every kind of hardware we have encountered.
 
+.. seealso::
+    :ref:`hardware_interface_packages`
+
 Specification
 -------------
 
@@ -68,22 +71,107 @@ conforms to the following definition
             A fraction of zero indicates completion.
             A fraction of one indicates progress has not started.
 
+
+Named Device
+++++++++++++
+
+Some of the interfaces below require a ``name`` attribute, they implement this
+interface:
+
+.. autoclass:: bluesky.protocols.HasName
+    :members:
+    :show-inheritance:
+
+Some of also require a ``parent`` attribute, they implement this interface:
+
+.. autoclass:: bluesky.protocols.HasParent
+    :members:
+    :show-inheritance:
+
+
 Readable Device
 +++++++++++++++
 
-The interface of a readable device:
+To produce data in a step scan, a device must be Readable:
 
 .. autoclass:: bluesky.protocols.Readable
     :members:
-    :undoc-members:
+    :show-inheritance:
+
+A dict of stream name to Descriptors is returned from :meth:`describe`, where a
+`Descriptor` is a dictionary with the following keys:
+
+.. autoclass:: bluesky.protocols.Descriptor
+    :members:
+
+The following keys can optionally be present in a `Desciptor`:
+
+.. autoclass:: bluesky.protocols.DescriptorOptional
+    :members:
 
-Movable (or "Settable")  Device
-+++++++++++++++++++++++++++++++
+A dict of stream name to Reading is returned from :meth:`read`, where a
+`Reading` is a dictionary with the following keys:
+
+.. autoclass:: bluesky.protocols.Reading
+    :members:
+
+The following keys can optionally be present in a `Reading`:
+
+.. autoclass:: bluesky.protocols.ReadingOptional
+    :members:
+
+If the device has configuration that only needs to be read once at the start of
+scan, the following interface can be implemented:
+
+.. autoclass:: bluesky.protocols.Configurable
+    :members:
+    :show-inheritance:
+
+If a device needs to do something before it can be read, the following interface
+can be implemented:
+
+.. autoclass:: bluesky.protocols.Triggerable
+    :members:
+    :show-inheritance:
+
+
+External Asset Writing Interface
+++++++++++++++++++++++++++++++++
+
+Devices that write their data in external files, rather than returning directly
+from ``read()`` should implement the following interface:
+
+.. autoclass:: bluesky.protocols.WritesExternalAssets
+    :members:
+    :show-inheritance:
+
+The yielded values are a tuple of the document type and the document as a dictionary.
+
+A Resource will be yielded to show that data will be written to an external resource
+like a file on disk:
+
+.. autoclass:: bluesky.protocols.PartialResource
+    :members:
+
+which can optionally include:
+
+.. autoclass:: bluesky.protocols.PartialResourceOptional
+    :members:
+
+While a Datum will be yielded to specify a single frame of data in a Resource:
+
+.. autoclass:: bluesky.protocols.Datum
+    :members:
+
+.. seealso:: https://blueskyproject.io/event-model/external.html
+
+
+Movable (or "Settable") Device
+++++++++++++++++++++++++++++++
 
 The interface of a movable device extends the interface of a readable device
 with the following additional methods and attributes.
 
-
 .. autoclass:: bluesky.protocols.Movable
     :members:
     :show-inheritance:
@@ -102,15 +190,28 @@ with the following additional methods and attributes.
 "Flyer" Interface
 +++++++++++++++++
 
-*For context on what we mean by "flyer", refer to the section on :doc:`async`.*
+*For context on what we mean by "flyer", refer to the section on* :doc:`async`.
 
 The interface of a "flyable" device is separate from the interface of a readable
 or settable device, though there is some overlap.
 
-
 .. autoclass:: bluesky.protocols.Flyable
     :members:
-    :undoc-members:
+    :show-inheritance:
+
+The yielded values from ``collect()`` are partial Event dictionaries:
+
+.. autoclass:: bluesky.protocols.PartialEvent
+    :members:
+
+If any of the data keys are in external assets rather than including the data,
+a ``filled`` key should be present:
+
+.. autoclass:: bluesky.protocols.PartialEventOptional
+    :members:
+
+Flyable devices can also implement :class:`Configurable` if they have
+configuration that only needs to be read once at the start of scan
 
 
 Optional Interfaces
@@ -126,42 +227,47 @@ the required method.
 
 .. autoclass:: bluesky.protocols.Stageable
     :members:
+    :show-inheritance:
 
 .. autoclass:: bluesky.protocols.Subscribable
     :members:
+    :show-inheritance:
 
 .. autoclass:: bluesky.protocols.Pausable
     :members:
+    :show-inheritance:
 
 .. autoclass:: bluesky.protocols.Stoppable
     :members:
+    :show-inheritance:
 
 .. autoclass:: bluesky.protocols.Checkable
     :members:
+    :show-inheritance:
+
+.. autoclass:: bluesky.protocols.HasHints
+    :members:
+    :show-inheritance:
 
-.. autoclass:: bluesky.protocols.Hinted
+.. autoclass:: bluesky.protocols.Hints
     :members:
 
 
-Implementations
----------------
+Checking if an object supports an interface
+-------------------------------------------
 
-Real Hardware
-+++++++++++++
+You can check at runtime if an object supports an interface with ``isinstance``:
+
+.. code-block:: python
+
+    from bluesky.protocols import Readable
+
+    assert isinstance(obj, Readable)
+    obj.read()
+
+This will check that the correct methods exist on the object, and are callable,
+but will not check any types.
+
+There is also a helper function for this:
 
-The `ophyd
-<https://nsls-ii.github.io/ophyd>`_ package implements this interface for
-a wide variety of hardware, communicating using
-`EPICS <http://www.aps.anl.gov/epics/>`_ via the Python bindings
-`pyepics <http://cars9.uchicago.edu/software/python/pyepics3/>`_.Other control
-systems (Tango, LabView, etc.) could be integrated with bluesky in the future
-by implementing this same interface.
-
-Simulated Hardware
-++++++++++++++++++
-
-A toy "test" implementation the interface is included in the
-:mod:`ophyd.sim` module. These implementations act as simulated hardware,
-and we use them extensively in examples, demos, and the test suite. They can
-also be useful for exercising analysis workflows before running a real
-experiment. API documentation is below.
+.. autofunction:: bluesky.protocols.check_supports


Test patch:
diff --git a/bluesky/tests/conftest.py b/bluesky/tests/conftest.py
index 9d5a586e..92254e71 100644
--- a/bluesky/tests/conftest.py
+++ b/bluesky/tests/conftest.py
@@ -1,6 +1,6 @@
 import asyncio
 from distutils.version import LooseVersion
-from bluesky.run_engine import RunEngine, TransitionError
+from bluesky.run_engine import RunEngine, TransitionError, set_bluesky_event_loop
 import numpy as np
 import os
 import pytest
@@ -11,6 +11,7 @@ def RE(request):
     loop = asyncio.new_event_loop()
     loop.set_debug(True)
     RE = RunEngine({}, call_returns_result=request.param, loop=loop)
+    set_bluesky_event_loop(loop)
 
     def clean_event_loop():
         if RE.state not in ('idle', 'panicked'):
diff --git a/bluesky/tests/test_devices.py b/bluesky/tests/test_devices.py
index eac6fa51..116b3cc4 100644
--- a/bluesky/tests/test_devices.py
+++ b/bluesky/tests/test_devices.py
@@ -1,4 +1,7 @@
 import itertools
+from time import time
+from typing import Dict, List
+from bluesky.protocols import Callback, Descriptor, Reading
 
 from bluesky.utils import ancestry, share_ancestor, separate_devices
 from bluesky.plan_stubs import trigger_and_read
@@ -7,6 +10,7 @@ from bluesky import Msg, RunEngineInterrupted
 import pytest
 from bluesky.tests import requires_ophyd, ophyd
 from .utils import DocCollector
+from unittest.mock import ANY
 
 if ophyd:
     from ophyd import Component as Cpt, Device, Signal
@@ -24,6 +28,28 @@ if ophyd:
         x = Cpt(Signal, value=0)
 
 
+class SigNew:
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self._callbacks: List[Callback] = []
+
+    def read(self) -> Dict[str, Reading]:
+        return {}
+
+    def describe(self) -> Dict[str, Descriptor]:
+        return {}
+
+    def subscribe(self, function: Callback) -> None:
+        self._callbacks.append(function)
+
+    def clear_sub(self, function: Callback) -> None:
+        self._callbacks.remove(function)
+
+    def _run_subs(self, *args, **kwargs):
+        for cb in self._callbacks:
+            cb({self.name: dict(value=0, timestamp=time())})
+
+
 @requires_ophyd
 def test_ancestry():
     b = B('', name='b')
@@ -56,22 +82,34 @@ def test_separate_devices():
     assert separate_devices([b1, a]) == [b1, a]
 
 
+@pytest.mark.parametrize('ophyd', ["v1", "v2"])
 @requires_ophyd
-def test_monitor(RE):
+def test_monitor(RE, ophyd):
     docs = []
 
     def collect(name, doc):
         docs.append(doc)
 
-    a = A('', name='a')
+    if ophyd == "v1":
+        sig = A('', name='a').s1
+    else:
+        sig = SigNew(name='a_s1')
 
     def plan():
         yield Msg('open_run')
-        yield Msg('monitor', a.s1)
-        a.s1._run_subs(sub_type='value')
+        yield Msg('monitor', sig)
+        sig._run_subs(sub_type='value')
         yield Msg('close_run')
     RE(plan(), collect)
     assert len(docs) == 4
+    assert docs[2] == {
+        'data': {'a_s1': 0},
+        'descriptor': ANY,
+        'seq_num': 1,
+        'time': pytest.approx(time(), rel=0.1),
+        'timestamps': {'a_s1': pytest.approx(time(), rel=0.1)},
+        'uid': ANY
+    }
 
 
 @requires_ophyd
diff --git a/bluesky/tests/test_flyer.py b/bluesky/tests/test_flyer.py
index f20a8f93..8c4b5b4f 100644
--- a/bluesky/tests/test_flyer.py
+++ b/bluesky/tests/test_flyer.py
@@ -262,7 +262,7 @@ def test_device_redundent_config_reading(RE):
         ]
     )
     assert flyer.call_counts["collect"] == 4
-    assert flyer.call_counts["describe_collect"] == 4
+    assert flyer.call_counts["describe_collect"] == 1
     assert flyer.call_counts["read_configuration"] == 1
     assert flyer.call_counts["describe_configuration"] == 1
     assert flyer.call_counts["kickoff"] == 1
diff --git a/bluesky/tests/test_new_examples.py b/bluesky/tests/test_new_examples.py
index 8170f562..58ac55f3 100644
--- a/bluesky/tests/test_new_examples.py
+++ b/bluesky/tests/test_new_examples.py
@@ -208,6 +208,7 @@ def test_monitor_during_wrapper(hw):
 def test_descriptor_layout_from_monitor(RE, hw):
     collector = []
     det = hw.rand
+    det.start_simulation()
 
     def collect(name, doc):
         if name == 'descriptor':
diff --git a/bluesky/tests/test_run_engine.py b/bluesky/tests/test_run_engine.py
index 6e292d91..fba4cefe 100644
--- a/bluesky/tests/test_run_engine.py
+++ b/bluesky/tests/test_run_engine.py
@@ -1,3 +1,4 @@
+import asyncio
 from event_model import DocumentNames
 import threading
 import types
@@ -272,6 +273,9 @@ def test_redundant_monitors_are_illegal(RE):
         def describe(self):
             return {}
 
+        def read(self):
+            return {}
+
         def subscribe(self, *args, **kwargs):
             pass
 
@@ -1388,16 +1392,17 @@ def test_hints(RE):
             self.parent = None
             self.hints = {'vis': 'placeholder'}
 
-        def read(self):
+        async def read(self):
+            await asyncio.sleep(0.1)
             return {}
 
-        def describe(self):
+        async def describe(self):
             return {}
 
-        def read_configuration(self):
+        async def read_configuration(self):
             return {}
 
-        def describe_configuration(self):
+        async def describe_configuration(self):
             return {}
 
     det = Detector('det')
@@ -1669,6 +1674,9 @@ def test_broken_read_exception(RE):
         def read(self):
             ...
 
+        def trigger(self):
+            ...
+
     obj = Dummy('broken read')
     with pytest.raises(RuntimeError):
         RE([Msg('read', obj)])
@@ -1693,5 +1701,11 @@ def test_thread_name(RE):
             assert threading.current_thread().name == "bluesky-run-engine"
             return Status()
 
+        def describe(self):
+            return {}
+
+        def read(self):
+            return {}
+
     d = MockDevice()
     RE([Msg("trigger", d)])
diff --git a/bluesky/tests/test_simulators.py b/bluesky/tests/test_simulators.py
index dd033a3b..b9e0543d 100644
--- a/bluesky/tests/test_simulators.py
+++ b/bluesky/tests/test_simulators.py
@@ -31,7 +31,7 @@ def test_old_module_name(hw):
         plot_raster_path(plan, 'motor1', 'motor2', probe_size=.3)
 
 
-def test_check_limits(hw):
+def test_check_limits(RE, hw):
     det = hw.det
     motor = hw.motor
     # The motor object does not currently implement limits.
@@ -58,6 +58,12 @@ def test_check_limits(hw):
     check_limits(scan([det], motor, -1, 1, 3))
 
 
+def test_check_limits_needs_RE():
+    with pytest.raises(RuntimeError) as ctx:
+        check_limits([])
+    assert str(ctx.value) == "Bluesky event loop not running"
+
+
 def test_plot_raster_path(hw):
     det = hw.det
     motor1 = hw.motor1
diff --git a/bluesky/tests/test_utils.py b/bluesky/tests/test_utils.py
index c894e294..f29ab1f7 100644
--- a/bluesky/tests/test_utils.py
+++ b/bluesky/tests/test_utils.py
@@ -4,7 +4,10 @@ import numpy as np
 from functools import reduce
 import operator
 
-from bluesky.utils import ensure_generator, Msg, merge_cycler, is_movable, CallbackRegistry
+from bluesky.utils import (
+    ensure_generator, Msg, merge_cycler, is_movable, CallbackRegistry,
+    warn_if_msg_args_or_kwargs
+)
 from cycler import cycler
 
 
@@ -436,3 +439,30 @@ def test_CallbackRegistry_2():
     # Process the signal that is not allowed
     with pytest.raises(ValueError, match=f"Allowed signals are {allowed_sigs}"):
         cb.process("some_signal")
+
+
+def test_msg_args_kwargs_emits_warning_first_time(recwarn):
+    class MyDevice:
+        def kickoff(self, *args, **kwargs):
+            pass
+
+    device = MyDevice()
+    msg = Msg("kickoff")
+
+    assert len(recwarn) == 0
+    warn_if_msg_args_or_kwargs(msg, device.kickoff, (), {"arg": "value"})
+    assert len(recwarn) == 1
+    w = recwarn.pop(PendingDeprecationWarning)
+    assert str(w.message) == """\
+About to call kickoff() with args () and kwargs {'arg': 'value'}.
+In the future the passing of Msg.args and Msg.kwargs down to hardware from
+Msg("kickoff") may be deprecated. If you have a use case for these,
+we would like to know about it, so please open an issue at
+https://github.com/bluesky/bluesky/issues"""
+    assert len(recwarn) == 0
+    # Second time doesn't warn
+    warn_if_msg_args_or_kwargs(msg, device.kickoff, (), {"arg": "value"})
+    assert len(recwarn) == 0
+    # Called without kwargs doesn't warn
+    warn_if_msg_args_or_kwargs(msg, device.kickoff, (), {})
+    assert len(recwarn) == 0