Nitpicky true and typing overhaul

docs/conf.py

@@ -48,13 +48,14 @@
 
 # If true, Sphinx will warn about all references where the target cannot
 # be found.
-nitpicky = False
+nitpicky = True
 
 # A list of (type, target) tuples (by default empty) that should be ignored when
 # generating warnings in "nitpicky mode". Note that type should include the
 # domain name if present. Example entries would be ('py:func', 'int') or
 # ('envvar', 'LD_LIBRARY_PATH').
 nitpick_ignore = [
+    # Defaults with skeleton.
     ("py:class", "NoneType"),
     ("py:class", "'str'"),
     ("py:class", "'float'"),
@@ -63,6 +64,12 @@
     ("py:class", "'object'"),
     ("py:class", "'id'"),
     ("py:class", "typing_extensions.Literal"),
+    # Added, but custom for pytac.
+    ("py:class", "poly1d"),
+    ("py:class", "PchipInterpolator"),
+    ("py:class", "numpy._typing._dtype_like._SupportsDType"),
+    ("py:class", "numpy._typing._dtype_like._DTypeDict"),
+    ("py:class", "numpy._typing._array_like._ScalarType_co"),
 ]
 
 # Both the class’ and the __init__ method’s docstring are concatenated and
@@ -98,7 +105,11 @@
 
 # This means you can link things like `str` and `asyncio` to the relevant
 # docs in the python documentation.
-intersphinx_mapping = dict(python=("https://docs.python.org/3/", None))
+intersphinx_mapping = {
+    "python": ("https://docs.python.org/3/", None),
+    "numpy": ("https://numpy.org/doc/stable/", None),
+    "scipy": ("https://docs.scipy.org/doc/scipy/", None),
+}
 
 # A dictionary of graphviz graph attributes for inheritance diagrams.
 inheritance_graph_attrs = dict(rankdir="TB")

pyproject.toml

@@ -64,6 +64,7 @@ profile = "black"
 extend-ignore = [
     "E203", # See https://github.com/PyCQA/pycodestyle/issues/373
     "E402", # allow isort:skip
+    "E502", # ignore line length
     "F811", # support typing.overload decorator
     "F722", # allow Annotated[typ, some_func("some string")]
 ]

src/pytac/cothread_cs.py

@@ -1,35 +1,60 @@
 import logging
+from typing import Any, List, Optional, Sequence
 
 from cothread.catools import ca_nothing, caget, caput
 
-from pytac.cs import ControlSystem
+from pytac.cs import AugmentedType, ControlSystem
 from pytac.exceptions import ControlSystemException
 
 
+class AugmentedValue(AugmentedType):
+    """A dummy class for typehinting,
+    taken from aioca.aioca.types and cothread.
+    """
+
+    name: str
+    """Name of the PV used to create this value."""
+    ok: bool
+    """True for normal data, False for error code."""
+
+
 class CothreadControlSystem(ControlSystem):
     """A control system using cothread to communicate with EPICS.
 
     N.B. this is the default control system. It is used to communicate over
     channel access with the hardware in the ring.
 
-    **Methods:**
+    Attributes:
+        _timeout: Timeout in seconds for the caget operations.
+        _wait: Caput operations will wait until the server acknowledges successful
+            completion before returning.
     """
 
-    def __init__(self, timeout=1.0, wait=False):
+    _timeout: float
+    _wait: bool
+
+    def __init__(self, timeout: float = 1.0, wait: bool = False) -> None:
+        """Initialise the Cothread control system.
+
+        Args:
+            _timeout: Timeout in seconds for the caget operation.
+            _wait: Caput operations will wait until the server acknowledges successful
+                completion before returning.
+        """
         self._timeout = timeout
         self._wait = wait
 
-    def get_single(self, pv, throw=True):
+    def get_single(self, pv: str, throw: bool = True) -> Optional[AugmentedType]:
         """Get the value of a given PV.
 
         Args:
-            pv (string): The process variable given as a string. It can be a
-                         readback or a setpoint PV.
-            throw (bool): On failure: if True, raise ControlSystemException; if
-                           False, return None and log a warning.
+            pv: The process variable given as a string. It can be a readback or a
+                setpoint PV.
+            throw: On failure: if True, raise ControlSystemException; if False, None
+                will be returned for any PV that fails and a warning will be logged.
 
         Returns:
-            object: the current value of the given PV.
+            The current value of the given PV.
 
         Raises:
             ControlSystemException: if it cannot connect to the specified PV.
@@ -44,23 +69,24 @@ def get_single(self, pv, throw=True):
                 logging.warning(error_msg)
                 return None
 
-    def get_multiple(self, pvs, throw=True):
+    def get_multiple(
+        self, pvs: Sequence[str], throw: bool = True
+    ) -> List[Optional[AugmentedType]]:
         """Get the value for given PVs.
 
         Args:
-            pvs (sequence): PVs to get values of.
-            throw (bool): On failure: if True, raise ControlSystemException; if
-                           False, None will be returned for any PV that fails
-                           and a warning will be logged.
+            pvs: PVs to get values of.
+            throw: On failure: if True, raise ControlSystemException; if False, None
+                will be returned for any PV that fails and a warning will be logged.
 
         Returns:
-            sequence: the current values of the PVs.
+            The current values of the PVs.
 
         Raises:
             ControlSystemException: if it cannot connect to one or more PVs.
         """
         results = caget(pvs, timeout=self._timeout, throw=False)
-        return_values = []
+        return_values: List[Optional[AugmentedType]] = []
         failures = []
         for result in results:
             if isinstance(result, ca_nothing):
@@ -75,17 +101,17 @@ def get_multiple(self, pvs, throw=True):
             raise ControlSystemException(f"{len(failures)} caget calls failed.")
         return return_values
 
-    def set_single(self, pv, value, throw=True):
+    def set_single(self, pv: str, value: AugmentedType, throw: bool = True) -> bool:
         """Set the value of a given PV.
 
         Args:
-            pv (string): PV to set the value of.
-            value (object): The value to set the PV to.
-            throw (bool): On failure: if True, raise ControlSystemException: if
-                           False, log a warning.
+            pv: PV to set the value of.
+            value: The value to set the PV to.
+            throw: On failure: if True, raise ControlSystemException; if False, None
+                will be returned for any PV that fails and a warning will be logged.
 
         Returns:
-            bool: True for success, False for failure
+            True for success, False for failure
 
         Raises:
             ControlSystemException: if it cannot connect to the specified PV.
@@ -101,20 +127,24 @@ def set_single(self, pv, value, throw=True):
                 logging.warning(error_msg)
                 return False
 
-    def set_multiple(self, pvs, values, throw=True):
+    def set_multiple(
+        self,
+        pvs: Sequence[str],
+        values: Sequence[AugmentedType],
+        throw: bool = True,
+    ) -> Optional[List[bool]]:
         """Set the values for given PVs.
 
         Args:
-            pvs (sequence): PVs to set the values of.
-            values (sequence): values to set to the PVs.
-            throw (bool): On failure, if True raise ControlSystemException, if
-                           False return a list of True and False values
-                           corresponding to successes and failures and log a
-                           warning for each PV that fails.
+            pvs: PVs to set the values of.
+            values: values to set to the PVs.
+            throw: On failure, if True raise ControlSystemException, if False return
+                a list of True and False values corresponding to successes and failures
+                and log a warning for each PV that fails.
 
         Returns:
-            list(bool): True for success, False for failure; only returned if
-                         throw is false and a failure occurs.
+            True for success, False for failure; only returned if throw is false and a
+                failure occurs.
 
         Raises:
             ValueError: if the lists of values and PVs are diffent lengths.
@@ -123,8 +153,8 @@ def set_multiple(self, pvs, values, throw=True):
         if len(pvs) != len(values):
             raise ValueError("Please enter the same number of values as PVs.")
         status = caput(pvs, values, timeout=self._timeout, throw=False, wait=self._wait)
-        return_values = []
-        failures = []
+        return_values: List[bool] = []
+        failures: List[Any] = []
         for stat in status:
             if not stat.ok:
                 return_values.append(False)
@@ -137,3 +167,4 @@ def set_multiple(self, pvs, values, throw=True):
                 raise ControlSystemException(f"{len(failures)} caput calls failed.")
             else:
                 return return_values
+        return None

src/pytac/cs.py

@@ -1,77 +1,87 @@
 """Class representing an abstract control system."""
+import sys
+from typing import List, Optional, Sequence, Sized, SupportsFloat, SupportsInt
+
+if sys.version_info < (3, 8):
+    from typing_extensions import SupportsIndex  # noqa
+else:
+    from typing import SupportsIndex  # noqa
+
+
+class AugmentedType(SupportsFloat, SupportsInt, SupportsIndex, Sized):
+    pass
 
 
 class ControlSystem(object):
     """Abstract base class representing a control system.
 
     A specialised implementation of this class would be used to communicate
     over channel access with the hardware in the ring.
-
-    **Methods:**
     """
 
-    def get_single(self, pv, throw):
+    def get_single(self, pv: str, throw: bool) -> Optional[AugmentedType]:
         """Get the value of a given PV.
 
         Args:
-            pv (string): PV to get the value of.
-                         readback or a setpoint PV.
-            throw (bool): On failure: if True, raise ControlSystemException; if
-                           False, return None and log a warning.
+            pv: PV to get the value of readback or a setpoint PV.
+            throw: On failure: if True, raise ControlSystemException; if False,
+                return None and log a warning.
 
         Returns:
-            object: the current value of the given PV.
+            The current value of the given PV.
 
         Raises:
             ControlSystemException: if it cannot connect to the specified PVs.
         """
         raise NotImplementedError()
 
-    def get_multiple(self, pvs, throw):
+    def get_multiple(
+        self, pvs: Sequence[str], throw: bool
+    ) -> List[Optional[AugmentedType]]:
         """Get the value for given PVs.
 
         Args:
-            pvs (sequence): PVs to get values of.
-            throw (bool): On failure: if True, raise ControlSystemException; if
-                           False, None will be returned for any PV that fails
-                           and a warning will be logged.
+            pvs: PVs to get values of.
+            throw: On failure: if True, raise ControlSystemException; if False, None
+                will be returned for any PV that fails and a warning will be logged.
 
         Returns:
-            list(object): the current values of the PVs.
+            The current values of the PVs.
 
         Raises:
             ControlSystemException: if it cannot connect to the specified PV.
         """
         raise NotImplementedError()
 
-    def set_single(self, pv, value, throw):
+    def set_single(self, pv: str, value: AugmentedType, throw: bool) -> bool:
         """Set the value of a given PV.
 
         Args:
-            pv (string): The PV to set the value of.
-            value (object): The value to set the PV to.
-            throw (bool): On failure: if True, raise ControlSystemException: if
-                           False, log a warning.
+            pv: The PV to set the value of.
+            value: The value to set the PV to.
+            throw: On failure: if True, raise ControlSystemException: if False,
+                log a warning.
 
         Raises:
             ControlSystemException: if it cannot connect to the specified PV.
         """
         raise NotImplementedError()
 
-    def set_multiple(self, pvs, values, throw):
+    def set_multiple(
+        self, pvs: Sequence[str], values: Sequence[AugmentedType], throw: bool
+    ) -> Optional[List[bool]]:
         """Set the values for given PVs.
 
         Args:
-            pvs (sequence): PVs to set the values of.
-            values (sequence): values to set no the PVs.
-            throw (bool): On failure, if True raise ControlSystemException, if
-                           False return a list of True and False values
-                           corresponding to successes and failures and log a
-                           warning for each PV that fails.
+            pvs: PVs to set the values of.
+            values: values to set no the PVs.
+            throw: On failure, if True raise ControlSystemException, if False return a
+                list of True and False values corresponding to successes and failures
+                and log a warning for each PV that fails.
 
         Raises:
-            ValueError: if the PVs or values are not passed in as sequences
-                        or if they have different lengths
+            ValueError: if the PVs or values are not passed in as sequences or if they
+                have different lengths
             ControlSystemException: if it cannot connect to one or more PVs.
         """
         raise NotImplementedError()