Repository: bluesky/bluesky (python)

Issue:
ABCs for Devices that implement the "Bluesky Interface"
@untzag and @ksunden have suggested adding some ABCs for the expected Bluesky Interfaces. ABCs would be a useful tool for scaffolding implementations of the Readble, Movable, and Flyable interfaces. I would want to avoid accidentally giving the impression that implementation _must_ subclass from them. A key selling point for Bluesky is that other libraries can implement its interfaces and be interoperable without a software dependency. In our documentation, we should emphasize that they are optional, to be used or referred to if they help, or ignored otherwise.

We should also be sure that these ABCs implement https://docs.python.org/3/library/abc.html#abc.ABCMeta.__subclasshook__ so that they will behave correctly if `isinstance` is used.

Pull request description:
Initial ABC for Status, Readable, Movable, Flyable

This provides an initial implementation of [PEP 544 Protocol](https://www.python.org/dev/peps/pep-0544/) Abstract Base Classes for some core interfaces in the bluesky ecosystem.

## Description
The use of protocols allows for both static type checking using mypy (or similar) as well as runtime duck-type checking via isinstance (without requiring that the object being checked _actually_ inherits from the ABC)

Still to do:
- [x] Documentation
- [x] Unit tests
- [x] Confirm action on `configure` which currently active implementations do not actually provide (ophyd flyer, yaqc_bluesky)
- [x] Confirm action for `Status.finished_cb` which is listed as deprecated elsewhere

## Motivation and Context
Closes #1445

This will provide an easy `isinstance` check that still allows duck typed implementations to work as expected, just ensuring that all required methods/properties are present

## How Has This Been Tested?
I intend to add actual unit tests before merging, but initial ad-hoc tests with ophyd simulated hardware to show it works:
(Note, no modifications to ophyd were needed, and none should be except perhaps regarding `configure` on flyers, which is currently not defined, and thus left out of the ABC)

```python
>>> from bluesky import abc
>>> from ophyd import sim
>>> isinstance(sim.motor1, abc.Readable)
True
>>> isinstance(sim.motor1, abc.Movable)
True
>>> isinstance(sim.motor1, abc.Flyable)
False
>>> isinstance(sim.det1, abc.Readable)
True
>>> isinstance(sim.det1, abc.Movable)
False
>>> isinstance(sim.det1, abc.Flyable)
False
>>> isinstance(sim.flyer1, abc.Readable)
False
>>> isinstance(sim.flyer1, abc.Movable)
False
>>> isinstance(sim.flyer1, abc.Flyable)
True
```
<!--
## Screenshots (if appropriate):
-->


Patch:
diff --git a/MANIFEST.in b/MANIFEST.in
index 9a6b1f5c..202cd118 100644
--- a/MANIFEST.in
+++ b/MANIFEST.in
@@ -7,3 +7,4 @@ include README.md
 include requirements.txt
 include setup.py
 include versioneer.py
+include bluesky/py.typed
diff --git a/bluesky/protocols.py b/bluesky/protocols.py
new file mode 100644
index 00000000..ef139b36
--- /dev/null
+++ b/bluesky/protocols.py
@@ -0,0 +1,252 @@
+try:
+    from typing import Protocol, 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]]
+
+
+@runtime_checkable
+class Status(Protocol):
+    done: bool
+    success: bool
+
+    def add_callback(self, callback: Callable[["Status"], NoReturn]) -> NoReturn:
+        """Add a callback function to be called upon completion.
+
+        The function must take the status as an argument.
+
+        If the Status object is done when the function is added, it should be
+        called immediately.
+        """
+        ...
+
+
+@runtime_checkable
+class Readable(Protocol):
+    name: str
+
+    @property
+    def parent(self) -> Optional[Any]:
+        """``None``, or a reference to a parent device."""
+        ...
+
+    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).
+
+        Example return value:
+
+        .. code-block:: python
+
+            OrderedDict(('channel1',
+                         {'value': 5, 'timestamp': 1472493713.271991}),
+                         ('channel2',
+                         {'value': 16, 'timestamp': 1472493713.539238}))
+        """
+        ...
+
+    def describe(self) -> Configuration:
+        """Return an OrderedDict with exactly the same keys as the ``read``
+        method, here mapped to metadata about each field.
+
+        Example return value:
+
+        .. code-block:: python
+
+            OrderedDict(('channel1',
+                         {'source': 'XF23-ID:SOME_PV_NAME',
+                          'dtype': 'number',
+                          'shape': []}),
+                        ('channel2',
+                         {'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``."""
+        ...
+
+
+@runtime_checkable
+class Movable(Readable, Protocol):
+    def set(self, value) -> Status:
+        """Return a ``Status`` that is marked done when the device is done moving."""
+        ...
+
+
+@runtime_checkable
+class Flyable(Protocol):
+    name: str
+
+    @property
+    def parent(self) -> Optional[Any]:
+        """``None``, or a reference to a parent device."""
+        ...
+
+    def kickoff(self) -> Status:
+        """Begin acculumating data.
+
+        Return a ``Status`` and mark it done when acqusition has begun.
+        """
+        ...
+
+    def complete(self) -> Status:
+        """Return a ``Status`` and mark it done when acquisition has completed."""
+        ...
+
+    def collect(self) -> Generator[Dict[str, Any], None, None]:
+        """Yield dictionaries that are partial Event documents.
+
+        They should contain the keys 'time', 'data', and 'timestamps'.
+        A 'uid' is added by the RunEngine.
+        """
+        ...
+
+    def describe_collect(self) -> Dict[str, Configuration]:
+        """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."""
+        ...
+
+
+@runtime_checkable
+class Stageable(Protocol):
+    def stage(self) -> List[Any]:
+        """An optional hook for "setting up" the device for acquisition.
+
+        It should return a list of devices including itself and any other
+        devices that are staged as a result of staging this one.
+        (The ``parent`` attribute expresses this relationship: a device should
+        be staged/unstaged whenever its parent is staged/unstaged.)
+        """
+        ...
+
+    def unstage(self) -> List[Any]:
+        """A hook for "cleaning up" the device after acquisition.
+
+        It should return a list of devices including itself and any other
+        devices that are unstaged as a result of unstaging this one.
+        """
+        ...
+
+
+@runtime_checkable
+class Pausable(Protocol):
+    def pause(self) -> Union[NoReturn, None]:
+        """Perform device-specific work when the RunEngine pauses."""
+        ...
+
+    def resume(self) -> Union[NoReturn, None]:
+        """Perform device-specific work when the RunEngine resumes after a pause."""
+        ...
+
+
+@runtime_checkable
+class Stoppable(Protocol):
+    def stop(self, sucess=True) -> Union[NoReturn, None]:
+        """Safely stop a device that may or may not be in motion.
+
+        The argument ``success`` is a boolean.
+        When ``success`` is true, bluesky is stopping the device as planned
+        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.
+        """
+        ...
+
+
+@runtime_checkable
+class Subscribable(Protocol):
+    def subscribe(self, function: Callable) -> Union[NoReturn, 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.
+
+        Needed for :doc:`monitored <async>`.
+        """
+        ...
+
+    def clear_sub(self, function: Callable) -> Union[NoReturn, None]:
+        """Remove a subscription."""
+        ...
+
+
+@runtime_checkable
+class Checkable(Protocol):
+    def check_value(self, value: Any) -> Union[NoReturn, None]:
+        """Test for a valid setpoint without actually moving.
+
+        This should accept the same arguments as ``set``. It should raise an
+        Exception if the argument represent an illegal setting --- e.g. a
+        position that would move a motor outside its limits or a temperature
+        controller outside of its settable range.
+
+        This method is used by simulators that check limits. If not implemented
+        those simulators should assume all values are valid, but may warn.
+        """
+        ...
+
+
+@runtime_checkable
+class Hinted(Protocol):
+    @property
+    def hints(self) -> Dict:
+        """A dictionary of suggestions for best-effort visualization and processing.
+
+        This does not affect what data is read or saved; it is only
+        a suggestion to enable automated tools to provide helpful information
+        with minimal guidance from the user. See :ref:`hints`.
+        """
+        ...
diff --git a/docs/source/hardware.rst b/docs/source/hardware.rst
index 96511b01..2e99266b 100644
--- a/docs/source/hardware.rst
+++ b/docs/source/hardware.rst
@@ -26,30 +26,19 @@ Status object
 The interface of a "status" object, which the ``RunEngine`` uses to
 asynchronously monitor the compeletion of having triggered or set a device.
 
-.. class:: Status:
+.. autoclass:: bluesky.protocols.Status
+   :members:
+   :undoc-members:
 
-    .. attribute:: done
+If ``success`` is ``False`` when the Status is marked done, this is taken
+to mean, "We have given up." For example, "The motor is stuck and will
+never get where it is going." A ``FailedStatus`` exception will be raised
+inside the RunEngine.
 
-        boolean
+Additionally, ``Status`` objects may (optionally) add a watch function that
+conforms to the following definition
 
-    .. attribute:: success
-
-        boolean
-
-    If ``success`` is ``False`` when the Status is marked done, this is taken
-    to mean, "We have given up." For example, "The motor is stuck and will
-    never get where it is going." A ``FailedStatus`` exception will be raised
-    inside the RunEngine.
-
-    .. attribute:: finished_cb
-
-        a callback function that ``Status`` will call when it is marked done.
-
-    It may be that ``Status`` is done before a function has been attached to
-    ``finished_cb``. In that case, the function should be called as soon as it
-    is attached.
-
-    .. method:: watch(func)
+   .. method:: watch(func)
 
         Subscribe to notifications about progress. Useful for progress bars.
 
@@ -77,182 +66,20 @@ Readable Device
 
 The interface of a readable device:
 
-.. class:: ReadableDevice
-
-    .. attribute:: name
-
-        a human-readable string identifying the device
-
-    .. attribute:: parent
-
-        ``None``, or a reference to a parent device
-
-        See the ``stage`` method below for the operational signifance of
-        ``parent``.
-
-    .. method:: read()
-
-        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)
-        lsits. The timestamps should be UNIX time (seconds since 1970).
-
-        Example return value:
-
-        .. code-block:: python
-
-            OrderedDict(('channel1',
-                         {'value': 5, 'timestamp': 1472493713.271991}),
-                         ('channel2',
-                         {'value': 16, 'timestamp': 1472493713.539238}))
-
-
-    .. method:: describe()
-
-        Return an OrderedDict with exactly the same keys as the ``read``
-        method, here mapped to metadata about each field.
-
-        Example return value:
-
-        .. code-block:: python
-
-            OrderedDict(('channel1',
-                         {'source': 'XF23-ID:SOME_PV_NAME',
-                          'dtype': 'number',
-                          'shape': []}),
-                        ('channel2',
-                         {'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.
-
-    .. method:: trigger()
-
-        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.
-
-    .. method:: read_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.
-
-    .. method:: describe_configuration()
-
-        Same API as ``describe``, but corresponding to the keys in
-        ``read_configuration``.
-
-    .. attribute:: hints
-
-        A dictionary of suggestions for best-effort visualization and
-        processing. This does not affect what data is read or saved; it is only
-        a suggestion to enable automated tools to provide helpful information
-        with minimal guidance from the user. See :ref:`hints`.
-
-    .. method:: configure(*args, **kwargs)
-
-        This can change the device's configuration in an arbitrary way. When
-        the RunEngine calls this method, it also emits a fresh Event Descriptor
-        because it assumes that the configuration in the previous Event
-        Descriptor might no longer be valid.
-
-        Returns a tuple of the *old* result of ``read_configuration()`` and the
-        *new* result of ``read_configuration()``.
-
-    *This concludes the required API. The following are optional.*
-
-    .. method:: stage()
-
-        An optional hook for "setting up" the device for acquisition.
-
-        It should return a list of devices including itself and any other
-        devices that are staged as a result of staging this one.
-        (The ``parent`` attribute expresses this relationship: a device should
-        be staged/unstaged whenever its parent is staged/unstaged.)
-
-    .. method:: unstage()
-
-        A hook for "cleaning up" the device after acquisition.
-
-        It should return a list of devices including itself and any other
-        devices that are unstaged as a result of unstaging this one.
-
-    .. method:: subscribe(function)
-
-        Optional, needed only if the device will be :doc:`monitored <async>`.
-
-        When the device has a new value ready, it should call ``function``
-        asynchronously in a separate thread.
-
-    .. method:: clear_sub(function)
-
-        Remove a subscription. (See ``subscribe``, above.)
-
-    .. method:: pause()
-
-        An optional hook to do some device-specific work when the RunEngine
-        pauses.
-
-    .. method:: resume()
-
-        An optional hook to do some device-specific work when the RunEngine
-        resumes after a pause.
-
+.. autoclass:: bluesky.protocols.Readable
+    :members:
+    :undoc-members:
 
 Movable (or "Settable")  Device
 +++++++++++++++++++++++++++++++
 
 The interface of a movable device extends the interface of a readable device
-with the following additional methods and attributes. The utility function
-:func:`bluesky.utils.is_movable` can be used to check if an object meets the
-expected interface of a Movable.
-
-.. class:: MovableDevice:
-
-    .. method:: set(*args, **kwargs)
-
-        Return a ``Status`` that is marked done when the device is done
-        moving. This is the only *required* method that the Movable interace
-        adds to the  Readable one.
+with the following additional methods and attributes.
 
-    .. method:: stop(success=True)
 
-        Safely stop a device that may or may not be in motion.
-        The argument ``success`` is a boolean.
-        When ``success`` is true, bluesky is stopping the device as planned
-        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.
-
-        Optional: devices that cannot be stopped should not implement this
-        method.
-
-    .. method:: check_value(*args, **kwargs)
-
-       This should accept the same arguments as ``set``. It should raise an
-       Exception if the argument represent an illegal setting --- e.g. a
-       position that would move a motor outside its limits or a temperature
-       controller outside of its settable range.
-
-       Optional: If this method is not present, simulators that check limits
-       such as :func:`bluesky.simulators.check_limits` may issue a warning but
-       should assume that all values are legal.
+.. autoclass:: bluesky.protocols.Movable
+    :members:
+    :show-inheritance:
 
     .. attribute:: position
 
@@ -273,67 +100,41 @@ expected interface of a Movable.
 The interface of a "flyable" device is separate from the interface of a readable
 or settable device, though there is some overlap.
 
-.. class:: FlyableDevice:
-
-    .. method:: kickoff()
-
-       Begin acculumating data. Return a ``Status`` and mark it done when
-       acqusition has begun.
-
-    .. method:: complete()
-
-       Return a ``Status`` and mark it done when acquisition has completed.
-
-    .. method:: collect()
-
-        Yield dictionaries that are partial Event documents. They should
-        contain the keys 'time', 'data', and 'timestamps'. A 'uid' is added by
-        the RunEngine.
-
-    .. method:: describe_collect()
-
-        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.
-
-    *The remaining methods and attributes match ReadableDevice.*
-
-    .. method:: configure(*args, **kwargs)
-
-        same as for a readable device
-
-    .. method:: read_configuration()
-
-        same as for a readable device
-
-    .. method:: describe_configuration()
-
-        same as for a readable device
 
-    .. attribute:: name
+.. autoclass:: bluesky.protocols.Flyable
+    :members:
+    :undoc-members:
 
-        same as for a readable device
 
-    .. attribute:: parent
+Optional Interfaces
+-------------------
 
-        same as for a readable device
+These are additional interfaces for providing *optional* behavior to ``Readable``, ``Movable``,
+and ``Flyable`` devices.
 
-    .. method:: stage()
+The methods described here are either hooks for various plans/RunEngine messages which are
+ignored if not present or required by only a subset of RunEngine messages.
+In the latter case, the RunEngine may error if it tries to use a device which does not define
+the required method.
 
-        optional, same as for a readable device
+.. autoclass:: bluesky.protocols.Stageable
+    :members:
 
-    .. method:: unstage()
+.. autoclass:: bluesky.protocols.Subscribable
+    :members:
 
-        optional, same as for a readable device
+.. autoclass:: bluesky.protocols.Pausable
+    :members:
 
-    .. method:: pause()
+.. autoclass:: bluesky.protocols.Stoppable
+    :members:
 
-        optional, same as for a readable device
+.. autoclass:: bluesky.protocols.Checkable
+    :members:
 
-    .. method:: resume()
+.. autoclass:: bluesky.protocols.Hinted
+    :members:
 
-        optional, same as for a readable device
 
 Implementations
 ---------------
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 422a73df..82bbfd3a 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -11,6 +11,7 @@ jinja2
 lmfit
 matplotlib != 3.0.1  # bug in scatter method
 mongoquery
+mypy
 networkx
 nose
 numpydoc
diff --git a/requirements.txt b/requirements.txt
index d45dff2a..8aa35cc0 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -7,4 +7,5 @@ numpy
 super_state_machine
 toolz
 tqdm>=4.44
+typing-extensions;python_version<'3.8'
 zict
diff --git a/setup.py b/setup.py
index 74ebd3ca..3abd6cc8 100644
--- a/setup.py
+++ b/setup.py
@@ -35,6 +35,7 @@ setuptools.setup(
     license="BSD (3-clause)",
     url="https://github.com/bluesky/bluesky",
     packages=setuptools.find_packages(),
+    package_data={"bluesky": ["py.typed"]},
     python_requires='>={}'.format('.'.join(str(n) for n in min_version)),
     install_requires=requirements,
     classifiers=[


Test patch:
diff --git a/bluesky/py.typed b/bluesky/py.typed
new file mode 100644
index 00000000..e69de29b
diff --git a/bluesky/tests/test_protocols.py b/bluesky/tests/test_protocols.py
new file mode 100644
index 00000000..f1fcca5c
--- /dev/null
+++ b/bluesky/tests/test_protocols.py
@@ -0,0 +1,90 @@
+import tempfile
+
+import mypy.api
+import pytest
+
+from bluesky import protocols as bs_protocols
+from ophyd import sim
+
+
+def test_status():
+    assert isinstance(sim.NullStatus(), bs_protocols.Status)
+    assert isinstance(sim.StatusBase(), bs_protocols.Status)
+
+
+def test_readable():
+    assert isinstance(sim.motor1, bs_protocols.Readable)
+    assert isinstance(sim.det1, bs_protocols.Readable)
+    assert not isinstance(sim.flyer1, bs_protocols.Readable)
+
+
+def test_movable():
+    assert isinstance(sim.motor1, bs_protocols.Movable)
+    assert not isinstance(sim.det1, bs_protocols.Movable)
+    assert not isinstance(sim.flyer1, bs_protocols.Movable)
+
+
+def test_flyable():
+    assert isinstance(sim.flyer1, bs_protocols.Flyable)
+    assert not isinstance(sim.det1, bs_protocols.Flyable)
+    assert not isinstance(sim.motor1, bs_protocols.Flyable)
+
+
+def test_stageable():
+    assert isinstance(sim.det1, bs_protocols.Stageable)
+
+
+def test_pausable():
+    assert isinstance(sim.det1, bs_protocols.Pausable)
+
+
+def test_subscribable():
+    assert isinstance(sim.det1, bs_protocols.Subscribable)
+    assert isinstance(sim.motor1, bs_protocols.Subscribable)
+    assert not isinstance(sim.flyer1, bs_protocols.Subscribable)
+
+
+def test_checkable():
+    assert isinstance(sim.motor1, bs_protocols.Checkable)
+
+
+# I think the commented out tests pass because __getattr__ is implemented, but not sure
+@pytest.mark.skip(reason="ophyd missing py.typed to communicate type hints to mypy")
+@pytest.mark.parametrize(
+    "type_, hardware, pass_",
+    [
+        ("Readable", "ABDetector(name='hi')", True),
+        ("Readable", "SynAxis(name='motor1')", True),
+        ("Readable", "TrivialFlyer()", False),
+        ("Configurable", "ABDetector(name='hi')", True),
+        # ("Movable", "ABDetector(name='hi')", False),  # mypy passed when it shouldn't
+        ("Movable", "SynAxis(name='motor1')", True),
+        ("Movable", "TrivialFlyer()", False),
+        ("Stageable", "ABDetector(name='hi')", True),
+        ("Pausable", "ABDetector(name='hi')", True),
+        ("Checkable", "SynAxis(name='motor1')", True),
+        # ("Flyable", "ABDetector(name='hi')", False),  # mypy passed when it shouldn't
+        # ("Flyable", "SynAxis(name='motor1')", False),  # mypy passed when it shouldn't
+        ("Flyable", "TrivialFlyer()", True),
+        ("Subscribable", "SynAxis(name='motor1')", True),
+    ],
+)
+def test_mypy(type_, hardware, pass_):
+    template = f"""
+from bluesky import protocols as bs_protocols
+from ophyd import sim
+
+var: bs_protocols.{type_} = sim.{hardware}
+"""
+
+    with tempfile.NamedTemporaryFile("wt") as f:
+        f.write(template)
+        f.seek(0)
+        stdout, stderr, exit = mypy.api.run([f.name])
+        # pass true means exit 0, pass false means nonzero exit
+        try:
+            assert exit != pass_
+        except AssertionError:
+            print(stdout)
+            print(stderr)
+            raise