PK%H8O,wideq/__init__.py"""Reverse-engineered client for the LG SmartThinQ API. """ from .core import * # noqa from .client import * # noqa from .ac import * # noqa __version__ = '1.2.0' PKENPWZ"" wideq/ac.py"""A `Device` class representing air conditioning/climate devices. """ import enum from .client import Device class ACVSwingMode(enum.Enum): """The vertical swing mode for an AC/HVAC device. Blades are numbered vertically from 1 (topmost) to 6. All is 100. """ OFF = "@OFF" ONE = "@1" TWO = "@2" THREE = "@3" FOUR = "@4" FIVE = "@5" SIX = "@6" ALL = "@100" class ACHSwingMode(enum.Enum): """The horizontal swing mode for an AC/HVAC device. Blades are numbered horizontally from 1 (leftmost) to 5. Left half goes from 1-3, and right half goes from 3-5. All is 100. """ OFF = "@OFF" ONE = "@1" TWO = "@2" THREE = "@3" FOUR = "@4" FIVE = "@5" LEFT_HALF = "@13" RIGHT_HALF = "@35" ALL = "@100" class ACMode(enum.Enum): """The operation mode for an AC/HVAC device.""" COOL = "@AC_MAIN_OPERATION_MODE_COOL_W" DRY = "@AC_MAIN_OPERATION_MODE_DRY_W" FAN = "@AC_MAIN_OPERATION_MODE_FAN_W" AI = "@AC_MAIN_OPERATION_MODE_AI_W" HEAT = "@AC_MAIN_OPERATION_MODE_HEAT_W" AIRCLEAN = "@AC_MAIN_OPERATION_MODE_AIRCLEAN_W" ACO = "@AC_MAIN_OPERATION_MODE_ACO_W" AROMA = "@AC_MAIN_OPERATION_MODE_AROMA_W" ENERGY_SAVING = "@AC_MAIN_OPERATION_MODE_ENERGY_SAVING_W" ENERGY_SAVER = "@AC_MAIN_OPERATION_MODE_ENERGY_SAVER_W" class ACFanSpeed(enum.Enum): """The fan speed for an AC/HVAC device.""" SLOW = '@AC_MAIN_WIND_STRENGTH_SLOW_W' SLOW_LOW = '@AC_MAIN_WIND_STRENGTH_SLOW_LOW_W' LOW = '@AC_MAIN_WIND_STRENGTH_LOW_W' LOW_MID = '@AC_MAIN_WIND_STRENGTH_LOW_MID_W' MID = '@AC_MAIN_WIND_STRENGTH_MID_W' MID_HIGH = '@AC_MAIN_WIND_STRENGTH_MID_HIGH_W' HIGH = '@AC_MAIN_WIND_STRENGTH_HIGH_W' POWER = '@AC_MAIN_WIND_STRENGTH_POWER_W' AUTO = '@AC_MAIN_WIND_STRENGTH_AUTO_W' class ACOp(enum.Enum): """Whether a device is on or off.""" OFF = "@AC_MAIN_OPERATION_OFF_W" RIGHT_ON = "@AC_MAIN_OPERATION_RIGHT_ON_W" # This one seems to mean "on"? LEFT_ON = "@AC_MAIN_OPERATION_LEFT_ON_W" ALL_ON = "@AC_MAIN_OPERATION_ALL_ON_W" class ACDevice(Device): """Higher-level operations on an AC/HVAC device, such as a heat pump. """ @property def f2c(self): """Get a dictionary mapping Fahrenheit to Celsius temperatures for this device. Unbelievably, SmartThinQ devices have their own lookup tables for mapping the two temperature scales. You can get *close* by using a real conversion between the two temperature scales, but precise control requires using the custom LUT. """ mapping = self.model.value('TempFahToCel').options return {int(f): c for f, c in mapping.items()} @property def c2f(self): """Get an inverse mapping from Celsius to Fahrenheit. Just as unbelievably, this is not exactly the inverse of the `f2c` map. There are a few values in this reverse mapping that are not in the other. """ mapping = self.model.value('TempCelToFah').options out = {} for c, f in mapping.items(): try: c_num = int(c) except ValueError: c_num = float(c) out[c_num] = f return out def set_celsius(self, c): """Set the device's target temperature in Celsius degrees. """ self._set_control('TempCfg', c) def set_fahrenheit(self, f): """Set the device's target temperature in Fahrenheit degrees. """ self.set_celsius(self.f2c[f]) def set_zones(self, zones): """Turn off or on the device's zones. The `zones` parameter is a list of dicts with these keys: - "No": The zone index. A string containing a number, starting from 1. - "Cfg": Whether the zone is enabled. A string, either "1" or "0". - "State": Whether the zone is open. Also "1" or "0". """ # Ensure at least one zone is enabled: we can't turn all zones # off simultaneously. on_count = sum(int(zone['State']) for zone in zones) if on_count > 0: zone_cmd = '/'.join( '{}_{}'.format(zone['No'], zone['State']) for zone in zones if zone['Cfg'] == '1' ) self._set_control('DuctZone', zone_cmd) def get_zones(self): """Get the status of the zones, including whether a zone is configured. The result is a list of dicts with the same format as described in `set_zones`. """ return self._get_config('DuctZone') def set_fan_speed(self, speed): """Set the fan speed to a value from the `ACFanSpeed` enum. """ speed_value = self.model.enum_value('WindStrength', speed.value) self._set_control('WindStrength', speed_value) def set_horz_swing(self, swing): """Set the horizontal swing to a value from the `ACHSwingMode` enum. """ swing_value = self.model.enum_value('WDirHStep', swing.value) self._set_control('WDirHStep', swing_value) def set_vert_swing(self, swing): """Set the vertical swing to a value from the `ACVSwingMode` enum. """ swing_value = self.model.enum_value('WDirVStep', swing.value) self._set_control('WDirVStep', swing_value) def set_mode(self, mode): """Set the device's operating mode to an `OpMode` value. """ mode_value = self.model.enum_value('OpMode', mode.value) self._set_control('OpMode', mode_value) def set_on(self, is_on): """Turn on or off the device (according to a boolean). """ op = ACOp.RIGHT_ON if is_on else ACOp.OFF op_value = self.model.enum_value('Operation', op.value) self._set_control('Operation', op_value) def get_filter_state(self): """Get information about the filter.""" return self._get_config('Filter') def get_mfilter_state(self): """Get information about the "MFilter" (not sure what this is). """ return self._get_config('MFilter') def get_energy_target(self): """Get the configured energy target data.""" return self._get_config('EnergyDesiredValue') def get_light(self): """Get a Boolean indicating whether the display light is on.""" value = self._get_control('DisplayControl') return value == '0' # Seems backwards, but isn't. def get_volume(self): """Get the speaker volume level.""" value = self._get_control('SpkVolume') return int(value) def poll(self): """Poll the device's current state. Monitoring must be started first with `monitor_start`. Return either an `ACStatus` object or `None` if the status is not yet available. """ # Abort if monitoring has not started yet. if not hasattr(self, 'mon'): return None res = self.mon.poll_json() if res: return ACStatus(self, res) else: return None class ACStatus(object): """Higher-level information about an AC device's current status. """ def __init__(self, ac, data): self.ac = ac self.data = data @staticmethod def _str_to_num(s): """Convert a string to either an `int` or a `float`. Troublingly, the API likes values like "18", without a trailing ".0", for whole numbers. So we use `int`s for integers and `float`s for non-whole numbers. """ f = float(s) if f == int(f): return int(f) else: return f @property def temp_cur_c(self): return self._str_to_num(self.data['TempCur']) @property def temp_cur_f(self): return self.ac.c2f[self.temp_cur_c] @property def temp_cfg_c(self): return self._str_to_num(self.data['TempCfg']) @property def temp_cfg_f(self): return self.ac.c2f[self.temp_cfg_c] def lookup_enum(self, key): return self.ac.model.enum_name(key, self.data[key]) @property def mode(self): return ACMode(self.lookup_enum('OpMode')) @property def fan_speed(self): return ACFanSpeed(self.lookup_enum('WindStrength')) @property def horz_swing(self): return ACHSwingMode(self.lookup_enum('WDirHStep')) @property def vert_swing(self): return ACVSwingMode(self.lookup_enum('WDirVStep')) @property def is_on(self): op = ACOp(self.lookup_enum('Operation')) return op != ACOp.OFF PKPN*55wideq/client.py"""A high-level, convenient abstraction for interacting with the LG SmartThinQ API for most use cases. """ import json import enum import logging import requests import base64 from collections import namedtuple from typing import Any, Optional from . import core DEFAULT_COUNTRY = 'US' DEFAULT_LANGUAGE = 'en-US' #: Represents an unknown enum value. _UNKNOWN = 'Unknown' class Monitor(object): """A monitoring task for a device. This task is robust to some API-level failures. If the monitoring task expires, it attempts to start a new one automatically. This makes one `Monitor` object suitable for long-term monitoring. """ def __init__(self, session, device_id): self.session = session self.device_id = device_id def start(self): self.work_id = self.session.monitor_start(self.device_id) def stop(self): self.session.monitor_stop(self.device_id, self.work_id) def poll(self): """Get the current status data (a bytestring) or None if the device is not yet ready. """ try: return self.session.monitor_poll(self.device_id, self.work_id) except core.MonitorError: # Try to restart the task. self.stop() self.start() return None @staticmethod def decode_json(data): """Decode a bytestring that encodes JSON status data.""" return json.loads(data.decode('utf8')) def poll_json(self): """For devices where status is reported via JSON data, get the decoded status result (or None if status is not available). """ data = self.poll() return self.decode_json(data) if data else None def __enter__(self): self.start() return self def __exit__(self, type, value, tb): self.stop() class Client(object): """A higher-level API wrapper that provides a session more easily and allows serialization of state. """ def __init__(self, gateway=None, auth=None, session=None, country=DEFAULT_COUNTRY, language=DEFAULT_LANGUAGE): # The three steps required to get access to call the API. self._gateway = gateway self._auth = auth self._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 # Cached model info data. This is a mapping from URLs to JSON # responses. self._model_info = {} # Locale information used to discover a gateway, if necessary. self._country = country self._language = language @property def gateway(self): if not self._gateway: self._gateway = core.Gateway.discover( self._country, self._language ) return self._gateway @property def auth(self): if not self._auth: assert False, "unauthenticated" return self._auth @property def session(self): if not self._session: self._session, self._devices = self.auth.start_session() return self._session @property def devices(self): """DeviceInfo objects describing the user's devices. """ if not self._devices: self._devices = self.session.get_devices() return (DeviceInfo(d) for d in self._devices) def get_device(self, device_id): """Look up a DeviceInfo object by device ID. Return None if the device does not exist. """ for device in self.devices: if device.id == device_id: return device return None @classmethod def load(cls, state): """Load a client from serialized state. """ client = cls() if 'gateway' in state: data = state['gateway'] client._gateway = core.Gateway( data['auth_base'], data['api_root'], data['oauth_root'], data.get('country', DEFAULT_COUNTRY), data.get('language', DEFAULT_LANGUAGE), ) if 'auth' in state: data = state['auth'] client._auth = core.Auth( client.gateway, data['access_token'], data['refresh_token'] ) if 'session' in state: client._session = core.Session(client.auth, state['session']) if 'model_info' in state: client._model_info = state['model_info'] if 'country' in state: client._country = state['country'] if 'language' in state: client._language = state['language'] return client def dump(self): """Serialize the client state.""" out = { 'model_info': self._model_info, } if self._gateway: out['gateway'] = { 'auth_base': self._gateway.auth_base, 'api_root': self._gateway.api_root, 'oauth_root': self._gateway.oauth_root, 'country': self._gateway.country, 'language': self._gateway.language, } if self._auth: out['auth'] = { 'access_token': self._auth.access_token, 'refresh_token': self._auth.refresh_token, } if self._session: out['session'] = self._session.session_id out['country'] = self._country out['language'] = self._language return out def refresh(self): self._auth = self.auth.refresh() self._session, self._devices = self.auth.start_session() @classmethod def from_token(cls, refresh_token, country=None, language=None): """Construct a client using just a refresh token. This allows simpler state storage (e.g., for human-written configuration) but it is a little less efficient because we need to reload the gateway servers and restart the session. """ client = cls( country=country or DEFAULT_COUNTRY, language=language or DEFAULT_LANGUAGE, ) client._auth = core.Auth(client.gateway, None, refresh_token) client.refresh() return client def model_info(self, device): """For a DeviceInfo object, get a ModelInfo object describing the model's capabilities. """ url = device.model_info_url if url not in self._model_info: self._model_info[url] = device.load_model_info() return ModelInfo(self._model_info[url]) class DeviceType(enum.Enum): """The category of device.""" REFRIGERATOR = 101 KIMCHI_REFRIGERATOR = 102 WATER_PURIFIER = 103 WASHER = 201 DRYER = 202 STYLER = 203 DISHWASHER = 204 OVEN = 301 MICROWAVE = 302 COOKTOP = 303 HOOD = 304 AC = 401 # Includes heat pumps, etc., possibly all HVAC devices. AIR_PURIFIER = 402 DEHUMIDIFIER = 403 ROBOT_KING = 501 # Robotic vacuum cleaner? ARCH = 1001 MISSG = 3001 SENSOR = 3002 SOLAR_SENSOR = 3102 IOT_LIGHTING = 3003 IOT_MOTION_SENSOR = 3004 IOT_SMART_PLUG = 3005 IOT_DUST_SENSOR = 3006 EMS_AIR_STATION = 4001 AIR_SENSOR = 4003 class DeviceInfo(object): """Details about a user's device. This is populated from a JSON dictionary provided by the API. """ def __init__(self, data): self.data = data @property def model_id(self): return self.data['modelNm'] @property def id(self): return self.data['deviceId'] @property def model_info_url(self): return self.data['modelJsonUrl'] @property def name(self): return self.data['alias'] @property def type(self): """The kind of device, as a `DeviceType` value.""" return DeviceType(self.data['deviceType']) def load_model_info(self): """Load JSON data describing the model's capabilities. """ return requests.get(self.model_info_url).json() BitValue = namedtuple('BitValue', ['options']) EnumValue = namedtuple('EnumValue', ['options']) RangeValue = namedtuple('RangeValue', ['min', 'max', 'step']) #: This is a value that is a reference to another key in the data that is at #: the same level as the `Value` key. ReferenceValue = namedtuple('ReferenceValue', ['reference']) class ModelInfo(object): """A description of a device model's capabilities. """ def __init__(self, data): self.data = data def value(self, name: str): """Look up information about a value. :param name: The name to look up. :returns: One of (`BitValue`, `EnumValue`, `RangeValue`, `ReferenceValue`). :raises ValueError: If an unsupported type is encountered. """ d = self.data['Value'][name] if d['type'] in ('Enum', 'enum'): return EnumValue(d['option']) elif d['type'] == 'Range': return RangeValue( d['option']['min'], d['option']['max'], d['option'].get('step', 1) ) elif d['type'].lower() == 'bit': bit_values = {opt['startbit']: opt['value'] for opt in d['option']} return BitValue(bit_values) elif d['type'].lower() == 'reference': ref = d['option'][0] return ReferenceValue(self.data[ref]) else: raise ValueError("unsupported value type {}".format(d['type'])) def default(self, name): """Get the default value, if it exists, for a given value. """ return self.data['Value'][name]['default'] def enum_value(self, key, name): """Look up the encoded value for a friendly enum name. """ options = self.value(key).options options_inv = {v: k for k, v in options.items()} # Invert the map. return options_inv[name] def enum_name(self, key, value): """Look up the friendly enum name for an encoded value. """ options = self.value(key).options if value not in options: logging.warning( 'Value `%s` for key `%s` not in options: %s. Values from API: ' '%s', value, key, options, self.data['Value'][key]['option']) return _UNKNOWN return options[value] def reference_name(self, key: str, value: Any) -> Optional[str]: """Look up the friendly name for an encoded reference value. :param key: The referenced key. :param value: The value whose name we want to look up. :returns: The friendly name for the referenced value. If no name can be found None will be returned. """ value = str(value) reference = self.value(key).reference if value in reference: return reference[value]['_comment'] return None @property def binary_monitor_data(self): """Check that type of monitoring is BINARY(BYTE). """ return self.data['Monitoring']['type'] == 'BINARY(BYTE)' def decode_monitor_binary(self, data): """Decode binary encoded status data. """ decoded = {} for item in self.data['Monitoring']['protocol']: key = item['value'] value = 0 for v in data[item['startByte']:item['startByte'] + item['length']]: value = (value << 8) + v decoded[key] = str(value) return decoded def decode_monitor_json(self, data): """Decode a bytestring that encodes JSON status data.""" return json.loads(data.decode('utf8')) def decode_monitor(self, data): """Decode status data.""" if self.binary_monitor_data: return self.decode_monitor_binary(data) else: return self.decode_monitor_json(data) class Device(object): """A higher-level interface to a specific device. Unlike `DeviceInfo`, which just stores data *about* a device, `Device` objects refer to their client and can perform operations regarding the device. """ def __init__(self, client: Client, device: DeviceInfo): """Create a wrapper for a `DeviceInfo` object associated with a `Client`. """ self.client = client self.device = device self.model: ModelInfo = client.model_info(device) def _set_control(self, key, value): """Set a device's control for `key` to `value`.""" self.client.session.set_device_controls( self.device.id, {key: value}, ) def _get_config(self, key): """Look up a device's configuration for a given value. The response is parsed as base64-encoded JSON. """ data = self.client.session.get_device_config( self.device.id, key, ) return json.loads(base64.b64decode(data).decode('utf8')) def _get_control(self, key): """Look up a device's control value.""" data = self.client.session.get_device_config( self.device.id, key, 'Control', ) # The response comes in a funky key/value format: "(key:value)". _, value = data[1:-1].split(':') return value def monitor_start(self): """Start monitoring the device's status.""" mon = Monitor(self.client.session, self.device.id) mon.start() self.mon = mon def monitor_stop(self): """Stop monitoring the device's status.""" self.mon.stop() PKVN.. wideq/core.py"""A low-level, general abstraction for the LG SmartThinQ API. """ import base64 import uuid from urllib.parse import urljoin, urlencode, urlparse, parse_qs import hashlib import hmac import datetime import requests GATEWAY_URL = 'https://kic.lgthinq.com:46030/api/common/gatewayUriList' APP_KEY = 'wideq' SECURITY_KEY = 'nuts_securitykey' DATA_ROOT = 'lgedmRoot' SVC_CODE = 'SVC202' CLIENT_ID = 'LGAO221A02' OAUTH_SECRET_KEY = 'c053c2a6ddeb7ad97cb0eed0dcb31cf8' OAUTH_CLIENT_KEY = 'LGAO221A02' DATE_FORMAT = '%a, %d %b %Y %H:%M:%S +0000' def gen_uuid(): return str(uuid.uuid4()) def oauth2_signature(message, secret): """Get the base64-encoded SHA-1 HMAC digest of a string, as used in OAauth2 request signatures. Both the `secret` and `message` are given as text strings. We use their UTF-8 equivalents. """ secret_bytes = secret.encode('utf8') hashed = hmac.new(secret_bytes, message.encode('utf8'), hashlib.sha1) digest = hashed.digest() return base64.b64encode(digest) def get_list(obj, key): """Look up a list using a key from an object. If `obj[key]` is a list, return it unchanged. If is something else, return a single-element list containing it. If the key does not exist, return an empty list. """ try: val = obj[key] except KeyError: return [] if isinstance(val, list): return val else: return [val] class APIError(Exception): """An error reported by the API.""" def __init__(self, code, message): self.code = code self.message = message class NotLoggedInError(APIError): """The session is not valid or expired.""" def __init__(self): pass class NotConnectedError(APIError): """The service can't contact the specified device.""" def __init__(self): pass class TokenError(APIError): """An authentication token was rejected.""" def __init__(self): pass class MonitorError(APIError): """Monitoring a device failed, possibly because the monitoring session failed and needs to be restarted. """ def __init__(self, device_id, code): self.device_id = device_id self.code = code def lgedm_post(url, data=None, access_token=None, session_id=None): """Make an HTTP request in the format used by the API servers. In this format, the request POST data sent as JSON under a special key; authentication sent in headers. Return the JSON data extracted from the response. The `access_token` and `session_id` are required for most normal, authenticated requests. They are not required, for example, to load the gateway server data or to start a session. """ headers = { 'x-thinq-application-key': APP_KEY, 'x-thinq-security-key': SECURITY_KEY, 'Accept': 'application/json', } if access_token: headers['x-thinq-token'] = access_token if session_id: headers['x-thinq-jsessionId'] = session_id res = requests.post(url, json={DATA_ROOT: data}, headers=headers) out = res.json()[DATA_ROOT] # Check for API errors. if 'returnCd' in out: code = out['returnCd'] if code != '0000': message = out['returnMsg'] if code == "0102": raise NotLoggedInError() elif code == "0106": raise NotConnectedError() else: raise APIError(code, message) return out def gateway_info(country, language): """Load information about the hosts to use for API interaction. `country` and `language` are codes, like "US" and "en-US," respectively. """ return lgedm_post( GATEWAY_URL, {'countryCode': country, 'langCode': language}, ) def oauth_url(auth_base, country, language): """Construct the URL for users to log in (in a browser) to start an authenticated session. """ url = urljoin(auth_base, 'login/sign_in') query = urlencode({ 'country': country, 'language': language, 'svcCode': SVC_CODE, 'authSvr': 'oauth2', 'client_id': CLIENT_ID, 'division': 'ha', 'grant_type': 'password', }) return '{}?{}'.format(url, query) def parse_oauth_callback(url): """Parse the URL to which an OAuth login redirected to obtain two tokens: an access token for API credentials, and a refresh token for getting updated access tokens. """ params = parse_qs(urlparse(url).query) return params['access_token'][0], params['refresh_token'][0] def login(api_root, access_token, country, language): """Use an access token to log into the API and obtain a session and return information about the session. """ url = urljoin(api_root + '/', 'member/login') data = { 'countryCode': country, 'langCode': language, 'loginType': 'EMP', 'token': access_token, } return lgedm_post(url, data) def refresh_auth(oauth_root, refresh_token): """Get a new access_token using a refresh_token. May raise a `TokenError`. """ token_url = urljoin(oauth_root, '/oauth2/token') data = { 'grant_type': 'refresh_token', 'refresh_token': refresh_token, } # The timestamp for labeling OAuth requests can be obtained # through a request to the date/time endpoint: # https://us.lgeapi.com/datetime # But we can also just generate a timestamp. timestamp = datetime.datetime.utcnow().strftime(DATE_FORMAT) # The signature for the requests is on a string consisting of two # parts: (1) a fake request URL containing the refresh token, and (2) # the timestamp. req_url = ('/oauth2/token?grant_type=refresh_token&refresh_token=' + refresh_token) sig = oauth2_signature('{}\n{}'.format(req_url, timestamp), OAUTH_SECRET_KEY) headers = { 'lgemp-x-app-key': OAUTH_CLIENT_KEY, 'lgemp-x-signature': sig, 'lgemp-x-date': timestamp, 'Accept': 'application/json', } res = requests.post(token_url, data=data, headers=headers) res_data = res.json() if res_data['status'] != 1: raise TokenError() return res_data['access_token'] class Gateway(object): def __init__(self, auth_base, api_root, oauth_root, country, language): self.auth_base = auth_base self.api_root = api_root self.oauth_root = oauth_root self.country = country self.language = language @classmethod def discover(cls, country, language): gw = gateway_info(country, language) return cls(gw['empUri'], gw['thinqUri'], gw['oauthUri'], country, language) def oauth_url(self): return oauth_url(self.auth_base, self.country, self.language) class Auth(object): def __init__(self, gateway, access_token, refresh_token): self.gateway = gateway self.access_token = access_token self.refresh_token = refresh_token @classmethod def from_url(cls, gateway, url): """Create an authentication using an OAuth callback URL. """ access_token, refresh_token = parse_oauth_callback(url) return cls(gateway, access_token, refresh_token) def start_session(self): """Start an API session for the logged-in user. Return the Session object and a list of the user's devices. """ session_info = login(self.gateway.api_root, self.access_token, self.gateway.country, self.gateway.language) session_id = session_info['jsessionId'] return Session(self, session_id), get_list(session_info, 'item') def refresh(self): """Refresh the authentication, returning a new Auth object. """ new_access_token = refresh_auth(self.gateway.oauth_root, self.refresh_token) return Auth(self.gateway, new_access_token, self.refresh_token) class Session(object): def __init__(self, auth, session_id): self.auth = auth self.session_id = session_id def post(self, path, data=None): """Make a POST request to the API server. This is like `lgedm_post`, but it pulls the context for the request from an active Session. """ url = urljoin(self.auth.gateway.api_root + '/', path) return lgedm_post(url, data, self.auth.access_token, self.session_id) def get_devices(self): """Get a list of devices associated with the user's account. Return a list of dicts with information about the devices. """ return get_list(self.post('device/deviceList'), 'item') def monitor_start(self, device_id): """Begin monitoring a device's status. Return a "work ID" that can be used to retrieve the result of monitoring. """ res = self.post('rti/rtiMon', { 'cmd': 'Mon', 'cmdOpt': 'Start', 'deviceId': device_id, 'workId': gen_uuid(), }) return res['workId'] def monitor_poll(self, device_id, work_id): """Get the result of a monitoring task. `work_id` is a string ID retrieved from `monitor_start`. Return a status result, which is a bytestring, or None if the monitoring is not yet ready. May raise a `MonitorError`, in which case the right course of action is probably to restart the monitoring task. """ work_list = [{'deviceId': device_id, 'workId': work_id}] res = self.post('rti/rtiResult', {'workList': work_list})['workList'] # When monitoring first starts, it usually takes a few # iterations before data becomes available. In the initial # "warmup" phase, `returnCode` is missing from the response. if 'returnCode' not in res: return None # Check for errors. code = res.get('returnCode') # returnCode can be missing. if code != '0000': raise MonitorError(device_id, code) # The return data may or may not be present, depending on the # monitoring task status. if 'returnData' in res: # The main response payload is base64-encoded binary data in # the `returnData` field. This sometimes contains JSON data # and sometimes other binary data. return base64.b64decode(res['returnData']) else: return None def monitor_stop(self, device_id, work_id): """Stop monitoring a device.""" self.post('rti/rtiMon', { 'cmd': 'Mon', 'cmdOpt': 'Stop', 'deviceId': device_id, 'workId': work_id, }) def set_device_controls(self, device_id, values): """Control a device's settings. `values` is a key/value map containing the settings to update. """ return self.post('rti/rtiControl', { 'cmd': 'Control', 'cmdOpt': 'Set', 'value': values, 'deviceId': device_id, 'workId': gen_uuid(), 'data': '', }) def get_device_config(self, device_id, key, category='Config'): """Get a device configuration option. The `category` string should probably either be "Config" or "Control"; the right choice appears to depend on the key. """ res = self.post('rti/rtiControl', { 'cmd': category, 'cmdOpt': 'Get', 'value': key, 'deviceId': device_id, 'workId': gen_uuid(), 'data': '', }) return res['returnData'] PKL O-i66wideq/dishwasher.pyimport enum from typing import Optional from .client import Device from .util import lookup_enum, lookup_reference class DishWasherState(enum.Enum): """The state of the dishwasher device.""" INITIAL = '@DW_STATE_INITIAL_W' RUNNING = '@DW_STATE_RUNNING_W' PAUSED = "@DW_STATE_PAUSE_W" OFF = '@DW_STATE_POWER_OFF_W' COMPLETE = '@DW_STATE_COMPLETE_W' POWER_FAIL = "@DW_STATE_POWER_FAIL_W" DISHWASHER_STATE_READABLE = { 'INITIAL': 'Standby', 'RUNNING': 'Running', 'PAUSED': 'Paused', 'OFF': 'Off', 'COMPLETE': 'Complete', 'POWER_FAIL': 'Power Failed' } class DishWasherProcess(enum.Enum): """The process within the dishwasher state.""" RESERVE = '@DW_STATE_RESERVE_W' RUNNING = '@DW_STATE_RUNNING_W' RINSING = '@DW_STATE_RINSING_W' DRYING = '@DW_STATE_DRYING_W' COMPLETE = '@DW_STATE_COMPLETE_W' NIGHT_DRYING = '@DW_STATE_NIGHTDRY_W' CANCELLED = '@DW_STATE_CANCEL_W' DISHWASHER_PROCESS_READABLE = { 'RESERVE': 'Delayed Start', 'RUNNING': DISHWASHER_STATE_READABLE['RUNNING'], 'RINSING': 'Rinsing', 'DRYING': 'Drying', 'COMPLETE': DISHWASHER_STATE_READABLE['COMPLETE'], 'NIGHT_DRYING': 'Night Drying', 'CANCELLED': 'Cancelled', } # Provide a map to correct typos in the official course names. DISHWASHER_COURSE_MAP = { 'Haeavy': 'Heavy', } class DishWasherDevice(Device): """A higher-level interface for a dishwasher.""" def poll(self) -> Optional['DishWasherStatus']: """Poll the device's current state. Monitoring must be started first with `monitor_start`. :returns: Either a `DishWasherStatus` instance or `None` if the status is not yet available. """ # Abort if monitoring has not started yet. if not hasattr(self, 'mon'): return None data = self.mon.poll() if data: res = self.model.decode_monitor(data) return DishWasherStatus(self, res) else: return None class DishWasherStatus(object): """Higher-level information about a dishwasher's current status. :param dishwasher: The DishWasherDevice instance. :param data: Binary data from the API. """ def __init__(self, dishwasher: DishWasherDevice, data: dict): self.dishwasher = dishwasher self.data = data @property def state(self) -> DishWasherState: """Get the state of the dishwasher.""" return DishWasherState( lookup_enum('State', self.data, self.dishwasher)) @property def readable_state(self) -> str: """Get a human readable state of the dishwasher.""" return DISHWASHER_STATE_READABLE[self.state.name] @property def process(self) -> DishWasherProcess: """Get the process of the dishwasher.""" process = lookup_enum('Process', self.data, self.dishwasher) if process and process != '-': return DishWasherProcess(process) else: return None @property def readable_process(self) -> str: """Get a human readable process of the dishwasher.""" if self.process: return DISHWASHER_PROCESS_READABLE[self.process.name] else: return None @property def is_on(self) -> bool: """Check if the dishwasher is on or not.""" return self.state != DishWasherState.OFF @property def remaining_time(self) -> int: """Get the remaining time in minutes.""" return (int(self.data['Remain_Time_H']) * 60 + int(self.data['Remain_Time_M'])) @property def initial_time(self) -> int: """Get the initial time in minutes.""" return ( int(self.data['Initial_Time_H']) * 60 + int(self.data['Initial_Time_M'])) @property def reserve_time(self) -> int: """Get the reserve time in minutes.""" return ( int(self.data['Reserve_Time_H']) * 60 + int(self.data['Reserve_Time_M'])) @property def course(self) -> str: """Get the current course.""" course = lookup_reference('Course', self.data, self.dishwasher) if course in DISHWASHER_COURSE_MAP: return DISHWASHER_COURSE_MAP[course] else: return course @property def smart_course(self) -> str: """Get the current smart course.""" return lookup_reference('SmartCourse', self.data, self.dishwasher) @property def error(self) -> str: """Get the current error.""" return lookup_reference('Error', self.data, self.dishwasher) PK Optional['DryerStatus']: """Poll the device's current state. Monitoring must be started first with `monitor_start`. :returns: Either a `DryerStatus` instance or `None` if the status is not yet available. """ # Abort if monitoring has not started yet. if not hasattr(self, 'mon'): return None data = self.mon.poll() if data: res = self.model.decode_monitor(data) return DryerStatus(self, res) else: return None class DryerStatus(object): """Higher-level information about a dryer's current status. :param dryer: The DryerDevice instance. :param data: JSON data from the API. """ def __init__(self, dryer: DryerDevice, data: dict): self.dryer = dryer self.data = data def get_bit(self, key: str, index: int) -> str: bit_value = int(self.data[key]) bit_index = 2 ** index mode = bin(bit_value & bit_index) if mode == bin(0): return 'OFF' else: return 'ON' @property def state(self) -> DryerState: """Get the state of the dryer.""" return DryerState(lookup_enum('State', self.data, self.dryer)) @property def previous_state(self) -> DryerState: """Get the previous state of the dryer.""" return DryerState(lookup_enum('PreState', self.data, self.dryer)) @property def dry_level(self) -> DryLevel: """Get the dry level.""" return DryLevel(lookup_enum('DryLevel', self.data, self.dryer)) @property def temperature_control(self) -> TempControl: """Get the temperature control setting.""" return TempControl(lookup_enum('TempControl', self.data, self.dryer)) @property def time_dry(self) -> TimeDry: """Get the time dry setting.""" return TimeDry(lookup_enum('TimeDry', self.data, self.dryer)) @property def is_on(self) -> bool: """Check if the dryer is on or not.""" return self.state != DryerState.OFF @property def remaining_time(self) -> int: """Get the remaining time in minutes.""" return (int(self.data['Remain_Time_H']) * 60 + int(self.data['Remain_Time_M'])) @property def initial_time(self) -> int: """Get the initial time in minutes.""" return ( int(self.data['Initial_Time_H']) * 60 + int(self.data['Initial_Time_M'])) @property def course(self) -> str: """Get the current course.""" return lookup_reference('Course', self.data, self.dryer) @property def smart_course(self) -> str: """Get the current smart course.""" return lookup_reference('SmartCourse', self.data, self.dryer) @property def error(self) -> str: """Get the current error.""" return lookup_reference('Error', self.data, self.dryer) PKPNP_ mm wideq/util.pyfrom typing import TypeVar from .client import Device T = TypeVar('T', bound=Device) def lookup_enum(attr: str, data: dict, device: T): """Looks up an enum value for the provided attr. :param attr: The attribute to lookup in the enum. :param data: The JSON data from the API. :param device: A sub-class instance of a Device. :returns: The enum value. """ return device.model.enum_name(attr, data[attr]) def lookup_reference(attr: str, data: dict, device: T) -> str: """Look up a reference value for the provided attribute. :param attr: The attribute to find the value for. :param data: The JSON data from the API. :param device: A sub-class instance of a Device. :returns: The looked up value. """ value = device.model.reference_name(attr, data[attr]) if value is None: return 'Off' return value PKFKOnLbbwideq/washer.pyimport enum from typing import Optional from .client import Device from .util import lookup_enum, lookup_reference class WasherState(enum.Enum): """The state of the washer device.""" ADD_DRAIN = '@WM_STATE_ADD_DRAIN_W' COMPLETE = '@WM_STATE_COMPLETE_W' DETECTING = '@WM_STATE_DETECTING_W' DETERGENT_AMOUNT = '@WM_STATE_DETERGENT_AMOUNT_W' DRYING = '@WM_STATE_DRYING_W' END = '@WM_STATE_END_W' ERROR_AUTO_OFF = '@WM_STATE_ERROR_AUTO_OFF_W' FRESH_CARE = '@WM_STATE_FRESHCARE_W' FROZEN_PREVENT_INITIAL = '@WM_STATE_FROZEN_PREVENT_INITIAL_W' FROZEN_PREVENT_PAUSE = '@WM_STATE_FROZEN_PREVENT_PAUSE_W' FROZEN_PREVENT_RUNNING = '@WM_STATE_FROZEN_PREVENT_RUNNING_W' INITIAL = '@WM_STATE_INITIAL_W' OFF = '@WM_STATE_POWER_OFF_W' PAUSE = '@WM_STATE_PAUSE_W' PRE_WASH = '@WM_STATE_PREWASH_W' RESERVE = '@WM_STATE_RESERVE_W' RINSING = '@WM_STATE_RINSING_W' RINSE_HOLD = '@WM_STATE_RINSE_HOLD_W' RUNNING = '@WM_STATE_RUNNING_W' SMART_DIAGNOSIS = '@WM_STATE_SMART_DIAG_W' SMART_DIAGNOSIS_DATA = '@WM_STATE_SMART_DIAGDATA_W' SPINNING = '@WM_STATE_SPINNING_W' TCL_ALARM_NORMAL = 'TCL_ALARM_NORMAL' TUBCLEAN_COUNT_ALARM = '@WM_STATE_TUBCLEAN_COUNT_ALRAM_W' class WasherDevice(Device): """A higher-level interface for a washer.""" def poll(self) -> Optional['WasherStatus']: """Poll the device's current state. Monitoring must be started first with `monitor_start`. :returns: Either a `WasherStatus` instance or `None` if the status is not yet available. """ # Abort if monitoring has not started yet. if not hasattr(self, 'mon'): return None data = self.mon.poll() if data: res = self.model.decode_monitor(data) return WasherStatus(self, res) else: return None class WasherStatus(object): """Higher-level information about a washer's current status. :param washer: The WasherDevice instance. :param data: JSON data from the API. """ def __init__(self, washer: WasherDevice, data: dict): self.washer = washer self.data = data @property def state(self) -> WasherState: """Get the state of the washer.""" return WasherState(lookup_enum('State', self.data, self.washer)) @property def previous_state(self) -> WasherState: """Get the previous state of the washer.""" return WasherState(lookup_enum('PreState', self.data, self.washer)) @property def is_on(self) -> bool: """Check if the washer is on or not.""" return self.state != WasherState.OFF @property def remaining_time(self) -> int: """Get the remaining time in minutes.""" return (int(self.data['Remain_Time_H']) * 60 + int(self.data['Remain_Time_M'])) @property def initial_time(self) -> int: """Get the initial time in minutes.""" return ( int(self.data['Initial_Time_H']) * 60 + int(self.data['Initial_Time_M'])) def _lookup_reference(self, attr: str) -> str: """Look up a reference value for the provided attribute. :param attr: The attribute to find the value for. :returns: The looked up value. """ value = self.washer.model.reference_name(attr, self.data[attr]) if value is None: return 'Off' return value @property def course(self) -> str: """Get the current course.""" return lookup_reference('APCourse', self.data, self.washer) @property def smart_course(self) -> str: """Get the current smart course.""" return lookup_reference('SmartCourse', self.data, self.washer) @property def error(self) -> str: """Get the current error.""" return lookup_reference('Error', self.data, self.washer) PK7&Lxa33wideq-1.2.0.dist-info/LICENSEThe MIT License Copyright (c) 2018 Adrian Sampson Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HPOwideq-1.2.0.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,szd&Y)r$[)T&UrPK!H5! wideq-1.2.0.dist-info/METADATA}Vo6οf [2Xb_4_n %ZbC I9=Rv,@ݻwoa4zHo7WbH2yM66Um;H.Š2%6|)Jh:IgSn8җ*L5p]k] `ywu~u_;Qn%t.'R ?-W)5`7⡑Vd|8yNˬ=rKF{ Ofmx^<QJd)w]mN8n9PòǦlK>y?̧t8!"u*w4`qMq [!Am7hp|͔J&#E[%N^cƾvFBdZy hȗQ C5<ȵ΋OJP EFQ)A8:'deJJo/dEkk!-Uf% ^)c3CLwbI&'HBD-yRg N)|,Һ]w.]~u@XFed˸LԒ֊5rʍP1#Qhn)Q$trwY . @s&{˵̶@H-]cC1΀xNk2Vw0RfEte 3T`K xSϤ<{ߧ`6wtT+qOT9}:XCVk1_-b;ۄ[t˹`Wx?24WHW6jy+WBKC@8G]8 6~rFKyl:8%K67sŊ`R:6q{ǓS2+VoHpނx "B^n U7UFݩ$I"P|V)+Td81r4+)lN4Vψ譋볳c-: E]t =  qJqt$aq-O@GT6Kw:_pc~JbI`V[4y.ݝ:]v~9-XtxT<4r jD&y.H\|u6w*Ϙ~WX A`&\1PK!H,1[wideq-1.2.0.dist-info/RECORDuˎ@}? es,PAP. ܄z̴I"QTY;Q}s|KaqE/=ܴJh7x) 'o|p9V,(c LD$ PrFRYޜˡnX`r.VG=j/|u7pʹb8_Peo'Zէn<>GY{~>K;Fo&-}uoXX:ը4^,i/6f䖇6[˭ƾUwXZ!݄E TEKivvm (CK9=DѮ 75Gk@?93N k TVq{y;'q_03vFϾ*!PԗԗS^6<3Zصj;Xc-YY`+G ZQ̗iwAQC0i/+; 28"<( Z^^{t/ҳ5L^@5!M͎-)MͫLq,w>~PK%H8O,wideq/__init__.pyPKENPWZ"" wideq/ac.pyPKPN*55#wideq/client.pyPKVN.. Xwideq/core.pyPKL O-i66wideq/dishwasher.pyPK