Start adding type hints, test with mypy (#228)

* Start adding type hints, test with mypy

* Turn on and address mypy check_untyped_defs

Author: scop

.gitignore

@@ -10,3 +10,4 @@ lib64
 pip-selfcheck.json
 .tox
 .pytest_cache/
+.mypy_cache/

.travis.yml

@@ -6,6 +6,8 @@ matrix:
       env: TOXENV=py34
     - python: "3.4.2"
       env: TOXENV=lint
+    - python: "3.4.2"
+      env: TOXENV=typing
     - python: "3.5"
       env: TOXENV=py35
     - python: "3.6"

netdisco/daikin.py

@@ -2,6 +2,7 @@
 import socket
 
 from datetime import timedelta
+from typing import Dict, List  # noqa: F401
 from urllib.parse import unquote
 
 DISCOVERY_MSG = b"DAIKIN_UDP/common/basic_info"
@@ -18,7 +19,7 @@ class Daikin:
 
     def __init__(self):
         """Initialize the Daikin discovery."""
-        self.entries = []
+        self.entries = []  # type: List[Dict[str, str]]
 
     def scan(self):
         """Scan the network."""
@@ -47,9 +48,9 @@ def update(self):
                 try:
                     data, (address, _) = sock.recvfrom(1024)
 
-                    # pylint: disable=consider-using-dict-comprehension
-                    entry = dict([e.split('=')
-                                  for e in data.decode("UTF-8").split(',')])
+                    entry = {x[0]: x[1] for x in (
+                        e.split('=', 1)
+                        for e in data.decode("UTF-8").split(','))}
 
                     # expecting product, mac, activation code, version
                     if 'ret' not in entry or entry['ret'] != 'OK':

netdisco/discoverables/__init__.py

@@ -1,13 +1,17 @@
 """Provides helpful stuff for discoverables."""
 # pylint: disable=abstract-method
 import ipaddress
+from typing import Dict, TYPE_CHECKING  # noqa: F401
 from urllib.parse import urlparse
 
 from ..const import (
     ATTR_NAME, ATTR_MODEL_NAME, ATTR_HOST, ATTR_PORT, ATTR_SSDP_DESCRIPTION,
     ATTR_SERIAL, ATTR_MODEL_NUMBER, ATTR_HOSTNAME, ATTR_MAC_ADDRESS,
     ATTR_PROPERTIES, ATTR_MANUFACTURER, ATTR_UDN, ATTR_UPNP_DEVICE_TYPE)
 
+if TYPE_CHECKING:
+    from zeroconf import ServiceInfo  # noqa: F401
+
 
 class BaseDiscoverable:
     """Base class for discoverable services or device types."""
@@ -80,7 +84,7 @@ def __init__(self, netdis, typ):
         """Initialize MDNSDiscoverable."""
         self.netdis = netdis
         self.typ = typ
-        self.services = {}
+        self.services = {}  # type: Dict[str, ServiceInfo]
 
         netdis.mdns.register_service(self)
 

netdisco/discoverables/wink.py

@@ -1,13 +1,16 @@
 """Discover Wink hub devices."""
+from typing import List  # noqa: F401
+
 from . import SSDPDiscoverable
+from ..ssdp import UPNPEntry  # noqa: F401
 
 
 class Discoverable(SSDPDiscoverable):
     """Add support for discovering Wink hub devices."""
 
     def get_entries(self):
         """Return all Wink entries."""
-        results = []
+        results = []  # type: List[UPNPEntry]
         results.extend(self.find_by_st('urn:wink-com:device:hub2:2'))
         results.extend(self.find_by_st('urn:wink-com:device:hub:2'))
         results.extend(self.find_by_st('urn:wink-com:device:relay:2'))

netdisco/discovery.py

@@ -123,7 +123,8 @@ def _load_device_support(self):
             module = importlib.import_module(
                 discoverables_format.format(module_name))
 
-            self.discoverables[module_name] = module.Discoverable(self)
+            self.discoverables[module_name] = \
+                getattr(module, 'Discoverable')(self)
 
     def print_raw_data(self):
         """Helper method to show what is discovered in your network."""

netdisco/gdm.py

@@ -8,13 +8,14 @@
 """
 import socket
 import struct
+from typing import Any, Dict, List  # noqa: F401
 
 
 class GDM:
     """Base class to discover GDM services."""
 
     def __init__(self):
-        self.entries = []
+        self.entries = []  # type: List[Dict[str, Any]]
         self.last_scan = None
 
     def scan(self):
@@ -81,13 +82,13 @@ def update(self):
             # Look for responses from all recipients
             while True:
                 try:
-                    data, server = sock.recvfrom(1024)
-                    data = data.decode('utf-8')
+                    bdata, server = sock.recvfrom(1024)
+                    data = bdata.decode('utf-8')
                     if '200 OK' in data.splitlines()[0]:
-                        data = {k: v.strip() for (k, v) in (
+                        ddata = {k: v.strip() for (k, v) in (
                             line.split(':') for line in
                             data.splitlines() if ':' in line)}
-                        self.entries.append({'data': data,
+                        self.entries.append({'data': ddata,
                                              'from': server})
                 except socket.timeout:
                     break

netdisco/lms.py

@@ -1,5 +1,6 @@
 """Squeezebox/Logitech Media server discovery."""
 import socket
+from typing import Dict, List, Union  # noqa: F401
 
 from .const import ATTR_HOST, ATTR_PORT
 
@@ -12,7 +13,7 @@ class LMS:
 
     def __init__(self):
         """Initialize the Logitech discovery."""
-        self.entries = []
+        self.entries = []  # type: List[Dict[str, Union[str, int]]]
         self.last_scan = None
 
     def scan(self):
@@ -49,7 +50,8 @@ def update(self):
                         # Where YY is length of port string (ie 4)
                         # And XXXX is the web interface port (ie 9000)
                         port = None
-                        if data.startswith(b'JSON', 1):
+                        # https://github.com/python/typeshed/pull/2696
+                        if data.startswith(b'JSON', 1):  # type: ignore
                             length = data[5:6][0]
                             port = int(data[0-length:])
                         entries.append({

netdisco/mdns.py

@@ -1,4 +1,6 @@
 """Add support for discovering mDNS services."""
+from typing import List  # noqa: F401
+
 import zeroconf
 
 
@@ -8,8 +10,8 @@ class MDNS:
     def __init__(self):
         """Initialize the discovery."""
         self.zeroconf = None
-        self.services = []
-        self._browsers = []
+        self.services = []  # type: List[zeroconf.ServiceInfo]
+        self._browsers = []  # type: List[zeroconf.ServiceBrowser]
 
     def register_service(self, service):
         """Register a mDNS service."""

netdisco/service.py

@@ -3,6 +3,7 @@
 import threading
 import time
 from collections import defaultdict
+from typing import Any, Callable, Dict, List  # noqa: F401
 
 from .discovery import NetworkDiscovery
 
@@ -25,7 +26,7 @@ def __init__(self, interval=DEFAULT_INTERVAL):
         self.interval = interval
 
         # Listeners for new services
-        self.listeners = []
+        self.listeners = []  # type: List[Callable[[str, Any], None]]
 
         # To track when we have to stop
         self._stop = threading.Event()
@@ -38,7 +39,7 @@ def __init__(self, interval=DEFAULT_INTERVAL):
 
         # Dict to keep track of found services. We do not want to
         # broadcast the same found service twice.
-        self._found = defaultdict(list)
+        self._found = defaultdict(list)  # type: Dict[str, List]
 
     def add_listener(self, listener):
         """Add a listener for new services."""

netdisco/smartglass.py

@@ -3,6 +3,7 @@
 import struct
 import binascii
 from datetime import timedelta
+from typing import Any, Dict, List, Optional, Tuple  # noqa: F401
 
 
 DISCOVERY_PORT = 5050
@@ -25,13 +26,15 @@
 """
 DISCOVERY_CLIENT_TYPE = 4
 
+_Response = Dict[str, Any]
+
 
 class XboxSmartGlass:
     """Base class to discover Xbox SmartGlass devices."""
 
     def __init__(self):
         """Initialize the Xbox SmartGlass discovery."""
-        self.entries = []
+        self.entries = []  # type: List[Tuple[str, Optional[_Response]]]
         self._discovery_payload = self.discovery_packet()
 
     @staticmethod

netdisco/ssdp.py

@@ -4,6 +4,7 @@
 import socket
 import logging
 from datetime import datetime, timedelta
+from typing import Dict, List, Optional, Set  # noqa: F401
 from xml.etree import ElementTree
 
 import requests
@@ -33,7 +34,7 @@ class SSDP:
 
     def __init__(self):
         """Initialize the discovery."""
-        self.entries = []
+        self.entries = []  # type: List[UPNPEntry]
         self.last_scan = None
 
     def scan(self):
@@ -65,7 +66,7 @@ def find_by_device_description(self, values):
         """
         self.update()
 
-        seen = set()
+        seen = set()  # type: Set[Optional[str]]
         results = []
 
         # Make unique based on the location since we don't care about ST here
@@ -100,7 +101,7 @@ def remove_expired(self):
 class UPNPEntry:
     """Found uPnP entry."""
 
-    DESCRIPTION_CACHE = {'_NO_LOCATION': {}}
+    DESCRIPTION_CACHE = {'_NO_LOCATION': {}}  # type: Dict[str, Dict]
 
     def __init__(self, values):
         """Initialize the discovery."""

netdisco/tellstick.py

@@ -2,6 +2,7 @@
 import socket
 from datetime import timedelta
 import logging
+from typing import List, Tuple  # noqa: F401
 
 
 DISCOVERY_PORT = 30303
@@ -15,7 +16,7 @@ class Tellstick:
 
     def __init__(self):
         """Initialize the Tellstick discovery."""
-        self.entries = []
+        self.entries = []  # type: List[Tuple[str]]
 
     def scan(self):
         """Scan the network."""
@@ -43,8 +44,8 @@ def update(self):
                 # expecting product, mac, activation code, version
                 if len(entry) != 4:
                     continue
-                entry = (address,) + tuple(entry)
-                entries.append(entry)
+                entry.insert(0, address)
+                entries.append(tuple(entry))
 
             except socket.timeout:
                 break

netdisco/util.py

@@ -1,5 +1,6 @@
 """Util functions used by Netdisco."""
 from collections import defaultdict
+from typing import Any, Dict, List, Optional  # noqa: F401
 
 
 # Taken from http://stackoverflow.com/a/10077069
@@ -9,21 +10,26 @@ def etree_to_dict(t):
     # strip namespace
     tag_name = t.tag[t.tag.find("}")+1:]
 
-    d = {tag_name: {} if t.attrib else None}
+    d = {
+        tag_name: {} if t.attrib else None
+    }  # type: Dict[str, Optional[Dict[str, Any]]]
     children = list(t)
     if children:
-        dd = defaultdict(list)
+        dd = defaultdict(list)  # type: Dict[str, List]
         for dc in map(etree_to_dict, children):
             for k, v in dc.items():
                 dd[k].append(v)
         d = {tag_name: {k: v[0] if len(v) == 1 else v for k, v in dd.items()}}
+    dt = d[tag_name]
     if t.attrib:
-        d[tag_name].update(('@' + k, v) for k, v in t.attrib.items())
+        assert dt is not None
+        dt.update(('@' + k, v) for k, v in t.attrib.items())
     if t.text:
         text = t.text.strip()
         if children or t.attrib:
             if text:
-                d[tag_name]['#text'] = text
+                assert dt is not None
+                dt['#text'] = text
         else:
             d[tag_name] = text
     return d

requirements.txt

@@ -1,2 +1,3 @@
 zeroconf>=0.21.0
 requests>=2.0
+typing; python_version<'3.5'

setup.cfg

@@ -1,3 +1,16 @@
 [tool:pytest]
 testpaths = tests
 norecursedirs = .git
+
+[mypy]
+check_untyped_defs = true
+# TODO disallow_untyped_calls = true
+# TODO disallow_untyped_defs = true
+follow_imports = silent
+ignore_missing_imports = true
+no_implicit_optional = true
+warn_incomplete_stub = true
+warn_redundant_casts = true
+warn_return_any = true
+warn_unused_configs = true
+warn_unused_ignores = true

tox.ini

@@ -1,5 +1,5 @@
 [tox]
-envlist = py34, py35, py36, py37, lint
+envlist = py34, py35, py36, py37, lint, typing
 skip_missing_interpreters = True
 
 [testenv]
@@ -18,3 +18,9 @@ deps =
 commands =
      flake8 netdisco tests setup.py example_service.py
      pylint netdisco tests
+
+[testenv:typing]
+deps =
+    mypy>=0.650
+commands =
+    mypy netdisco tests setup.py example_service.py