diff --git a/resources/wideq/example.py b/resources/wideq/example.py old mode 100644 new mode 100755 index b88e0e6..f2ab39d --- a/resources/wideq/example.py +++ b/resources/wideq/example.py @@ -1,8 +1,17 @@ +#!/usr/bin/env python3 + import wideq import json import time import argparse import sys +import re +import os.path +import logging +from typing import List + +STATE_FILE = 'wideq_state.json' +LOGGER = logging.getLogger("wideq.example") def authenticate(gateway): @@ -44,11 +53,9 @@ def mon(client, device_id): res = model.decode_monitor(data) except ValueError: print('status data: {!r}'.format(data)) - sys.stdout.flush() else: for key, value in res.items(): try: - desc = None desc = model.value(key) except KeyError: print('- {}: {}'.format(key, value)) @@ -60,7 +67,6 @@ def mon(client, device_id): print('- {0}: {1} ({2.min}-{2.max})'.format( key, value, desc, )) - sys.stdout.flush() except KeyboardInterrupt: pass @@ -80,6 +86,11 @@ def ac_mon(client, device_id): try: ac.monitor_start() + except wideq.core.NotConnectedError: + print('Device not available.') + return + + try: while True: time.sleep(1) state = ac.poll() @@ -87,15 +98,14 @@ def ac_mon(client, device_id): print( '{1}; ' '{0.mode.name}; ' - 'cur {0.temp_cur_c}°C; ' - 'cfg {0.temp_cfg_c}°C; ' + 'cur {0.temp_cur_f}°F; ' + 'cfg {0.temp_cfg_f}°F; ' 'fan speed {0.fan_speed.name}' .format( state, 'on' if state.is_on else 'off' ) ) - sys.stdout.flush() except KeyboardInterrupt: pass @@ -124,7 +134,7 @@ def set_temp(client, device_id, temp): """Set the configured temperature for an AC device.""" ac = wideq.ACDevice(client, _force_device(client, device_id)) - ac.set_celsius(int(temp)) + ac.set_fahrenheit(int(temp)) def turn(client, device_id, on_off): @@ -136,26 +146,17 @@ def turn(client, device_id, on_off): def ac_config(client, device_id): ac = wideq.ACDevice(client, _force_device(client, device_id)) + print(ac.supported_operations) + print(ac.supported_on_operation) print(ac.get_filter_state()) print(ac.get_mfilter_state()) print(ac.get_energy_target()) + print(ac.get_power(), " watts") + print(ac.get_outdoor_power(), " watts") print(ac.get_volume()) print(ac.get_light()) print(ac.get_zones()) -def set_speed(client, device_id, speed): - ac = wideq.ACDevice(client, _force_device(client, device_id)) - speed_mapping = { - '12.5': wideq.ACFanSpeed.SLOW, #Not supported - '25': wideq.ACFanSpeed.SLOW_LOW, #Not supported - '37.5': wideq.ACFanSpeed.LOW, - '50': wideq.ACFanSpeed.LOW_MID, - '62.5': wideq.ACFanSpeed.MID, - '75': wideq.ACFanSpeed.MID_HIGH, - '87.5': wideq.ACFanSpeed.HIGH, - '100': wideq.ACFanSpeed.POWER #Not supported - } - ac.set_fan_speed(speed_mapping[speed]) EXAMPLE_COMMANDS = { 'ls': ls, @@ -164,24 +165,32 @@ def set_speed(client, device_id, speed): 'set-temp': set_temp, 'turn': turn, 'ac-config': ac_config, - 'set_speed': set_speed, } def example_command(client, cmd, args): - func = EXAMPLE_COMMANDS[cmd] + func = EXAMPLE_COMMANDS.get(cmd) + if not func: + LOGGER.error("Invalid command: '%s'.\n" + "Use one of: %s", cmd, ', '.join(EXAMPLE_COMMANDS)) + return func(client, *args) -def example(country, language, state_file, cmd, args): - # Load the current state for the example. +def example(country: str, language: str, verbose: bool, + cmd: str, args: List[str]) -> None: + if verbose: + wideq.set_log_level(logging.DEBUG) + # Load the current state for the example. try: - with open(state_file) as f: + with open(STATE_FILE) as f: + LOGGER.debug("State file found '%s'", os.path.abspath(STATE_FILE)) state = json.load(f) - except IOError as e: - print(e) + except IOError: state = {} + LOGGER.debug("No state file found (tried: '%s')", + os.path.abspath(STATE_FILE)) client = wideq.Client.load(state) if country: @@ -200,48 +209,62 @@ def example(country, language, state_file, cmd, args): break except wideq.NotLoggedInError: - print('Session expired.') + LOGGER.info('Session expired.') client.refresh() except UserError as exc: - print(exc.msg, file=sys.stderr) + LOGGER.error(exc.msg) sys.exit(1) # Save the updated state. state = client.dump() - with open(state_file, 'w') as f: + with open(STATE_FILE, 'w') as f: json.dump(state, f) + LOGGER.debug("Wrote state file '%s'", os.path.abspath(STATE_FILE)) -def main(): +def main() -> None: """The main command-line entry point. """ parser = argparse.ArgumentParser( description='Interact with the LG SmartThinQ API.' ) parser.add_argument('cmd', metavar='CMD', nargs='?', default='ls', - help='one of {}'.format(', '.join(EXAMPLE_COMMANDS))) + help=f'one of: {", ".join(EXAMPLE_COMMANDS)}') parser.add_argument('args', metavar='ARGS', nargs='*', help='subcommand arguments') parser.add_argument( '--country', '-c', - help='country code for account (default: {})' - .format(wideq.DEFAULT_COUNTRY) + help=f'country code for account (default: {wideq.DEFAULT_COUNTRY})', + default=wideq.DEFAULT_COUNTRY ) parser.add_argument( '--language', '-l', - help='language code for the API (default: {})' - .format(wideq.DEFAULT_LANGUAGE) + help=f'language code for the API (default: {wideq.DEFAULT_LANGUAGE})', + default=wideq.DEFAULT_LANGUAGE ) parser.add_argument( - '--state-file', '-s', - help='Absolute path to the state file (default: wideq_state.json)' - .format('wideq_state.json') + '--verbose', '-v', + help='verbose mode to help debugging', + action='store_true', default=False ) args = parser.parse_args() - example(args.country, args.language, args.state_file, args.cmd, args.args) + country_regex = re.compile(r"^[A-Z]{2,3}$") + if not country_regex.match(args.country): + LOGGER.error("Country must be two or three letters" + " all upper case (e.g. US, NO, KR) got: '%s'", + args.country) + exit(1) + language_regex = re.compile(r"^[a-z]{2,3}-[A-Z]{2,3}$") + if not language_regex.match(args.language): + LOGGER.error("Language must be a combination of language" + " and country (e.g. en-US, no-NO, kr-KR)" + " got: '%s'", + args.language) + exit(1) + example(args.country, args.language, args.verbose, args.cmd, args.args) if __name__ == '__main__': diff --git a/resources/wideq/wideq/__init__.py b/resources/wideq/wideq/__init__.py index 1bf8ed8..6e1ca71 100644 --- a/resources/wideq/wideq/__init__.py +++ b/resources/wideq/wideq/__init__.py @@ -3,5 +3,9 @@ from .core import * # noqa from .client import * # noqa from .ac import * # noqa +from .dishwasher import * # noqa +from .dryer import * # noqa +from .refrigerator import * # noqa +from .washer import * # noqa -__version__ = '1.0.1' +__version__ = '1.4.0' diff --git a/resources/wideq/wideq/ac.py b/resources/wideq/wideq/ac.py index a20ffd5..c6252eb 100644 --- a/resources/wideq/wideq/ac.py +++ b/resources/wideq/wideq/ac.py @@ -3,6 +3,48 @@ import enum from .client import Device +from .util import lookup_enum +from .core import FailedRequestError + + +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): @@ -32,15 +74,46 @@ class ACFanSpeed(enum.Enum): HIGH = '@AC_MAIN_WIND_STRENGTH_HIGH_W' POWER = '@AC_MAIN_WIND_STRENGTH_POWER_W' AUTO = '@AC_MAIN_WIND_STRENGTH_AUTO_W' + NATURE = '@AC_MAIN_WIND_STRENGTH_NATURE_W' + R_LOW = '@AC_MAIN_WIND_STRENGTH_LOW_RIGHT_W' + R_MID = '@AC_MAIN_WIND_STRENGTH_MID_RIGHT_W' + R_HIGH = '@AC_MAIN_WIND_STRENGTH_HIGH_RIGHT_W' + L_LOW = '@AC_MAIN_WIND_STRENGTH_LOW_LEFT_W' + L_MID = '@AC_MAIN_WIND_STRENGTH_MID_LEFT_W' + L_HIGH = '@AC_MAIN_WIND_STRENGTH_HIGH_LEFT_W' + L_LOWR_LOW = '@AC_MAIN_WIND_STRENGTH_LOW_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_LOW_RIGHT_W' + L_LOWR_MID = '@AC_MAIN_WIND_STRENGTH_LOW_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_MID_RIGHT_W' + L_LOWR_HIGH = '@AC_MAIN_WIND_STRENGTH_LOW_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_HIGH_RIGHT_W' + L_MIDR_LOW = '@AC_MAIN_WIND_STRENGTH_MID_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_LOW_RIGHT_W' + L_MIDR_MID = '@AC_MAIN_WIND_STRENGTH_MID_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_MID_RIGHT_W' + L_MIDR_HIGH = '@AC_MAIN_WIND_STRENGTH_MID_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_HIGH_RIGHT_W' + L_HIGHR_LOW = '@AC_MAIN_WIND_STRENGTH_HIGH_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_LOW_RIGHT_W' + L_HIGHR_MID = '@AC_MAIN_WIND_STRENGTH_HIGH_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_MID_RIGHT_W' + L_HIGHR_HIGH = '@AC_MAIN_WIND_STRENGTH_HIGH_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_HIGH_RIGHT_W' + AUTO_2 = '@AC_MAIN_WIND_STRENGTH_AUTO_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_AUTO_RIGHT_W' + POWER_2 = '@AC_MAIN_WIND_STRENGTH_POWER_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_POWER_RIGHT_W' + LONGPOWER = '@AC_MAIN_WIND_STRENGTH_LONGPOWER_LEFT_W|' \ + 'AC_MAIN_WIND_STRENGTH_LONGPOWER_RIGHT_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" + RIGHT_ON = "@AC_MAIN_OPERATION_RIGHT_ON_W" # Right fan only. + LEFT_ON = "@AC_MAIN_OPERATION_LEFT_ON_W" # Left fan only. + ALL_ON = "@AC_MAIN_OPERATION_ALL_ON_W" # Both fans (or only fan) on. class ACDevice(Device): @@ -81,6 +154,44 @@ def c2f(self): out[c_num] = f return out + @property + def supported_operations(self): + """Get a list of the ACOp Operations the device supports. + """ + + mapping = self.model.value('Operation').options + return [ACOp(o) for i, o in mapping.items()] + + @property + def supported_on_operation(self): + """Get the most correct "On" operation the device supports. + :raises ValueError: If ALL_ON is not supported, but there are + multiple supported ON operations. If a model raises this, + its behaviour needs to be determined so this function can + make a better decision. + """ + + operations = self.supported_operations + operations.remove(ACOp.OFF) + + # This ON operation appears to be supported in newer AC models + if ACOp.ALL_ON in operations: + return ACOp.ALL_ON + + # Older models, or possibly just the LP1419IVSM, do not support ALL_ON, + # instead advertising only a single operation of RIGHT_ON. + # Thus, if there's only one ON operation, we use that. + if len(operations) == 1: + return operations[0] + + # Hypothetically, the API could return multiple ON operations, neither + # of which are ALL_ON. This will raise in that case, as we don't know + # what that model will expect us to do to turn everything on. + # Or, this code will never actually be reached! We can only hope. :) + raise ValueError( + f"could not determine correct 'on' operation:" + f" too many reported operations: '{str(operations)}'") + def set_celsius(self, c): """Set the device's target temperature in Celsius degrees. """ @@ -131,6 +242,20 @@ def set_fan_speed(self, speed): 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. """ @@ -142,7 +267,7 @@ 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 = self.supported_on_operation if is_on else ACOp.OFF op_value = self.model.enum_value('Operation', op.value) self._set_control('Operation', op_value) @@ -162,17 +287,37 @@ def get_energy_target(self): return self._get_config('EnergyDesiredValue') + def get_outdoor_power(self): + """Get instant power usage in watts of the outdoor unit""" + + value = self._get_config('OutTotalInstantPower') + return value['OutTotalInstantPower'] + + def get_power(self): + """Get the instant power usage in watts of the whole unit""" + + value = self._get_config('InOutInstantPower') + return value['InOutInstantPower'] + 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. + try: + value = self._get_control('DisplayControl') + return value == '0' # Seems backwards, but isn't. + except FailedRequestError: + # Device does not support reporting display light status. + # Since it's probably not changeable the it must be on. + return True def get_volume(self): """Get the speaker volume level.""" - value = self._get_control('SpkVolume') - return int(value) + try: + value = self._get_control('SpkVolume') + return int(value) + except FailedRequestError: + return 0 # Device does not support volume control. def poll(self): """Poll the device's current state. @@ -232,18 +377,23 @@ def temp_cfg_c(self): 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')) + return ACMode(lookup_enum('OpMode', self.data, self.ac)) @property def fan_speed(self): - return ACFanSpeed(self.lookup_enum('WindStrength')) + return ACFanSpeed(lookup_enum('WindStrength', self.data, self.ac)) + + @property + def horz_swing(self): + return ACHSwingMode(lookup_enum('WDirHStep', self.data, self.ac)) + + @property + def vert_swing(self): + return ACVSwingMode(lookup_enum('WDirVStep', self.data, self.ac)) @property def is_on(self): - op = ACOp(self.lookup_enum('Operation')) + op = ACOp(lookup_enum('Operation', self.data, self.ac)) return op != ACOp.OFF diff --git a/resources/wideq/wideq/client.py b/resources/wideq/wideq/client.py index 0d664e1..e154320 100644 --- a/resources/wideq/wideq/client.py +++ b/resources/wideq/wideq/client.py @@ -7,14 +7,13 @@ import requests import base64 from collections import namedtuple -from typing import Any, Optional +from typing import Any, Dict, Generator, List, Optional from . import core -DEFAULT_COUNTRY = 'US' -DEFAULT_LANGUAGE = 'en-US' #: Represents an unknown enum value. _UNKNOWN = 'Unknown' +LOGGER = logging.getLogger("wideq.client") class Monitor(object): @@ -25,17 +24,17 @@ class Monitor(object): makes one `Monitor` object suitable for long-term monitoring. """ - def __init__(self, session, device_id): + def __init__(self, session: core.Session, device_id: str) -> 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 +48,12 @@ def poll(self): 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 +61,11 @@ def poll_json(self): 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() @@ -75,27 +74,31 @@ class Client(object): and allows serialization of state. """ - def __init__(self, gateway=None, auth=None, session=None, - country=DEFAULT_COUNTRY, language=DEFAULT_LANGUAGE): + def __init__(self, + gateway: Optional[core.Gateway] = None, + auth: Optional[core.Auth] = None, + session: Optional[core.Session] = None, + country: str = core.DEFAULT_COUNTRY, + language: str = core.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: Optional[core.Gateway] = gateway + self._auth: Optional[core.Auth] = auth + self._session: Optional[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 +106,19 @@ def gateway(self): 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 +126,7 @@ def devices(self): 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,19 +138,14 @@ def get_device(self, device_id): return None @classmethod - def load(cls, state): + def load(cls, state: Dict[str, Any]) -> 'Client': """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), - ) + client._gateway = core.Gateway.deserialize(state['gateway']) if 'auth' in state: data = state['auth'] @@ -169,27 +167,18 @@ def load(cls, state): 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, } 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, - } + out['gateway'] = self._gateway.serialize() if self._auth: - out['auth'] = { - 'access_token': self._auth.access_token, - 'refresh_token': self._auth.refresh_token, - } + out['auth'] = self._auth.serialize() if self._session: out['session'] = self._session.session_id @@ -199,12 +188,13 @@ def dump(self): 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 @@ -213,14 +203,14 @@ def from_token(cls, refresh_token, country=None, language=None): """ client = cls( - country=country or DEFAULT_COUNTRY, - language=language or DEFAULT_LANGUAGE, + country=country or core.DEFAULT_COUNTRY, + language=language or core.DEFAULT_LANGUAGE, ) client._auth = core.Auth(client.gateway, None, refresh_token) 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 +256,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']) @@ -303,6 +293,7 @@ def load_model_info(self): #: 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']) +StringValue = namedtuple('StringValue', ['comment']) class ModelInfo(object): @@ -317,7 +308,7 @@ def value(self, name: str): :param name: The name to look up. :returns: One of (`BitValue`, `EnumValue`, `RangeValue`, - `ReferenceValue`). + `ReferenceValue`, `StringValue`). :raises ValueError: If an unsupported type is encountered. """ d = self.data['Value'][name] @@ -334,6 +325,12 @@ def value(self, name: str): elif d['type'].lower() == 'reference': ref = d['option'][0] return ReferenceValue(self.data[ref]) + elif d['type'].lower() == 'string': + return StringValue(d.get('_comment', '')) + else: + raise ValueError( + f"unsupported value name: '{name}'" + f" type: '{str(d['type'])}' data: '{str(d)}'") def default(self, name): """Get the default value, if it exists, for a given value. @@ -352,7 +349,7 @@ def enum_name(self, key, value): """ options = self.value(key).options if value not in options: - logging.warning( + LOGGER.warning( 'Value `%s` for key `%s` not in options: %s. Values from API: ' '%s', value, key, options, self.data['Value'][key]['option']) return _UNKNOWN @@ -417,7 +414,7 @@ def __init__(self, client: Client, device: DeviceInfo): """ self.client = client self.device = device - self.model = client.model_info(device) + self.model: ModelInfo = client.model_info(device) def _set_control(self, key, value): """Set a device's control for `key` to `value`.""" @@ -435,7 +432,8 @@ def _get_config(self, key): self.device.id, key, ) - return json.loads(base64.b64decode(data).decode('utf8')) + # Added fix to correct malformed JSON! + return json.loads(base64.b64decode(data).decode('utf8').replace('{[', '[').replace(']}', ']')) def _get_control(self, key): """Look up a device's control value.""" diff --git a/resources/wideq/wideq/core.py b/resources/wideq/wideq/core.py index 285203d..a4b6729 100644 --- a/resources/wideq/wideq/core.py +++ b/resources/wideq/wideq/core.py @@ -7,7 +7,10 @@ import hmac import datetime import requests -import urllib3 +import logging +from typing import Any, Dict, List, Tuple +from requests.adapters import HTTPAdapter +from requests.packages.urllib3.util.retry import Retry GATEWAY_URL = 'https://kic.lgthinq.com:46030/api/common/gatewayUriList' APP_KEY = 'wideq' @@ -18,22 +21,79 @@ OAUTH_SECRET_KEY = 'c053c2a6ddeb7ad97cb0eed0dcb31cf8' OAUTH_CLIENT_KEY = 'LGAO221A02' DATE_FORMAT = '%a, %d %b %Y %H:%M:%S +0000' +DEFAULT_COUNTRY = 'US' +DEFAULT_LANGUAGE = 'en-US' -# Fix for dh-key-too-small error in python 3.7.x -requests.packages.urllib3.disable_warnings() -requests.packages.urllib3.util.ssl_.DEFAULT_CIPHERS += 'HIGH:!DH:!aNULL' -try: - requests.packages.urllib3.contrib.pyopenssl.DEFAULT_SSL_CIPHER_LIST += 'HIGH:!DH:!aNULL' -except AttributeError: - # no pyopenssl support used / needed / available - pass +RETRY_COUNT = 5 # Anecdotally this seems sufficient. +RETRY_FACTOR = 0.5 +RETRY_STATUSES = (502, 503, 504) -def gen_uuid(): +def get_wideq_logger() -> logging.Logger: + level = logging.INFO + fmt = "%(asctime)s %(levelname)s [%(name)s] %(message)s" + datefmt = "%Y-%m-%d %H:%M:%S" + logger = logging.getLogger("wideq") + logger.setLevel(level) + + try: + import colorlog # type: ignore + colorfmt = f"%(log_color)s{fmt}%(reset)s" + handler = colorlog.StreamHandler() + handler.setFormatter( + colorlog.ColoredFormatter( + colorfmt, + datefmt=datefmt, + reset=True, + log_colors={ + "DEBUG": "cyan", + "INFO": "green", + "WARNING": "yellow", + "ERROR": "red", + "CRITICAL": "red", + }, + ) + ) + except ImportError: + handler = logging.StreamHandler() + handler.setFormatter(logging.Formatter(fmt=fmt, datefmt=datefmt)) + + logger.addHandler(handler) + return logger + + +LOGGER = get_wideq_logger() + + +def retry_session(): + """Get a Requests session that retries HTTP and HTTPS requests. + """ + # Adapted from: + # https://www.peterbe.com/plog/best-practice-with-retries-with-requests + session = requests.Session() + retry = Retry( + total=RETRY_COUNT, + read=RETRY_COUNT, + connect=RETRY_COUNT, + backoff_factor=RETRY_FACTOR, + status_forcelist=RETRY_STATUSES, + ) + adapter = HTTPAdapter(max_retries=retry) + session.mount('http://', adapter) + session.mount('https://', adapter) + return session + + +def set_log_level(level: int): + logger = get_wideq_logger() + logger.setLevel(level) + + +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. @@ -47,7 +107,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, @@ -77,16 +137,10 @@ def __init__(self, code, 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.""" @@ -95,6 +149,16 @@ def __init__(self): pass +class FailedRequestError(APIError): + """A failed request typically indicates an unsupported control on a + device. + """ + + +class InvalidRequestError(APIError): + """The server rejected a request as invalid.""" + + class MonitorError(APIError): """Monitoring a device failed, possibly because the monitoring session failed and needs to be restarted. @@ -105,6 +169,14 @@ def __init__(self, device_id, code): self.code = code +API_ERRORS = { + "0102": NotLoggedInError, + "0106": NotConnectedError, + "0100": FailedRequestError, + 9000: InvalidRequestError, # Surprisingly, an integer (not a string). +} + + def lgedm_post(url, data=None, access_token=None, session_id=None): """Make an HTTP request in the format used by the API servers. @@ -116,7 +188,6 @@ def lgedm_post(url, data=None, access_token=None, session_id=None): 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, @@ -127,7 +198,16 @@ def lgedm_post(url, data=None, access_token=None, session_id=None): if session_id: headers['x-thinq-jsessionId'] = session_id - res = requests.post(url, json={DATA_ROOT: data}, headers=headers) + with retry_session() as session: + print(url) + print(data) + res = session.post(url, json={DATA_ROOT: data}, headers=headers) + + if "rtiControl" in url: + res2 = session.post('https://eic.lgthinq.com:46030/api/rti/delControlPermission', json={DATA_ROOT: {'deviceId': data.deviceId}}, headers=headers) + print(res2.status_code) + print(res2.text) + out = res.json()[DATA_ROOT] # Check for API errors. @@ -135,29 +215,14 @@ def lgedm_post(url, data=None, access_token=None, session_id=None): code = out['returnCd'] if code != '0000': message = out['returnMsg'] - if code == "0102": - raise NotLoggedInError() - elif code == "0106": - raise NotConnectedError() + if code in API_ERRORS: + raise API_ERRORS[code](code, message) 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. @@ -234,7 +299,8 @@ def refresh_auth(oauth_root, refresh_token): 'Accept': 'application/json', } - res = requests.post(token_url, data=data, headers=headers) + with retry_session() as session: + res = session.post(token_url, data=data, headers=headers) res_data = res.json() if res_data['status'] != 1: @@ -251,14 +317,35 @@ def __init__(self, auth_base, api_root, oauth_root, country, language): self.language = language @classmethod - def discover(cls, country, language): - gw = gateway_info(country, language) + def discover(cls, country, language) -> 'Gateway': + """Load information about the hosts to use for API interaction. + + `country` and `language` are codes, like "US" and "en-US," + respectively. + """ + gw = lgedm_post(GATEWAY_URL, + {'countryCode': country, 'langCode': 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) + def serialize(self) -> Dict[str, str]: + return { + 'auth_base': self.auth_base, + 'api_root': self.api_root, + 'oauth_root': self.oauth_root, + 'country': self.country, + 'language': self.language, + } + + @classmethod + def deserialize(cls, data: Dict[str, Any]) -> 'Gateway': + return cls(data['auth_base'], data['api_root'], data['oauth_root'], + data.get('country', DEFAULT_COUNTRY), + data.get('language', DEFAULT_LANGUAGE)) + class Auth(object): def __init__(self, gateway, access_token, refresh_token): @@ -274,7 +361,7 @@ def from_url(cls, gateway, url): access_token, refresh_token = parse_oauth_callback(url) return cls(gateway, access_token, refresh_token) - def start_session(self): + def start_session(self) -> Tuple['Session', List[Dict[str, Any]]]: """Start an API session for the logged-in user. Return the Session object and a list of the user's devices. """ @@ -292,9 +379,15 @@ def refresh(self): self.refresh_token) return Auth(self.gateway, new_access_token, self.refresh_token) + def serialize(self) -> Dict[str, str]: + return { + 'access_token': self.access_token, + 'refresh_token': self.refresh_token, + } + 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 +401,7 @@ def post(self, path, data=None): 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. @@ -406,4 +499,4 @@ def get_device_config(self, device_id, key, category='Config'): 'workId': gen_uuid(), 'data': '', }) - return res['returnData'] \ No newline at end of file + return res['returnData'] diff --git a/resources/wideq/wideq/dishwasher.py b/resources/wideq/wideq/dishwasher.py new file mode 100644 index 0000000..f912eff --- /dev/null +++ b/resources/wideq/wideq/dishwasher.py @@ -0,0 +1,160 @@ +import 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) -> Optional[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 "" + + @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) diff --git a/resources/wideq/wideq/dryer.py b/resources/wideq/wideq/dryer.py index aca02e7..0b25203 100644 --- a/resources/wideq/wideq/dryer.py +++ b/resources/wideq/wideq/dryer.py @@ -97,8 +97,9 @@ def poll(self) -> Optional['DryerStatus']: if not hasattr(self, 'mon'): return None - res = self.mon.poll_json() - if res: + data = self.mon.poll() + if data: + res = self.model.decode_monitor(data) return DryerStatus(self, res) else: return None diff --git a/resources/wideq/wideq/refrigerator.py b/resources/wideq/wideq/refrigerator.py new file mode 100644 index 0000000..7dff39f --- /dev/null +++ b/resources/wideq/wideq/refrigerator.py @@ -0,0 +1,140 @@ +import enum +from typing import Optional + +from .client import Device +from .util import lookup_enum + + +class IcePlus(enum.Enum): + OFF = "@CP_OFF_EN_W" + ON = "@CP_ON_EN_W" + ICE_PLUS = "@RE_TERM_ICE_PLUS_W" + ICE_PLUS_FREEZE = "@RE_MAIN_SPEED_FREEZE_TERM_W" + ICE_PLUS_OFF = "@CP_TERM_OFF_KO_W" + + +class FreshAirFilter(enum.Enum): + OFF = "@CP_TERM_OFF_KO_W" + AUTO = "@RE_STATE_FRESH_AIR_FILTER_MODE_AUTO_W" + POWER = "@RE_STATE_FRESH_AIR_FILTER_MODE_POWER_W" + REPLACE_FILTER = "@RE_STATE_REPLACE_FILTER_W" + SMARTCARE_ON = "@RE_STATE_SMART_SMART_CARE_ON" + SMARTCARE_OFF = "@RE_STATE_SMART_SMART_CARE_OFF" + SMARTCARE_WAIT = "@RE_STATE_SMART_SMART_CARE_WAIT" + EMPTY = "" + + +class SmartSavingMode(enum.Enum): + OFF = "@CP_TERM_USE_NOT_W" + NIGHT = "@RE_SMARTSAVING_MODE_NIGHT_W" + CUSTOM = "@RE_SMARTSAVING_MODE_CUSTOM_W" + SMART_GRID_OFF = "@CP_OFF_EN_W" + SMART_GRID_DEMAND_RESPONSE = "@RE_TERM_DEMAND_RESPONSE_FUNCTIONALITY_W" + SMART_GRID_CUSTOM = "@RE_TERM_DELAY_DEFROST_CAPABILITY_W" + EMPTY = "" + + +class RefrigeratorDevice(Device): + """A higher-level interface for a refrigerator.""" + + def set_temp_refrigerator_c(self, temp): + """Set the refrigerator temperature in Celsius. + """ + value = self.model.enum_value('TempRefrigerator', str(temp)) + self._set_control('RETM', value) + + def set_temp_freezer_c(self, temp): + """Set the freezer temperature in Celsius. + """ + value = self.model.enum_value('TempFreezer', str(temp)) + self._set_control('REFT', value) + + def poll(self) -> Optional['RefrigeratorStatus']: + """Poll the device's current state. + + Monitoring must be started first with `monitor_start`. + + :returns: Either a `RefrigeratorStatus` 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 RefrigeratorStatus(self, res) + else: + return None + + +class RefrigeratorStatus(object): + """Higher-level information about a refrigerator's current status. + + :param refrigerator: The RefrigeratorDevice instance. + :param data: JSON data from the API. + """ + + def __init__(self, refrigerator: RefrigeratorDevice, data: dict): + self.refrigerator = refrigerator + self.data = data + + @property + def temp_refrigerator_c(self): + temp = lookup_enum('TempRefrigerator', self.data, self.refrigerator) + return int(temp) + + @property + def temp_freezer_c(self): + temp = lookup_enum('TempFreezer', self.data, self.refrigerator) + return int(temp) + + @property + def ice_plus_status(self): + status = lookup_enum('IcePlus', self.data, self.refrigerator) + return IcePlus(status) + + @property + def fresh_air_filter_status(self): + status = lookup_enum('FreshAirFilter', self.data, self.refrigerator) + return FreshAirFilter(status) + + @property + def energy_saving_mode(self): + mode = lookup_enum('SmartSavingMode', self.data, self.refrigerator) + return SmartSavingMode(mode) + + @property + def door_opened(self): + state = lookup_enum('DoorOpenState', self.data, self.refrigerator) + return state == "OPEN" + + @property + def temp_unit(self): + return lookup_enum('TempUnit', self.data, self.refrigerator) + + @property + def energy_saving_enabled(self): + mode = lookup_enum( + 'SmartSavingModeStatus', self.data, self.refrigerator + ) + return mode == 'ON' + + @property + def locked(self): + status = lookup_enum('LockingStatus', self.data, self.refrigerator) + return status == "LOCK" + + @property + def active_saving_status(self): + return self.data['ActiveSavingStatus'] + + @property + def eco_enabled(self): + eco = lookup_enum('EcoFriendly', self.data, self.refrigerator) + return eco == "@CP_ON_EN_W" + + @property + def water_filter_used_month(self): + return self.data['WaterFilterUsedMonth'] diff --git a/resources/wideq/wideq/washer.py b/resources/wideq/wideq/washer.py index 26b40a0..974d3be 100644 --- a/resources/wideq/wideq/washer.py +++ b/resources/wideq/wideq/washer.py @@ -49,8 +49,9 @@ def poll(self) -> Optional['WasherStatus']: if not hasattr(self, 'mon'): return None - res = self.mon.poll_json() - if res: + data = self.mon.poll() + if data: + res = self.model.decode_monitor(data) return WasherStatus(self, res) else: return None