Introduce optional static typing

Type hints are added within the comment-based syntax that should work
with Python 2.

Author: tueda

.gitignore

@@ -7,6 +7,8 @@ dist/
 *.egg/
 *.eggs/
 
+.mypy_cache
+
 doc/_build/
 
 .coverage*

form/__init__.py

@@ -14,10 +14,14 @@
 4*p1.p2*p3.p4-4*p1.p3*p2.p4+4*p1.p4*p2.p3
 """
 
+if False:
+    from typing import Optional, Sequence, Union  # noqa: F401
+
 from .formlink import FormError, FormLink  # noqa: F401
 
 
 def open(args=None, keep_log=False):
+    # type: (Optional[Union[str, Sequence[str]]], Union[bool, int]) -> FormLink
     """Open a connection to FORM and return a link object.
 
     Open a connection to a new FORM process and return a

form/datapath.py

@@ -4,16 +4,26 @@
 import pkgutil
 import sys
 
+if False:
+    from typing import Optional  # noqa: F401
+
 
 def get_data_path(package, resource):
+    # type: (str, str) -> str
     """Return the full file path of a resource of a package."""
     loader = pkgutil.get_loader(package)
     if loader is None or not hasattr(loader, 'get_data'):
-        return None
+        raise PackageResourceError("Failed to load package: '{0}'".format(
+            package))
     mod = sys.modules.get(package) or loader.load_module(package)
     if mod is None or not hasattr(mod, '__file__'):
-        return None
+        raise PackageResourceError("Failed to load module: '{0}'".format(
+            package))
     parts = resource.split('/')
     parts.insert(0, os.path.dirname(mod.__file__))
     resource_name = os.path.join(*parts)
     return resource_name
+
+
+class PackageResourceError(IOError):
+    """Package resource not found."""

form/formlink.py

@@ -12,6 +12,12 @@
 from .io import PushbackReader, set_nonblock
 from .six import PY32, string_types
 
+if False:
+    from typing import Any, IO, MutableSequence, Optional, Sequence, Tuple, Union, overload  # noqa: E501, F401
+if True:
+    def overload(f):  # type: ignore  # noqa: D103, F811
+        return None
+
 
 class FormLink(object):
     """Connection to a FORM process."""
@@ -25,18 +31,20 @@ class FormLink(object):
     _PROMPT = '\n__READY__\n'
 
     def __init__(self, args=None, keep_log=False):
+        # type: (Optional[Union[str, Sequence[str]]], Union[bool, int]) -> None
         """Open a connection to a FORM process."""
         self._closed = True
-        self._head = None
-        self._log = None
-        self._childpid = None
-        self._formpid = None
-        self._parentin = None
-        self._parentout = None
-        self._loggingin = None
+        self._head = None       # type: Optional[str]
+        self._log = None        # type: Optional[MutableSequence[str]]
+        self._childpid = None   # type: Optional[int]
+        self._formpid = None    # type: Optional[int]
+        self._parentin = None   # type: Optional[PushbackReader]
+        self._parentout = None  # type: Optional[IO[str]]
+        self._loggingin = None  # type: Optional[PushbackReader]
         self.open(args, keep_log)
 
     def __del__(self):
+        # type: () -> None
         """Destructor.
 
         Free the connection to the FORM process if it still exists. Since in
@@ -51,14 +59,17 @@ def __del__(self):
             pass
 
     def __enter__(self):
+        # type: () -> FormLink
         """Enter the runtime context."""
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
+        # type: (Any, Any, Any) -> None
         """Exit the runtime context."""
         self.close()
 
     def open(self, args=None, keep_log=False):
+        # type: (Optional[Union[str, Sequence[str]]], Union[bool, int]) -> None
         """Open a connection to FORM.
 
         Open a connection to a new FORM process. The opened connection should
@@ -94,11 +105,12 @@ def open(self, args=None, keep_log=False):
                 args = os.environ['FORM']
             else:
                 args = 'form'
-
         if isinstance(args, string_types):
             args = shlex.split(args)  # Split the arguments.
         elif isinstance(args, (list, tuple)):
             args = list(args)  # As a modifiable mutable object.
+        else:
+            raise TypeError("invalid args = {0}".format(args))
 
         self.close()
 
@@ -146,8 +158,6 @@ def open(self, args=None, keep_log=False):
 
             set_nonblock(fd_parentin)
             set_nonblock(fd_loggingin)
-            parentin = PushbackReader(parentin)
-            loggingin = PushbackReader(loggingin)
 
             self._closed = False
             self._head = head
@@ -164,9 +174,9 @@ def open(self, args=None, keep_log=False):
                 parentout.write('#-\n')
             self._childpid = pid
             self._formpid = formpid
-            self._parentin = parentin
+            self._parentin = PushbackReader(parentin)
             self._parentout = parentout
-            self._loggingin = loggingin
+            self._loggingin = PushbackReader(loggingin)
         else:
             # child process
 
@@ -202,6 +212,7 @@ def open(self, args=None, keep_log=False):
             os._exit(0)
 
     def close(self):
+        # type: () -> None
         """Close the connection to FORM.
 
         Close the connection to the FORM process established by :meth:`open`.
@@ -212,7 +223,13 @@ def close(self):
         self._close()
 
     def _close(self, term=False, kill=False):
+        # type: (float, float) -> None
         if not self._closed:
+            assert self._childpid is not None
+            assert self._formpid is not None
+            assert self._parentin is not None
+            assert self._parentout is not None
+            assert self._loggingin is not None
             try:
                 # We ignore broken pipes.
                 try:
@@ -236,7 +253,9 @@ def _close(self, term=False, kill=False):
                     # finish shortly.
 
                     def wait(timeout):  # timeout <= 0 means no wait
+                        # type: (float) -> bool
                         # Wait for the child to finish.
+                        assert self._childpid is not None
                         t = 0.0
                         dt = 0.01
                         if timeout > 0:
@@ -281,11 +300,13 @@ def wait(timeout):  # timeout <= 0 means no wait
                 self._loggingin = None
 
     def kill(self):
+        # type: () -> None
         """Kill the FORM process and close the connection."""
         self._close(kill=-1)  # Kill it immediately.
 #       self._close(term=-1, kill=1)
 
     def write(self, script):
+        # type: (str) -> None
         """Send a script to FORM.
 
         Write the given script to the communication channel to FORM. It could
@@ -296,10 +317,12 @@ def write(self, script):
             raise IOError('tried to write to closed connection')
         script = script.strip()
         if script:
+            assert self._parentout is not None
             self._parentout.write(script)
             self._parentout.write('\n')
 
     def flush(self):
+        # type: () -> None
         """Flush the channel to FORM.
 
         Flush the communication channel to FORM. Because :meth:`write` is
@@ -308,9 +331,71 @@ def flush(self):
         """
         if self._closed:
             raise IOError('tried to flush closed connection')
+        assert self._parentout is not None
         self._parentout.flush()
 
-    def read(self, *names):
+    @overload
+    def read(self, name):  # noqa: D102
+        # type: (str) -> str
+        pass
+
+    @overload  # noqa: F811
+    def read(self, name1, name2):  # noqa: D102
+        # type: (str, str) -> Tuple[str, str]
+        pass
+
+    @overload  # noqa: F811
+    def read(self, name1, name2, name3):  # noqa: D102
+        # type: (str, str, str) -> Tuple[str, str, str]
+        pass
+
+    @overload  # noqa: F811
+    def read(self, name1, name2, name3, name4):  # noqa: D102
+        # type: (str, str, str, str) -> Tuple[str, str, str, str]
+        pass
+
+    @overload  # noqa: F811
+    def read(self, name1, name2, name3, name4, name5):  # noqa: D102
+        # type: (str, str, str, str, str) -> Tuple[str, str, str, str, str]
+        pass
+
+    @overload  # noqa: F811
+    def read(self, name1, name2, name3, name4, name5, name6):  # noqa: D102
+        # type: (str, str, str, str, str, str) -> Tuple[str, str, str, str, str, str]  # noqa: E501
+        pass
+
+    @overload  # noqa: F811
+    def read(self, name1, name2, name3, name4, name5, name6, name7):  # noqa: D102, E501
+        # type: (str, str, str, str, str, str, str) -> Tuple[str, str, str, str, str, str, str]  # noqa: E501
+        pass
+
+    @overload  # noqa: F811
+    def read(self, name1, name2, name3, name4, name5, name6, name7, name8):  # noqa: D102, E501
+        # type: (str, str, str, str, str, str, str, str) -> Tuple[str, str, str, str, str, str, str, str]  # noqa: E501
+        pass
+
+    @overload  # noqa: F811
+    def read(self, name1, name2, name3, name4, name5, name6, name7, name8, name9):  # noqa: D102, E501
+        # type: (str, str, str, str, str, str, str, str, str) -> Tuple[str, str, str, str, str, str, str, str, str]  # noqa: E501
+        pass
+
+    @overload  # noqa: F811
+    def read(self, names):  # noqa: D102
+        # type: (Sequence[str]) -> Sequence[str]
+        pass
+
+    @overload  # noqa: F811
+    def read(self, *names):  # noqa: D102
+        # type: (str) -> Sequence[str]
+        pass
+
+    @overload  # noqa: F811
+    def read(self, *names):  # noqa: D102
+        # type: (Any) -> Any
+        pass
+
+    def read(self, *names):  # type: ignore  # noqa: F811
+        # type: (Any) -> Any
         r"""Read results from FORM.
 
         Wait for a response of FORM to obtain the results specified by
@@ -373,6 +458,10 @@ def read(self, *names):
         if any(not isinstance(x, string_types) for x in names):
             return [self.read(x) for x in names]
 
+        assert self._parentin is not None
+        assert self._parentout is not None
+        assert self._loggingin is not None
+
         for e in names:
             if len(e) >= 2 and e[0] == '`' and e[-1] == "'":
                 self._parentout.write(
@@ -461,22 +550,26 @@ def read(self, *names):
 
     @property
     def closed(self):
+        # type: () -> bool
         """Return True if the connection is closed."""
         return self._closed
 
     @property
     def head(self):
+        # type: () -> str
         """Return the first line of the FORM output."""
+        assert self._head is not None
         return self._head
 
     @property
     def _dateversion(self):
+        # type: () -> int
         """Return the build/revision date as an integer "yyyymmdd"."""
         import re
         if self._head:
-            m = re.search(r'(?<=\()(.*)(?=\))', self._head)
-            if m:
-                s = re.split(r'[, ]+', m.group(0))
+            ma = re.search(r'(?<=\()(.*)(?=\))', self._head)
+            if ma:
+                s = re.split(r'[, ]+', ma.group(0))
                 if len(s) >= 3:
                     # month
                     month_names = ('Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',

form/io.py

@@ -3,8 +3,12 @@
 import fcntl
 import os
 
+if False:
+    from typing import IO  # noqa: F401
+
 
 def set_nonblock(fd):
+    # type: (int) -> None
     """Set the given file descriptor to non-blocking mode."""
     fcntl.fcntl(fd,
                 fcntl.F_SETFL,
@@ -15,25 +19,30 @@ class PushbackReader(object):
     """Wrapper for streams with push back operations."""
 
     def __init__(self, raw):
+        # type: (IO[str]) -> None
         """Initialize the reader."""
         self._raw = raw
         self._buf = ''
 
     def close(self):
+        # type: () -> None
         """Close the stream."""
         self._raw.close()
 
     def fileno(self):
+        # type: () -> int
         """Return the file descriptor."""
         return self._raw.fileno()
 
     def read(self):
+        # type: () -> str
         """Read data from the stream."""
         s = self._buf + self._raw.read()
         self._buf = ''
         return s
 
     def unread(self, s):
+        # type: (str) -> None
         """Push back a string.
 
         Push back the given string to the internal buffer, which will be used
@@ -42,6 +51,7 @@ def unread(self, s):
         self._buf = s + self._buf
 
     def read0(self):
+        # type: () -> str
         """Read the pushed-back string.
 
         Read a string pushed-back by a previous ``unread()``. No call to

setup.cfg

@@ -4,8 +4,22 @@ test=nosetests
 [bdist_wheel]
 universal=1
 
-[flake8]
-ignore=D203
+[mypy]
+follow_imports=normal
+ignore_missing_imports=True
+# strict mode
+disallow_untyped_calls=True
+disallow_untyped_defs=True
+disallow_incomplete_defs=True
+check_untyped_defs=True
+disallow_subclassing_any=True
+disallow_untyped_decorators=True
+warn_redundant_casts=True
+warn_return_any=True
+warn_unused_ignores=True
+warn_unused_configs=True
+no_implicit_optional=True
+strict_optional=True
 
 [nosetests]
 detailed-errors=1