diff --git a/wideq/client.py b/wideq/client.py index b80c909..2702808 100644 --- a/wideq/client.py +++ b/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']) diff --git a/wideq/core.py b/wideq/core.py index 993ca62..2151935 100644 --- a/wideq/core.py +++ b/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.