Merge pull request #1053 from klauer/enh_generic_component

ENH: Component[T] generic support for type hinting

.github/workflows/testing.yml

@@ -12,7 +12,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.6", "3.7", "3.8", "3.9"]
+        python-version: ["3.8", "3.9"]
       fail-fast: false
     env:
       TEST_CL: pyepics

ophyd/areadetector/base.py

@@ -3,15 +3,18 @@
 import re
 import sys
 import textwrap
-
 from collections import OrderedDict
+from typing import (Any, Callable, ClassVar, DefaultDict, Dict, List, Optional,
+                    Tuple, Type, TypeVar)
+
 import networkx as nx
 import numpy as np
 
+from ..device import Component, Device, DynamicDeviceComponent
+from ..ophydobj import Kind, OphydObject
+from ..signal import (ArrayAttributeSignal, DerivedSignal, EpicsSignal,
+                      EpicsSignalRO)
 from . import docs
-from ..signal import (EpicsSignal, DerivedSignal, EpicsSignalRO)
-from ..device import (Device, Component, DynamicDeviceComponent)
-from ..signal import (ArrayAttributeSignal)
 
 
 class EpicsSignalWithRBV(EpicsSignal):
@@ -134,9 +137,35 @@ def _array_shape_callback(self, **kwargs):
                        **self._metadata)
 
 
-class ADComponent(Component):
-    def __init__(self, cls, suffix=None, *, lazy=True, **kwargs):
-        super().__init__(cls, suffix=suffix, lazy=lazy, **kwargs)
+K = TypeVar("K", bound=OphydObject)
+
+
+class ADComponent(Component[K]):
+    #: Default laziness for the component.  All AreaDetector components are
+    #: by default lazy - as they contain thousands of PVs that aren't
+    #: necesssary for everyday functionality.
+    lazy_default: ClassVar[bool] = True
+
+    #: The attribute name of the component.
+    attr: Optional[str]
+    #: The class to instantiate when the device is created.
+    cls: Type[K]
+    #: Keyword arguments for the device creation.
+    kwargs: Dict[str, Any]
+    #: Lazily create components on access.
+    lazy: bool
+    #: PV or identifier suffix.
+    suffix: Optional[str]
+    #: Documentation string.
+    doc: Optional[str]
+    #: Value to send on ``trigger()``
+    trigger_value: Optional[Any]
+    #: The data acquisition kind.
+    kind: Kind
+    #: Names of kwarg keys to prepend the device PV prefix to.
+    add_prefix: Tuple[str, ...]
+    #: Subscription name -> subscriptions marked by decorator.
+    _subscriptions: DefaultDict[str, List[Callable]]
 
     def find_docs(self, parent_class):
         '''Find all the documentation related to this class, all the way up the

ophyd/areadetector/cam.py

@@ -1,11 +1,12 @@
 import logging
 
-from ..utils import enum
-from .base import (ADBase, ADComponent as ADCpt, ad_group,
-                   EpicsSignalWithRBV as SignalWithRBV)
-from ..signal import (EpicsSignalRO, EpicsSignal)
 from ..device import DynamicDeviceComponent as DDC
-
+from ..signal import EpicsSignal, EpicsSignalRO
+from ..utils import enum
+from .base import ADBase
+from .base import EpicsSignalWithRBV as SignalWithRBV
+from .base import ADComponent as ADCpt
+from .base import ad_group
 # Import FileBase class for cameras that use File PVs in their drivers
 from .plugins import FileBase
 

ophyd/areadetector/detectors.py

@@ -7,9 +7,9 @@
 '''
 import warnings
 
-from .base import (ADBase, ADComponent as C)
 from . import cam
-
+from .base import ADBase
+from .base import ADComponent as C
 
 __all__ = ['DetectorBase',
            'AreaDetector',

ophyd/device.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import collections
 import contextlib
 import functools
@@ -7,20 +9,20 @@
 import operator
 import textwrap
 import time as ttime
+import typing
 import warnings
-
+from collections import OrderedDict, namedtuple
+from collections.abc import Iterable, MutableSequence
 from enum import Enum
-from collections import (OrderedDict, namedtuple)
+from typing import (Any, Callable, ClassVar, DefaultDict, Dict, List, Optional,
+                    Sequence, Tuple, Type, TypeVar, Union)
 
-from .ophydobj import OphydObject, Kind
+from .ophydobj import Kind, OphydObject
 from .signal import Signal
 from .status import DeviceStatus, StatusBase
 from .utils import (ExceptionBundle, RedundantStaging,
-                    doc_annotation_forwarder, underscores_to_camel_case,
-                    getattrs)
-
-from typing import Dict, List, Any, TypeVar, Tuple
-from collections.abc import MutableSequence, Iterable
+                    doc_annotation_forwarder, getattrs,
+                    underscores_to_camel_case)
 
 A, B = TypeVar('A'), TypeVar('B')
 ALL_COMPONENTS = object()
@@ -59,7 +61,10 @@ class Staged(Enum):
                            'ancestors dotted_name item')
 
 
-class Component:
+K = TypeVar("K", bound=OphydObject)
+
+
+class Component(typing.Generic[K]):
     '''A descriptor representing a device component (or signal)
 
     Unrecognized keyword arguments will be passed directly to the component
@@ -87,7 +92,8 @@ def __init__(self, parent=None, **kwargs):
 
     lazy : bool, optional
         Lazily instantiate the signal. If ``False``, the signal will be
-        instantiated upon component instantiation
+        instantiated upon component instantiation.  Defaults to
+        ``component.lazy_default``.
 
     trigger_value : any, optional
         Mark as a signal to be set on trigger. The value is sent to the signal
@@ -102,13 +108,46 @@ def __init__(self, parent=None, **kwargs):
         string to attach to component DvcClass.component.__doc__
     '''
 
-    def __init__(self, cls, suffix=None, *, lazy=False, trigger_value=None,
-                 add_prefix=None, doc=None, kind=Kind.normal, **kwargs):
+    #: Default laziness for the component class.
+    lazy_default: ClassVar[bool] = False
+
+    #: The attribute name of the component.
+    attr: Optional[str]
+    #: The class to instantiate when the device is created.
+    cls: Type[K]
+    #: Keyword arguments for the device creation.
+    kwargs: Dict[str, Any]
+    #: Lazily create components on access.
+    lazy: bool
+    #: PV or identifier suffix.
+    suffix: Optional[str]
+    #: Documentation string.
+    doc: Optional[str]
+    #: Value to send on ``trigger()``
+    trigger_value: Optional[Any]
+    #: The data acquisition kind.
+    kind: Kind
+    #: Names of kwarg keys to prepend the device PV prefix to.
+    add_prefix: Tuple[str, ...]
+    #: Subscription name -> subscriptions marked by decorator.
+    _subscriptions: DefaultDict[str, List[Callable]]
+
+    def __init__(
+        self,
+        cls: Type[K],
+        suffix: Optional[str] = None,
+        *,
+        lazy: Optional[bool] = None,
+        trigger_value: Optional[Any] = None,
+        add_prefix: Optional[Sequence[str]] = None,
+        doc: Optional[str] = None,
+        kind: Union[str, Kind] = Kind.normal,
+        **kwargs
+    ):
         self.attr = None  # attr is set later by the device when known
-        self.parent = None  # parent is also to be set later when known
         self.cls = cls
         self.kwargs = kwargs
-        self.lazy = lazy  # False if self.is_device else lazy  (TODO)
+        self.lazy = lazy if lazy is not None else self.lazy_default
         self.suffix = suffix
         self.doc = doc
         self.trigger_value = trigger_value  # TODO discuss
@@ -119,7 +158,18 @@ def __init__(self, cls, suffix=None, *, lazy=False, trigger_value=None,
         self.add_prefix = tuple(add_prefix)
         self._subscriptions = collections.defaultdict(list)
 
-    def __set_name__(self, owner, attr_name):
+    def _get_class_from_annotation(self) -> Optional[Type[K]]:
+        """Get a class from the Component[cls] annotation."""
+        annotation = getattr(self, "__orig_class__", None)
+        if not annotation:
+            return None
+
+        args = typing.get_args(annotation)
+        if not args or not len(args) == 1:
+            return None
+        return args[0]
+
+    def __set_name__(self, owner, attr_name: str):
         self.attr = attr_name
         if self.doc is None:
             self.doc = self.make_docstring(owner)
@@ -215,7 +265,19 @@ def __repr__(self):
 
     __str__ = __repr__
 
-    def __get__(self, instance, owner):
+    @typing.overload
+    def __get__(self, instance: None, owner: type) -> Component[K]:
+        ...
+
+    @typing.overload
+    def __get__(self, instance: Device, owner: type) -> K:
+        ...
+
+    def __get__(
+        self,
+        instance: Optional[Device],
+        owner: type,
+    ) -> Union[Component, K]:
         if instance is None:
             return self
 
@@ -265,7 +327,7 @@ def sub_value(self, func):
         return self.subscriptions('value')(func)
 
 
-class FormattedComponent(Component):
+class FormattedComponent(Component[K]):
     '''A Component which takes a dynamic format string
 
     This differs from Component in that the parent prefix is not automatically
@@ -303,7 +365,7 @@ def maybe_add_prefix(self, instance, kw, suffix):
         return suffix.format(**format_dict)
 
 
-class DynamicDeviceComponent(Component):
+class DynamicDeviceComponent(Component["Device"]):
     '''An Device component that dynamically creates an ophyd Device
 
     Parameters

ophyd/sim.py

@@ -1090,7 +1090,7 @@ def make_fake_device(cls):
             if isinstance(cpt, DDCpt):
                 # Make a regular Cpt out of the DDC, as it already has
                 # been generated
-                fake_cpt = Cpt(cls=cpt.cls, suffix=cpt.suffix,
+                fake_cpt = Cpt(cpt.cls, suffix=cpt.suffix,
                                lazy=cpt.lazy,
                                trigger_value=cpt.trigger_value,
                                kind=cpt.kind, add_prefix=cpt.add_prefix,

ophyd/tests/test_areadetector.py

@@ -269,6 +269,9 @@ class MyDetector(SingleTrigger, SimDetector):
     assert det.cam is det.get_plugin_by_asyn_port(det.cam.port_name.get())
     assert det.roi1 is det.get_plugin_by_asyn_port(det.roi1.port_name.get())
 
+
+@pytest.mark.adsim
+def test_get_plugin_by_asyn_port_nested(ad_prefix, cleanup):
     # Support nested plugins
     class PluginGroup(Device):
         tiff1 = Cpt(TIFFPlugin, 'TIFF1:')
@@ -279,13 +282,14 @@ class MyDetector(SingleTrigger, SimDetector):
         stats1 = Cpt(StatsPlugin, 'Stats1:')
 
     nested_det = MyDetector(ad_prefix, name='nested_test')
+    cleanup.add(nested_det)
 
     nested_det.stats1.nd_array_port.put(nested_det.roi1.port_name.get())
     nested_det.plugins.tiff1.nd_array_port.put(nested_det.cam.port_name.get())
     nested_det.roi1.nd_array_port.put(nested_det.cam.port_name.get())
     nested_det.stats1.nd_array_port.put(nested_det.roi1.port_name.get())
 
-    det.validate_asyn_ports()
+    nested_det.validate_asyn_ports()
 
     tiff = nested_det.plugins.tiff1
     assert tiff is nested_det.get_plugin_by_asyn_port(tiff.port_name.get())

ophyd/tests/test_device.py

@@ -1,18 +1,17 @@
 import logging
-import pytest
 from unittest.mock import Mock
 
 import numpy as np
+import pytest
 
-from ophyd import (Device, Component, FormattedComponent)
-from ophyd.signal import (Signal, AttributeSignal, ArrayAttributeSignal,
-                          ReadOnlyError)
+from ophyd import Component, Device, FormattedComponent
 from ophyd.device import (ComponentWalk, create_device_from_components,
-                          required_for_connection, wait_for_lazy_connection,
-                          do_not_wait_for_lazy_connection)
+                          do_not_wait_for_lazy_connection,
+                          required_for_connection, wait_for_lazy_connection)
+from ophyd.signal import (ArrayAttributeSignal, AttributeSignal, ReadOnlyError,
+                          Signal, SignalRO)
 from ophyd.utils import ExceptionBundle
 
-
 logger = logging.getLogger(__name__)
 
 
@@ -624,7 +623,8 @@ class MyDevice(Device):
 
 
 def test_dotted_name():
-    from ophyd import Device, Component as Cpt
+    from ophyd import Component as Cpt
+    from ophyd import Device
     from ophyd.sim import SynSignal
 
     class Inner(Device):
@@ -790,3 +790,21 @@ class Target(Host, MixIn):
 
     with pytest.raises(RuntimeError):
         t.b
+
+
+def test_annotated_device():
+
+    class MyDevice(Device):
+        cpt1 = Component[Signal](Signal)
+        cpt2 = Component[SignalRO](SignalRO)
+        cpt3 = Component[SignalRO](SignalRO)
+        cpt4 = Component(SignalRO)
+
+    dev = MyDevice(name="dev")
+    assert isinstance(dev.cpt1, Signal)
+    assert isinstance(dev.cpt2, SignalRO)
+    assert MyDevice.cpt1._get_class_from_annotation() is Signal
+    assert MyDevice.cpt2._get_class_from_annotation() is SignalRO
+    assert MyDevice.cpt3._get_class_from_annotation() is SignalRO
+    assert MyDevice.cpt3.cls is SignalRO
+    assert MyDevice.cpt4._get_class_from_annotation() is None

setup.py

@@ -9,7 +9,7 @@
 # NOTE: This file must remain Python 2 compatible for the foreseeable future,
 # to ensure that we error out properly for people with outdated setuptools
 # and/or pip.
-min_version = (3, 6)
+min_version = (3, 8)
 if sys.version_info < min_version:
     error = """
 ophyd does not support Python {0}.{1}.
@@ -57,6 +57,6 @@
       },
       classifiers=[
           "Development Status :: 5 - Production/Stable",
-          "Programming Language :: Python :: 3.6",
-          "Programming Language :: Python :: 3.7",
+          "Programming Language :: Python :: 3.8",
+          "Programming Language :: Python :: 3.9",
       ])