Add type information

This helps later refactoring.
Use mypy to check types.

wideq/client.py

@@ -7,7 +7,7 @@ import logging
 import requests
 import base64
 from collections import namedtuple
-from typing import Any, Optional
+from typing import Any, Dict, Generator, List, Optional
 
 from . import core
 
@@ -25,17 +25,17 @@ class Monitor(object):
     makes one `Monitor` object suitable for long-term monitoring.
     """
 
-    def __init__(self, session, device_id):
+    def __init__(self, session, device_id) -> None:
         self.session = session
         self.device_id = device_id
 
-    def start(self):
+    def start(self) -> None:
         self.work_id = self.session.monitor_start(self.device_id)
 
-    def stop(self):
+    def stop(self) -> None:
         self.session.monitor_stop(self.device_id, self.work_id)
 
-    def poll(self):
+    def poll(self) -> Optional[bytes]:
         """Get the current status data (a bytestring) or None if the
         device is not yet ready.
         """
@@ -49,12 +49,12 @@ class Monitor(object):
             return None
 
     @staticmethod
-    def decode_json(data):
+    def decode_json(data: bytes) -> Dict[str, Any]:
         """Decode a bytestring that encodes JSON status data."""
 
         return json.loads(data.decode('utf8'))
 
-    def poll_json(self):
+    def poll_json(self) -> Optional[Dict[str, Any]]:
         """For devices where status is reported via JSON data, get the
         decoded status result (or None if status is not available).
         """
@@ -62,11 +62,11 @@ class Monitor(object):
         data = self.poll()
         return self.decode_json(data) if data else None
 
-    def __enter__(self):
+    def __enter__(self) -> 'Monitor':
         self.start()
         return self
 
-    def __exit__(self, type, value, tb):
+    def __exit__(self, type, value, tb) -> None:
         self.stop()
 
 
@@ -76,26 +76,27 @@ class Client(object):
     """
 
     def __init__(self, gateway=None, auth=None, session=None,
-                 country=DEFAULT_COUNTRY, language=DEFAULT_LANGUAGE):
+                 country: str = DEFAULT_COUNTRY,
+                 language: str = DEFAULT_LANGUAGE) -> None:
         # The three steps required to get access to call the API.
-        self._gateway = gateway
-        self._auth = auth
-        self._session = session
+        self._gateway: core.Gateway = gateway
+        self._auth: core.Auth = auth
+        self._session: core.Session = session
 
         # The last list of devices we got from the server. This is the
         # raw JSON list data describing the devices.
-        self._devices = None
+        self._devices: List[Dict[str, Any]] = []
 
         # Cached model info data. This is a mapping from URLs to JSON
         # responses.
-        self._model_info = {}
+        self._model_info: Dict[str, Any] = {}
 
         # Locale information used to discover a gateway, if necessary.
-        self._country = country
-        self._language = language
+        self._country: str = country
+        self._language: str = language
 
     @property
-    def gateway(self):
+    def gateway(self) -> core.Gateway:
         if not self._gateway:
             self._gateway = core.Gateway.discover(
                 self._country, self._language
@@ -103,19 +104,19 @@ class Client(object):
         return self._gateway
 
     @property
-    def auth(self):
+    def auth(self) -> core.Auth:
         if not self._auth:
             assert False, "unauthenticated"
         return self._auth
 
     @property
-    def session(self):
+    def session(self) -> core.Session:
         if not self._session:
             self._session, self._devices = self.auth.start_session()
         return self._session
 
     @property
-    def devices(self):
+    def devices(self) -> Generator['DeviceInfo', None, None]:
         """DeviceInfo objects describing the user's devices.
         """
 
@@ -123,7 +124,7 @@ class Client(object):
             self._devices = self.session.get_devices()
         return (DeviceInfo(d) for d in self._devices)
 
-    def get_device(self, device_id):
+    def get_device(self, device_id) -> Optional['DeviceInfo']:
         """Look up a DeviceInfo object by device ID.
 
         Return None if the device does not exist.
@@ -135,7 +136,7 @@ class Client(object):
         return None
 
     @classmethod
-    def load(cls, state):
+    def load(cls, state: Dict[str, Any]) -> 'Client':
         """Load a client from serialized state.
         """
 
@@ -169,10 +170,10 @@ class Client(object):
 
         return client
 
-    def dump(self):
+    def dump(self) -> Dict[str, Any]:
         """Serialize the client state."""
 
-        out = {
+        out: Dict[str, Any] = {
             'model_info': self._model_info,
         }
 
@@ -199,12 +200,13 @@ class Client(object):
 
         return out
 
-    def refresh(self):
+    def refresh(self) -> None:
         self._auth = self.auth.refresh()
         self._session, self._devices = self.auth.start_session()
 
     @classmethod
-    def from_token(cls, refresh_token, country=None, language=None):
+    def from_token(cls, refresh_token,
+                   country=None, language=None) -> 'Client':
         """Construct a client using just a refresh token.
 
         This allows simpler state storage (e.g., for human-written
@@ -220,7 +222,7 @@ class Client(object):
         client.refresh()
         return client
 
-    def model_info(self, device):
+    def model_info(self, device: 'DeviceInfo') -> 'ModelInfo':
         """For a DeviceInfo object, get a ModelInfo object describing
         the model's capabilities.
         """
@@ -266,27 +268,27 @@ class DeviceInfo(object):
     This is populated from a JSON dictionary provided by the API.
     """
 
-    def __init__(self, data):
+    def __init__(self, data: Dict[str, Any]) -> None:
         self.data = data
 
     @property
-    def model_id(self):
+    def model_id(self) -> str:
         return self.data['modelNm']
 
     @property
-    def id(self):
+    def id(self) -> str:
         return self.data['deviceId']
 
     @property
-    def model_info_url(self):
+    def model_info_url(self) -> str:
         return self.data['modelJsonUrl']
 
     @property
-    def name(self):
+    def name(self) -> str:
         return self.data['alias']
 
     @property
-    def type(self):
+    def type(self) -> DeviceType:
         """The kind of device, as a `DeviceType` value."""
 
         return DeviceType(self.data['deviceType'])

wideq/core.py

@@ -7,6 +7,7 @@ import hashlib
 import hmac
 import datetime
 import requests
+from typing import Any, Dict, List
 
 GATEWAY_URL = 'https://kic.lgthinq.com:46030/api/common/gatewayUriList'
 APP_KEY = 'wideq'
@@ -19,11 +20,11 @@ OAUTH_CLIENT_KEY = 'LGAO221A02'
 DATE_FORMAT = '%a, %d %b %Y %H:%M:%S +0000'
 
 
-def gen_uuid():
+def gen_uuid() -> str:
     return str(uuid.uuid4())
 
 
-def oauth2_signature(message, secret):
+def oauth2_signature(message: str, secret: str) -> bytes:
     """Get the base64-encoded SHA-1 HMAC digest of a string, as used in
     OAauth2 request signatures.
 
@@ -37,7 +38,7 @@ def oauth2_signature(message, secret):
     return base64.b64encode(digest)
 
 
-def get_list(obj, key):
+def get_list(obj, key: str) -> List[Dict[str, Any]]:
     """Look up a list using a key from an object.
 
     If `obj[key]` is a list, return it unchanged. If is something else,
@@ -294,7 +295,7 @@ class Auth(object):
 
 
 class Session(object):
-    def __init__(self, auth, session_id):
+    def __init__(self, auth, session_id) -> None:
         self.auth = auth
         self.session_id = session_id
 
@@ -308,7 +309,7 @@ class Session(object):
         url = urljoin(self.auth.gateway.api_root + '/', path)
         return lgedm_post(url, data, self.auth.access_token, self.session_id)
 
-    def get_devices(self):
+    def get_devices(self) -> List[Dict[str, Any]]:
         """Get a list of devices associated with the user's account.
 
         Return a list of dicts with information about the devices.