From fb3c755c2e87ee4fef693b231e05f493ed4e2f4c Mon Sep 17 00:00:00 2001 From: Michael Wei Date: Sat, 21 Nov 2020 19:46:57 +0000 Subject: [PATCH 1/5] introduce black formatter --- .flake8 | 5 + .pre-commit-config.yaml | 10 ++ README.md | 5 + pyproject.toml | 18 ++- tests/test_client.py | 181 +++++++++++++------------- tests/test_core.py | 57 ++++---- tests/test_dishwasher.py | 39 +++--- tests/test_dryer.py | 86 +++++++------ tests/test_washer.py | 76 +++++------ wideq/__init__.py | 2 +- wideq/ac.py | 223 +++++++++++++++++--------------- wideq/client.py | 188 ++++++++++++++------------- wideq/core.py | 271 +++++++++++++++++++++------------------ wideq/dishwasher.py | 86 +++++++------ wideq/dryer.py | 129 ++++++++++--------- wideq/refrigerator.py | 44 +++---- wideq/util.py | 7 +- wideq/washer.py | 75 +++++------ 18 files changed, 807 insertions(+), 695 deletions(-) create mode 100644 .flake8 create mode 100644 .pre-commit-config.yaml diff --git a/.flake8 b/.flake8 new file mode 100644 index 0000000..b9a569e --- /dev/null +++ b/.flake8 @@ -0,0 +1,5 @@ +[flake8] +ignore = E203, E266, E501, W503, F403, F401 +max-line-length = 79 +max-complexity = 18 +select = B,C,E,F,W,T4,B9 \ No newline at end of file diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..b398c11 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,10 @@ +repos: +- repo: https://github.com/ambv/black + rev: stable + hooks: + - id: black + language_version: python3.9 +- repo: https://github.com/pre-commit/pre-commit-hooks + rev: v1.2.3 + hooks: + - id: flake8 \ No newline at end of file diff --git a/README.md b/README.md index 2ec8a58..b82e5ac 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,11 @@ You can also specify one of several other commands: * `turn `: Turn an AC device on or off. Use "on" or "off" as the second argument. * `ac-config `: Print out some configuration information about an AC device. +Development +------- +To ensure consistent formatting across pull requests, install the precommit hooks to auto format your code using `pre-commit install`. + +The code will be auto-formatted by `black` to ensure consistent style. Credits ------- diff --git a/pyproject.toml b/pyproject.toml index 943cdf5..ff12792 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,5 +1,5 @@ [build-system] -requires = ["flit"] +requires = ["flit", "pre-commit"] build-backend = "flit.buildapi" [tool.flit.metadata] @@ -16,3 +16,19 @@ requires-python = ">=3.6" test = [ "responses" ] + +[tool.black] +line-length = 79 +include = '\.pyi?$' +exclude = ''' +/( + \.git + | \.hg + | \.mypy_cache + | \.tox + | \.venv + | _build + | build + | dist +)/ +''' \ No newline at end of file diff --git a/tests/test_client.py b/tests/test_client.py index a7f1f62..3582198 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,148 +1,151 @@ import unittest from wideq.client import ( - BitValue, EnumValue, ModelInfo, RangeValue, ReferenceValue, StringValue) + BitValue, + EnumValue, + ModelInfo, + RangeValue, + ReferenceValue, + StringValue, +) DATA = { - 'Value': { - 'AntiBacterial': { - 'default': '0', - 'label': '@WM_DRY27_BUTTON_ANTI_BACTERIAL_W', - 'option': { - '0': '@CP_OFF_EN_W', - '1': '@CP_ON_EN_W' - }, - 'type': 'Enum' + "Value": { + "AntiBacterial": { + "default": "0", + "label": "@WM_DRY27_BUTTON_ANTI_BACTERIAL_W", + "option": {"0": "@CP_OFF_EN_W", "1": "@CP_ON_EN_W"}, + "type": "Enum", }, - 'Course': { - 'option': ['Course'], - 'type': 'Reference', + "Course": { + "option": ["Course"], + "type": "Reference", }, - 'Initial_Time_H': { - 'default': 0, - 'option': {'max': 24, 'min': 0}, - 'type': 'Range' + "Initial_Time_H": { + "default": 0, + "option": {"max": 24, "min": 0}, + "type": "Range", }, - 'Option1': { - 'default': '0', - 'option': [ + "Option1": { + "default": "0", + "option": [ { - 'default': '0', - 'length': 1, - 'startbit': 0, - 'value': 'ChildLock' + "default": "0", + "length": 1, + "startbit": 0, + "value": "ChildLock", }, { - 'default': '0', - 'length': 1, - 'startbit': 1, - 'value': 'ReduceStatic' + "default": "0", + "length": 1, + "startbit": 1, + "value": "ReduceStatic", }, { - 'default': '0', - 'length': 1, - 'startbit': 2, - 'value': 'EasyIron' + "default": "0", + "length": 1, + "startbit": 2, + "value": "EasyIron", }, { - 'default': '0', - 'length': 1, - 'startbit': 3, - 'value': 'DampDrySingal' + "default": "0", + "length": 1, + "startbit": 3, + "value": "DampDrySingal", }, { - 'default': '0', - 'length': 1, - 'startbit': 4, - 'value': 'WrinkleCare' + "default": "0", + "length": 1, + "startbit": 4, + "value": "WrinkleCare", }, { - 'default': '0', - 'length': 1, - 'startbit': 7, - 'value': 'AntiBacterial' - } + "default": "0", + "length": 1, + "startbit": 7, + "value": "AntiBacterial", + }, ], - 'type': 'Bit' - }, - 'TimeBsOn': { - '_comment': - '오전 12시 30분은 0030, 오후12시30분은 1230 ,오후 4시30분은 1630 off는 0 ', - 'type': 'String' + "type": "Bit", }, - 'Unexpected': {'type': 'Unexpected'}, - 'Unexpected2': { - 'type': 'Unexpected', - 'option': 'some option' + "TimeBsOn": { + "_comment": "오전 12시 30분은 0030, 오후12시30분은 1230 ,오후 4시30분은 1630 off는 0 ", + "type": "String", }, + "Unexpected": {"type": "Unexpected"}, + "Unexpected2": {"type": "Unexpected", "option": "some option"}, }, - 'Course': { + "Course": { "3": { - "_comment": "Normal", - "courseType": "Course", - "id": 3, - "name": "@WM_DRY27_COURSE_NORMAL_W", - "script": "", - "controlEnable": True, - "freshcareEnable": True, - "imgIndex": 61, + "_comment": "Normal", + "courseType": "Course", + "id": 3, + "name": "@WM_DRY27_COURSE_NORMAL_W", + "script": "", + "controlEnable": True, + "freshcareEnable": True, + "imgIndex": 61, }, }, } class ModelInfoTest(unittest.TestCase): - def setUp(self): super().setUp() self.model_info = ModelInfo(DATA) def test_value_enum(self): - actual = self.model_info.value('AntiBacterial') - expected = EnumValue({'0': '@CP_OFF_EN_W', '1': '@CP_ON_EN_W'}) + actual = self.model_info.value("AntiBacterial") + expected = EnumValue({"0": "@CP_OFF_EN_W", "1": "@CP_ON_EN_W"}) self.assertEqual(expected, actual) def test_value_range(self): - actual = self.model_info.value('Initial_Time_H') + actual = self.model_info.value("Initial_Time_H") expected = RangeValue(min=0, max=24, step=1) self.assertEqual(expected, actual) def test_value_bit(self): - actual = self.model_info.value('Option1') - expected = BitValue({ - 0: 'ChildLock', - 1: 'ReduceStatic', - 2: 'EasyIron', - 3: 'DampDrySingal', - 4: 'WrinkleCare', - 7: 'AntiBacterial', - }) + actual = self.model_info.value("Option1") + expected = BitValue( + { + 0: "ChildLock", + 1: "ReduceStatic", + 2: "EasyIron", + 3: "DampDrySingal", + 4: "WrinkleCare", + 7: "AntiBacterial", + } + ) self.assertEqual(expected, actual) def test_value_reference(self): - actual = self.model_info.value('Course') - expected = ReferenceValue(DATA['Course']) + actual = self.model_info.value("Course") + expected = ReferenceValue(DATA["Course"]) self.assertEqual(expected, actual) def test_string(self): - actual = self.model_info.value('TimeBsOn') + actual = self.model_info.value("TimeBsOn") expected = StringValue( - "오전 12시 30분은 0030, 오후12시30분은 1230 ,오후 4시30분은 1630 off는 0 ") + "오전 12시 30분은 0030, 오후12시30분은 1230 ,오후 4시30분은 1630 off는 0 " + ) self.assertEqual(expected, actual) def test_value_unsupported(self): data = "{'type': 'Unexpected'}" with self.assertRaisesRegex( - ValueError, - f"unsupported value name: 'Unexpected' type: 'Unexpected' " - f"data: '{data}'"): - self.model_info.value('Unexpected') + ValueError, + f"unsupported value name: 'Unexpected' type: 'Unexpected' " + f"data: '{data}'", + ): + self.model_info.value("Unexpected") def test_value_unsupported_but_data_available(self): data = "{'type': 'Unexpected', 'option': 'some option'}" with self.assertRaisesRegex( - ValueError, - f"unsupported value name: 'Unexpected2'" - f" type: 'Unexpected' data: '{data}"): - self.model_info.value('Unexpected2') + ValueError, + f"unsupported value name: 'Unexpected2'" + f" type: 'Unexpected' data: '{data}", + ): + self.model_info.value("Unexpected2") diff --git a/tests/test_core.py b/tests/test_core.py index e647b1a..4b90f4b 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -9,47 +9,52 @@ class SimpleTest(unittest.TestCase): def test_gateway_en_US(self): responses.add( responses.POST, - 'https://kic.lgthinq.com:46030/api/common/gatewayUriList', + "https://kic.lgthinq.com:46030/api/common/gatewayUriList", json={ - 'lgedmRoot': { + "lgedmRoot": { "thinqUri": "https://aic.lgthinq.com:46030/api", "empUri": "https://us.m.lgaccount.com", "oauthUri": "https://us.lgeapi.com", "countryCode": "US", "langCode": "en-US", } - } + }, ) - gatewayInstance = wideq.core.Gateway.discover('US', 'en-US') + gatewayInstance = wideq.core.Gateway.discover("US", "en-US") self.assertEqual(len(responses.calls), 1) - self.assertEqual(gatewayInstance.country, 'US') - self.assertEqual(gatewayInstance.language, 'en-US') - self.assertEqual(gatewayInstance.auth_base, - 'https://us.m.lgaccount.com') - self.assertEqual(gatewayInstance.api_root, - 'https://aic.lgthinq.com:46030/api') - self.assertEqual(gatewayInstance.oauth_root, 'https://us.lgeapi.com') + self.assertEqual(gatewayInstance.country, "US") + self.assertEqual(gatewayInstance.language, "en-US") + self.assertEqual( + gatewayInstance.auth_base, "https://us.m.lgaccount.com" + ) + self.assertEqual( + gatewayInstance.api_root, "https://aic.lgthinq.com:46030/api" + ) + self.assertEqual(gatewayInstance.oauth_root, "https://us.lgeapi.com") @responses.activate def test_gateway_en_NO(self): responses.add( responses.POST, - 'https://kic.lgthinq.com:46030/api/common/gatewayUriList', + "https://kic.lgthinq.com:46030/api/common/gatewayUriList", json={ - 'lgedmRoot': { - "countryCode": "NO", "langCode": "en-NO", - "thinqUri": "https://eic.lgthinq.com:46030/api", - "empUri": "https://no.m.lgaccount.com", - "oauthUri": "https://no.lgeapi.com", + "lgedmRoot": { + "countryCode": "NO", + "langCode": "en-NO", + "thinqUri": "https://eic.lgthinq.com:46030/api", + "empUri": "https://no.m.lgaccount.com", + "oauthUri": "https://no.lgeapi.com", } - } + }, ) - gatewayInstance = wideq.core.Gateway.discover('NO', 'en-NO') + gatewayInstance = wideq.core.Gateway.discover("NO", "en-NO") self.assertEqual(len(responses.calls), 1) - self.assertEqual(gatewayInstance.country, 'NO') - self.assertEqual(gatewayInstance.language, 'en-NO') - self.assertEqual(gatewayInstance.auth_base, - 'https://no.m.lgaccount.com') - self.assertEqual(gatewayInstance.api_root, - 'https://eic.lgthinq.com:46030/api') - self.assertEqual(gatewayInstance.oauth_root, 'https://no.lgeapi.com') + self.assertEqual(gatewayInstance.country, "NO") + self.assertEqual(gatewayInstance.language, "en-NO") + self.assertEqual( + gatewayInstance.auth_base, "https://no.m.lgaccount.com" + ) + self.assertEqual( + gatewayInstance.api_root, "https://eic.lgthinq.com:46030/api" + ) + self.assertEqual(gatewayInstance.oauth_root, "https://no.lgeapi.com") diff --git a/tests/test_dishwasher.py b/tests/test_dishwasher.py index a182f61..cc0bb02 100644 --- a/tests/test_dishwasher.py +++ b/tests/test_dishwasher.py @@ -2,8 +2,11 @@ import unittest from wideq.client import Client, DeviceInfo -from wideq.dishwasher import DishWasherDevice, DishWasherState, \ - DishWasherStatus +from wideq.dishwasher import ( + DishWasherDevice, + DishWasherState, + DishWasherStatus, +) POLL_DATA = { "16~19": "0", @@ -30,22 +33,24 @@ class DishWasherStatusTest(unittest.TestCase): - def setUp(self): super().setUp() - with open('./tests/fixtures/client.json') as fp: + with open("./tests/fixtures/client.json") as fp: state = json.load(fp) self.client = Client.load(state) - self.device_info = DeviceInfo({ - 'alias': 'DISHWASHER', - 'deviceId': '33330ba80-107d-11e9-96c8-0051ede8ad3c', - 'deviceType': 204, - 'modelJsonUrl': ( - 'https://aic.lgthinq.com:46030/api/webContents/modelJSON?' - 'modelName=D3210&countryCode=WW&contentsId=' - 'JS0719082250749334&authKey=thinq'), - 'modelNm': 'D3210', - }) + self.device_info = DeviceInfo( + { + "alias": "DISHWASHER", + "deviceId": "33330ba80-107d-11e9-96c8-0051ede8ad3c", + "deviceType": 204, + "modelJsonUrl": ( + "https://aic.lgthinq.com:46030/api/webContents/modelJSON?" + "modelName=D3210&countryCode=WW&contentsId=" + "JS0719082250749334&authKey=thinq" + ), + "modelNm": "D3210", + } + ) self.dishwasher = DishWasherDevice(self.client, self.device_info) def test_properties(self): @@ -54,6 +59,6 @@ def test_properties(self): self.assertTrue(status.is_on) self.assertEqual(119, status.remaining_time) self.assertEqual(194, status.initial_time) - self.assertEqual('Heavy', status.course) - self.assertEqual('Casseroles', status.smart_course) - self.assertEqual('No Error', status.error) + self.assertEqual("Heavy", status.course) + self.assertEqual("Casseroles", status.smart_course) + self.assertEqual("No Error", status.error) diff --git a/tests/test_dryer.py b/tests/test_dryer.py index 6564c1f..d0a4747 100644 --- a/tests/test_dryer.py +++ b/tests/test_dryer.py @@ -4,47 +4,55 @@ from wideq.client import Client, DeviceInfo from wideq.dryer import ( - DryerDevice, DryLevel, DryerState, DryerStatus, TempControl, TimeDry) + DryerDevice, + DryLevel, + DryerState, + DryerStatus, + TempControl, + TimeDry, +) POLL_DATA = { - 'Course': '2', - 'CurrentDownloadCourse': '100', - 'DryLevel': '3', - 'Error': '0', - 'Initial_Time_H': '1', - 'Initial_Time_M': '11', - 'LoadItem': '0', - 'MoreLessTime': '0', - 'Option1': '0', - 'Option2': '168', - 'PreState': '1', - 'Remain_Time_H': '0', - 'Remain_Time_M': '54', - 'SmartCourse': '0', - 'State': '50', - 'TempControl': '4', - 'TimeDry': '0', + "Course": "2", + "CurrentDownloadCourse": "100", + "DryLevel": "3", + "Error": "0", + "Initial_Time_H": "1", + "Initial_Time_M": "11", + "LoadItem": "0", + "MoreLessTime": "0", + "Option1": "0", + "Option2": "168", + "PreState": "1", + "Remain_Time_H": "0", + "Remain_Time_M": "54", + "SmartCourse": "0", + "State": "50", + "TempControl": "4", + "TimeDry": "0", } class DryerStatusTest(unittest.TestCase): - def setUp(self): super().setUp() - with open('./tests/fixtures/client.json') as fp: + with open("./tests/fixtures/client.json") as fp: state = json.load(fp) self.client = Client.load(state) - self.device_info = DeviceInfo({ - 'alias': 'DRYER', - 'deviceId': '33330ba80-107d-11e9-96c8-0051ede85d3f', - 'deviceType': 202, - 'modelJsonUrl': ( - 'https://aic.lgthinq.com:46030/api/webContents/modelJSON?' - 'modelName=RV13B6ES_D_US_WIFI&countryCode=WW&contentsId=' - 'JS11260025236447318&authKey=thinq'), - 'modelNm': 'RV13B6ES_D_US_WIFI', - }) + self.device_info = DeviceInfo( + { + "alias": "DRYER", + "deviceId": "33330ba80-107d-11e9-96c8-0051ede85d3f", + "deviceType": 202, + "modelJsonUrl": ( + "https://aic.lgthinq.com:46030/api/webContents/modelJSON?" + "modelName=RV13B6ES_D_US_WIFI&countryCode=WW&contentsId=" + "JS11260025236447318&authKey=thinq" + ), + "modelNm": "RV13B6ES_D_US_WIFI", + } + ) self.dryer = DryerDevice(self.client, self.device_info) def test_properties(self): @@ -57,22 +65,26 @@ def test_properties(self): self.assertTrue(status.is_on) self.assertEqual(54, status.remaining_time) self.assertEqual(71, status.initial_time) - self.assertEqual('Towels', status.course) - self.assertEqual('Off', status.smart_course) - self.assertEqual('No Error', status.error) + self.assertEqual("Towels", status.course) + self.assertEqual("Off", status.smart_course) + self.assertEqual("No Error", status.error) self.assertEqual(TempControl.MID_HIGH, status.temperature_control) self.assertEqual(TimeDry.OFF, status.time_dry) - @mock.patch('wideq.client.LOGGER') + @mock.patch("wideq.client.LOGGER") def test_properties_unknown_enum_value(self, mock_logging): """ This should not raise an error for an invalid enum value and instead use the `UNKNOWN` enum value. """ - data = dict(POLL_DATA, State='5000') + data = dict(POLL_DATA, State="5000") status = DryerStatus(self.dryer, data) self.assertEqual(DryerState.UNKNOWN, status.state) expected_call = mock.call( - 'Value `%s` for key `%s` not in options: %s. Values from API: %s', - '5000', 'State', mock.ANY, mock.ANY) + "Value `%s` for key `%s` not in options: %s. Values from API: %s", + "5000", + "State", + mock.ANY, + mock.ANY, + ) self.assertEqual(expected_call, mock_logging.warning.call_args) diff --git a/tests/test_washer.py b/tests/test_washer.py index ca9b083..dd5659a 100644 --- a/tests/test_washer.py +++ b/tests/test_washer.py @@ -6,48 +6,50 @@ POLL_DATA = { - 'APCourse': '10', - 'DryLevel': '0', - 'Error': '0', - 'Initial_Time_H': '0', - 'Initial_Time_M': '58', - 'LoadLevel': '4', - 'OPCourse': '0', - 'Option1': '0', - 'Option2': '0', - 'Option3': '2', - 'PreState': '23', - 'Remain_Time_H': '0', - 'Remain_Time_M': '13', - 'Reserve_Time_H': '0', - 'Reserve_Time_M': '0', - 'RinseOption': '1', - 'SmartCourse': '51', - 'Soil': '0', - 'SpinSpeed': '5', - 'State': '30', - 'TCLCount': '15', - 'WaterTemp': '4', + "APCourse": "10", + "DryLevel": "0", + "Error": "0", + "Initial_Time_H": "0", + "Initial_Time_M": "58", + "LoadLevel": "4", + "OPCourse": "0", + "Option1": "0", + "Option2": "0", + "Option3": "2", + "PreState": "23", + "Remain_Time_H": "0", + "Remain_Time_M": "13", + "Reserve_Time_H": "0", + "Reserve_Time_M": "0", + "RinseOption": "1", + "SmartCourse": "51", + "Soil": "0", + "SpinSpeed": "5", + "State": "30", + "TCLCount": "15", + "WaterTemp": "4", } class WasherStatusTest(unittest.TestCase): - def setUp(self): super().setUp() - with open('./tests/fixtures/client.json') as fp: + with open("./tests/fixtures/client.json") as fp: state = json.load(fp) self.client = Client.load(state) - self.device_info = DeviceInfo({ - 'alias': 'WASHER', - 'deviceId': '33330ba80-107d-11e9-96c8-0051ede85d3f', - 'deviceType': 201, - 'modelJsonUrl': ( - 'https://aic.lgthinq.com:46030/api/webContents/modelJSON?' - 'modelName=F3L2CYV5W_WIFI&countryCode=WW&contentsId=' - 'JS1217232703654216&authKey=thinq'), - 'modelNm': 'F3L2CYV5W_WIFI', - }) + self.device_info = DeviceInfo( + { + "alias": "WASHER", + "deviceId": "33330ba80-107d-11e9-96c8-0051ede85d3f", + "deviceType": 201, + "modelJsonUrl": ( + "https://aic.lgthinq.com:46030/api/webContents/modelJSON?" + "modelName=F3L2CYV5W_WIFI&countryCode=WW&contentsId=" + "JS1217232703654216&authKey=thinq" + ), + "modelNm": "F3L2CYV5W_WIFI", + } + ) self.washer = WasherDevice(self.client, self.device_info) def test_properties(self): @@ -57,6 +59,6 @@ def test_properties(self): self.assertTrue(status.is_on) self.assertEqual(13, status.remaining_time) self.assertEqual(58, status.initial_time) - self.assertEqual('Towels', status.course) - self.assertEqual('SmallLoad', status.smart_course) - self.assertEqual('No Error', status.error) + self.assertEqual("Towels", status.course) + self.assertEqual("SmallLoad", status.smart_course) + self.assertEqual("No Error", status.error) diff --git a/wideq/__init__.py b/wideq/__init__.py index 4f4b048..9b4663e 100644 --- a/wideq/__init__.py +++ b/wideq/__init__.py @@ -8,4 +8,4 @@ from .refrigerator import * # noqa from .washer import * # noqa -__version__ = '1.5.0' +__version__ = "1.5.0" diff --git a/wideq/ac.py b/wideq/ac.py index 54bf93e..f154015 100644 --- a/wideq/ac.py +++ b/wideq/ac.py @@ -33,6 +33,7 @@ class ACVSwingMode(enum.Enum): All is 100. """ + OFF = "@OFF" ONE = "@1" TWO = "@2" @@ -54,6 +55,7 @@ class ACHSwingMode(enum.Enum): All is 100. """ + OFF = "@OFF" ONE = "@1" TWO = "@2" @@ -83,46 +85,70 @@ class ACMode(enum.Enum): 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' - 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' + 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" + 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): @@ -150,7 +176,7 @@ def f2c(self): precise control requires using the custom LUT. """ - mapping = self.model.value('TempFahToCel').options + mapping = self.model.value("TempFahToCel").options return {int(f): c for f, c in mapping.items()} @property @@ -162,7 +188,7 @@ def c2f(self): are not in the other. """ - mapping = self.model.value('TempCelToFah').options + mapping = self.model.value("TempCelToFah").options out = {} for c, f in mapping.items(): try: @@ -174,10 +200,9 @@ def c2f(self): @property def supported_operations(self): - """Get a list of the ACOp Operations the device supports. - """ + """Get a list of the ACOp Operations the device supports.""" - mapping = self.model.value('Operation').options + mapping = self.model.value("Operation").options return [ACOp(o) for i, o in mapping.items()] @property @@ -208,17 +233,16 @@ def supported_on_operation(self): # 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)}'") + f" too many reported operations: '{str(operations)}'" + ) def set_celsius(self, c): - """Set the device's target temperature in Celsius degrees. - """ + """Set the device's target temperature in Celsius degrees.""" - self._set_control('TempCfg', c) + self._set_control("TempCfg", c) def set_fahrenheit(self, f): - """Set the device's target temperature in Fahrenheit degrees. - """ + """Set the device's target temperature in Fahrenheit degrees.""" self.set_celsius(self.f2c[f]) @@ -235,13 +259,14 @@ def set_zones(self, zones): # 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) + 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' + zone_cmd = "/".join( + "{}_{}".format(zone["No"], zone["State"]) + for zone in zones + if zone["Cfg"] == "1" ) - self._set_control('DuctZone', zone_cmd) + self._set_control("DuctZone", zone_cmd) def get_zones(self): """Get the status of the zones, including whether a zone is @@ -251,85 +276,78 @@ def get_zones(self): `set_zones`. """ - return self._get_config('DuctZone') + return self._get_config("DuctZone") def set_jet_mode(self, jet_opt): - """Set jet mode to a value from the `ACJetMode` enum. - """ + """Set jet mode to a value from the `ACJetMode` enum.""" - jet_opt_value = self.model.enum_value('Jet', jet_opt.value) - self._set_control('Jet', jet_opt_value) + jet_opt_value = self.model.enum_value("Jet", jet_opt.value) + self._set_control("Jet", jet_opt_value) def set_fan_speed(self, speed): - """Set the fan speed to a value from the `ACFanSpeed` enum. - """ + """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) + 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. - """ + """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) + 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. - """ + """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) + 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. - """ + """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) + 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). - """ + """Turn on or off the device (according to a boolean).""" 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) + 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') + return self._get_config("Filter") def get_mfilter_state(self): - """Get information about the "MFilter" (not sure what this is). - """ + """Get information about the "MFilter" (not sure what this is).""" - return self._get_config('MFilter') + return self._get_config("MFilter") def get_energy_target(self): """Get the configured energy target data.""" - return self._get_config('EnergyDesiredValue') + 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'] + 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'] + value = self._get_config("InOutInstantPower") + return value["InOutInstantPower"] def get_light(self): """Get a Boolean indicating whether the display light is on.""" try: - value = self._get_control('DisplayControl') - return value == '0' # Seems backwards, but isn't. + 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. @@ -339,7 +357,7 @@ def get_volume(self): """Get the speaker volume level.""" try: - value = self._get_control('SpkVolume') + value = self._get_control("SpkVolume") return int(value) except FailedRequestError: return 0 # Device does not support volume control. @@ -353,7 +371,7 @@ def poll(self): """ # Abort if monitoring has not started yet. - if not hasattr(self, 'mon'): + if not hasattr(self, "mon"): return None res = self.mon.poll_json() @@ -364,8 +382,7 @@ def poll(self): class ACStatus(object): - """Higher-level information about an AC device's current status. - """ + """Higher-level information about an AC device's current status.""" def __init__(self, ac, data): self.ac = ac @@ -388,7 +405,7 @@ def _str_to_num(s): @property def temp_cur_c(self): - return self._str_to_num(self.data['TempCur']) + return self._str_to_num(self.data["TempCur"]) @property def temp_cur_f(self): @@ -396,7 +413,7 @@ def temp_cur_f(self): @property def temp_cfg_c(self): - return self._str_to_num(self.data['TempCfg']) + return self._str_to_num(self.data["TempCfg"]) @property def temp_cfg_f(self): @@ -404,23 +421,23 @@ def temp_cfg_f(self): @property def mode(self): - return ACMode(lookup_enum('OpMode', self.data, self.ac)) + return ACMode(lookup_enum("OpMode", self.data, self.ac)) @property def fan_speed(self): - return ACFanSpeed(lookup_enum('WindStrength', self.data, self.ac)) + return ACFanSpeed(lookup_enum("WindStrength", self.data, self.ac)) @property def horz_swing(self): - return ACHSwingMode(lookup_enum('WDirHStep', self.data, self.ac)) + return ACHSwingMode(lookup_enum("WDirHStep", self.data, self.ac)) @property def vert_swing(self): - return ACVSwingMode(lookup_enum('WDirVStep', self.data, self.ac)) + return ACVSwingMode(lookup_enum("WDirVStep", self.data, self.ac)) @property def is_on(self): - op = ACOp(lookup_enum('Operation', self.data, self.ac)) + op = ACOp(lookup_enum("Operation", self.data, self.ac)) return op != ACOp.OFF def __str__(self): diff --git a/wideq/client.py b/wideq/client.py index 04988ed..19e0c03 100644 --- a/wideq/client.py +++ b/wideq/client.py @@ -14,7 +14,7 @@ #: Represents an unknown enum value. -_UNKNOWN = 'Unknown' +_UNKNOWN = "Unknown" LOGGER = logging.getLogger("wideq.client") @@ -53,7 +53,7 @@ def poll(self) -> Optional[bytes]: def decode_json(data: bytes) -> Dict[str, Any]: """Decode a bytestring that encodes JSON status data.""" - return json.loads(data.decode('utf8')) + return json.loads(data.decode("utf8")) def poll_json(self) -> Optional[Dict[str, Any]]: """For devices where status is reported via JSON data, get the @@ -63,7 +63,7 @@ def poll_json(self) -> Optional[Dict[str, Any]]: data = self.poll() return self.decode_json(data) if data else None - def __enter__(self) -> 'Monitor': + def __enter__(self) -> "Monitor": self.start() return self @@ -76,12 +76,14 @@ class Client(object): and allows serialization of state. """ - 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: + 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: Optional[core.Gateway] = gateway self._auth: Optional[core.Auth] = auth @@ -120,15 +122,14 @@ def session(self) -> core.Session: return self._session @property - def devices(self) -> Generator['DeviceInfo', None, None]: - """DeviceInfo objects describing the user's devices. - """ + def devices(self) -> Generator["DeviceInfo", None, None]: + """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) -> Optional['DeviceInfo']: + def get_device(self, device_id) -> Optional["DeviceInfo"]: """Look up a DeviceInfo object by device ID. Return None if the device does not exist. @@ -153,37 +154,38 @@ def get_device_obj(self, device_id): classes = util.device_classes() if device_info.type in classes: return classes[device_info.type](self, device_info) - LOGGER.debug('No specific subclass for deviceType %s, using default', - device_info.type) + LOGGER.debug( + "No specific subclass for deviceType %s, using default", + device_info.type, + ) return Device(self, device_info) @classmethod - def load(cls, state: Dict[str, Any]) -> 'Client': - """Load a client from serialized state. - """ + def load(cls, state: Dict[str, Any]) -> "Client": + """Load a client from serialized state.""" client = cls() - if 'gateway' in state: - client._gateway = core.Gateway.deserialize(state['gateway']) + if "gateway" in state: + client._gateway = core.Gateway.deserialize(state["gateway"]) - if 'auth' in state: - data = state['auth'] + if "auth" in state: + data = state["auth"] client._auth = core.Auth( - client.gateway, data['access_token'], data['refresh_token'] + client.gateway, data["access_token"], data["refresh_token"] ) - if 'session' in state: - client._session = core.Session(client.auth, state['session']) + if "session" in state: + client._session = core.Session(client.auth, state["session"]) - if 'model_info' in state: - client._model_info = state['model_info'] + if "model_info" in state: + client._model_info = state["model_info"] - if 'country' in state: - client._country = state['country'] + if "country" in state: + client._country = state["country"] - if 'language' in state: - client._language = state['language'] + if "language" in state: + client._language = state["language"] return client @@ -191,20 +193,20 @@ def dump(self) -> Dict[str, Any]: """Serialize the client state.""" out: Dict[str, Any] = { - 'model_info': self._model_info, + "model_info": self._model_info, } if self._gateway: - out['gateway'] = self._gateway.serialize() + out["gateway"] = self._gateway.serialize() if self._auth: - out['auth'] = self._auth.serialize() + out["auth"] = self._auth.serialize() if self._session: - out['session'] = self._session.session_id + out["session"] = self._session.session_id - out['country'] = self._country - out['language'] = self._language + out["country"] = self._country + out["language"] = self._language return out @@ -213,8 +215,9 @@ def refresh(self) -> None: self._session, self._devices = self.auth.start_session() @classmethod - def from_token(cls, refresh_token, - country=None, language=None) -> 'Client': + 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 @@ -230,7 +233,7 @@ def from_token(cls, refresh_token, client.refresh() return client - def model_info(self, device: 'DeviceInfo') -> 'ModelInfo': + def model_info(self, device: "DeviceInfo") -> "ModelInfo": """For a DeviceInfo object, get a ModelInfo object describing the model's capabilities. """ @@ -281,44 +284,42 @@ def __init__(self, data: Dict[str, Any]) -> None: @property def model_id(self) -> str: - return self.data['modelNm'] + return self.data["modelNm"] @property def id(self) -> str: - return self.data['deviceId'] + return self.data["deviceId"] @property def model_info_url(self) -> str: - return self.data['modelJsonUrl'] + return self.data["modelJsonUrl"] @property def name(self) -> str: - return str(self.data['alias']) + return str(self.data["alias"]) @property def type(self) -> DeviceType: """The kind of device, as a `DeviceType` value.""" - return DeviceType(self.data['deviceType']) + return DeviceType(self.data["deviceType"]) def load_model_info(self): - """Load JSON data describing the model's capabilities. - """ + """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']) +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']) -StringValue = namedtuple('StringValue', ['comment']) +ReferenceValue = namedtuple("ReferenceValue", ["reference"]) +StringValue = namedtuple("StringValue", ["comment"]) class ModelInfo(object): - """A description of a device model's capabilities. - """ + """A description of a device model's capabilities.""" def __init__(self, data): self.data = data @@ -331,47 +332,51 @@ def value(self, name: str): `ReferenceValue`, `StringValue`). :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': + 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) + 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']} + 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] + elif d["type"].lower() == "reference": + ref = d["option"][0] return ReferenceValue(self.data[ref]) - elif d['type'].lower() == 'string': - return StringValue(d.get('_comment', '')) + 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)}'") + f" type: '{str(d['type'])}' data: '{str(d)}'" + ) def default(self, name): - """Get the default value, if it exists, for a given value. - """ - return self.data['Value'][name]['default'] + """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. - """ + """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. - """ + """Look up the friendly enum name for an encoded value.""" options = self.value(key).options if value not in options: LOGGER.warning( - 'Value `%s` for key `%s` not in options: %s. Values from API: ' - '%s', value, key, options, self.data['Value'][key]['option']) + "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] @@ -386,31 +391,30 @@ def reference_name(self, key: str, value: Any) -> Optional[str]: value = str(value) reference = self.value(key).reference if value in reference: - return reference[value]['_comment'] + 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)' + """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. - """ + """Decode binary encoded status data.""" decoded = {} - for item in self.data['Monitoring']['protocol']: - key = item['value'] + for item in self.data["Monitoring"]["protocol"]: + key = item["value"] value = 0 - for v in data[item['startByte']:item['startByte'] + - item['length']]: + 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')) + return json.loads(data.decode("utf8")) def decode_monitor(self, data): """Decode status data.""" @@ -452,15 +456,15 @@ def _get_config(self, key): self.device.id, key, ) - data = base64.b64decode(data).decode('utf8') + data = base64.b64decode(data).decode("utf8") try: return json.loads(data) except json.decoder.JSONDecodeError: # Sometimes, the service returns JSON wrapped in an extra # pair of curly braces. Try removing them and re-parsing. - LOGGER.debug('attempting to fix JSON format') + LOGGER.debug("attempting to fix JSON format") try: - return json.loads(re.sub(r'^\{(.*?)\}$', r'\1', data)) + return json.loads(re.sub(r"^\{(.*?)\}$", r"\1", data)) except json.decoder.JSONDecodeError: raise core.MalformedResponseError(data) @@ -469,11 +473,11 @@ def _get_control(self, key): data = self.client.session.get_device_config( self.device.id, key, - 'Control', + "Control", ) # The response comes in a funky key/value format: "(key:value)". - _, value = data[1:-1].split(':') + _, value = data[1:-1].split(":") return value def monitor_start(self): diff --git a/wideq/core.py b/wideq/core.py index abde401..e169f11 100644 --- a/wideq/core.py +++ b/wideq/core.py @@ -12,17 +12,17 @@ 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' -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' -DEFAULT_COUNTRY = 'US' -DEFAULT_LANGUAGE = 'en-US' +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" +DEFAULT_COUNTRY = "US" +DEFAULT_LANGUAGE = "en-US" RETRY_COUNT = 5 # Anecdotally this seems sufficient. RETRY_FACTOR = 0.5 @@ -38,6 +38,7 @@ def get_wideq_logger() -> logging.Logger: try: import colorlog # type: ignore + colorfmt = f"%(log_color)s{fmt}%(reset)s" handler = colorlog.StreamHandler() handler.setFormatter( @@ -66,8 +67,7 @@ def get_wideq_logger() -> logging.Logger: def retry_session(): - """Get a Requests session that retries HTTP and HTTPS requests. - """ + """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() @@ -79,8 +79,8 @@ def retry_session(): status_forcelist=RETRY_STATUSES, ) adapter = HTTPAdapter(max_retries=retry) - session.mount('http://', adapter) - session.mount('https://', adapter) + session.mount("http://", adapter) + session.mount("https://", adapter) return session @@ -101,8 +101,8 @@ def oauth2_signature(message: str, secret: str) -> bytes: their UTF-8 equivalents. """ - secret_bytes = secret.encode('utf8') - hashed = hmac.new(secret_bytes, message.encode('utf8'), hashlib.sha1) + secret_bytes = secret.encode("utf8") + hashed = hmac.new(secret_bytes, message.encode("utf8"), hashlib.sha1) digest = hashed.digest() return base64.b64encode(digest) @@ -197,24 +197,24 @@ def lgedm_post(url, data=None, access_token=None, session_id=None): 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', + "x-thinq-application-key": APP_KEY, + "x-thinq-security-key": SECURITY_KEY, + "Accept": "application/json", } if access_token: - headers['x-thinq-token'] = access_token + headers["x-thinq-token"] = access_token if session_id: - headers['x-thinq-jsessionId'] = session_id + headers["x-thinq-jsessionId"] = session_id with retry_session() as session: res = session.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 "returnCd" in out: + code = out["returnCd"] + if code != "0000": + message = out["returnMsg"] if code in API_ERRORS: raise API_ERRORS[code](code, message) else: @@ -228,17 +228,19 @@ def oauth_url(auth_base, country, language): 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) + 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): @@ -248,7 +250,7 @@ def parse_oauth_callback(url): """ params = parse_qs(urlparse(url).query) - return params['access_token'][0], params['refresh_token'][0] + return params["access_token"][0], params["refresh_token"][0] def login(api_root, access_token, country, language): @@ -256,12 +258,12 @@ def login(api_root, access_token, country, language): return information about the session. """ - url = urljoin(api_root + '/', 'member/login') + url = urljoin(api_root + "/", "member/login") data = { - 'countryCode': country, - 'langCode': language, - 'loginType': 'EMP', - 'token': access_token, + "countryCode": country, + "langCode": language, + "loginType": "EMP", + "token": access_token, } return lgedm_post(url, data) @@ -272,10 +274,10 @@ def refresh_auth(oauth_root, refresh_token): May raise a `TokenError`. """ - token_url = urljoin(oauth_root, '/oauth2/token') + token_url = urljoin(oauth_root, "/oauth2/token") data = { - 'grant_type': 'refresh_token', - 'refresh_token': refresh_token, + "grant_type": "refresh_token", + "refresh_token": refresh_token, } # The timestamp for labeling OAuth requests can be obtained @@ -287,25 +289,27 @@ def refresh_auth(oauth_root, refresh_token): # 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) + 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', + "lgemp-x-app-key": OAUTH_CLIENT_KEY, + "lgemp-x-signature": sig, + "lgemp-x-date": timestamp, + "Accept": "application/json", } with retry_session() as session: res = session.post(token_url, data=data, headers=headers) res_data = res.json() - if res_data['status'] != 1: + if res_data["status"] != 1: raise TokenError() - return res_data['access_token'] + return res_data["access_token"] class Gateway(object): @@ -317,34 +321,40 @@ def __init__(self, auth_base, api_root, oauth_root, country, language): self.language = language @classmethod - def discover(cls, country, language) -> 'Gateway': + 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) + 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, + "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)) + 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): @@ -355,34 +365,37 @@ def __init__(self, gateway, access_token, refresh_token): @classmethod def from_url(cls, gateway, url): - """Create an authentication using an OAuth callback 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) -> Tuple['Session', List[Dict[str, Any]]]: + 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. """ - 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') + 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. - """ + """Refresh the authentication, returning a new Auth object.""" - new_access_token = refresh_auth(self.gateway.oauth_root, - self.refresh_token) + new_access_token = refresh_auth( + self.gateway.oauth_root, 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, + "access_token": self.access_token, + "refresh_token": self.refresh_token, } @@ -398,7 +411,7 @@ def post(self, path, data=None): request from an active Session. """ - url = urljoin(self.auth.gateway.api_root + '/', path) + url = urljoin(self.auth.gateway.api_root + "/", path) return lgedm_post(url, data, self.auth.access_token, self.session_id) def get_devices(self) -> List[Dict[str, Any]]: @@ -407,7 +420,7 @@ def get_devices(self) -> List[Dict[str, Any]]: Return a list of dicts with information about the devices. """ - return get_list(self.post('device/deviceList'), 'item') + return get_list(self.post("device/deviceList"), "item") def monitor_start(self, device_id): """Begin monitoring a device's status. @@ -416,13 +429,16 @@ def monitor_start(self, device_id): monitoring. """ - res = self.post('rti/rtiMon', { - 'cmd': 'Mon', - 'cmdOpt': 'Start', - 'deviceId': device_id, - 'workId': gen_uuid(), - }) - return res['workId'] + 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. @@ -435,39 +451,42 @@ def monitor_poll(self, device_id, work_id): 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'] + 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: + if "returnCode" not in res: return None # Check for errors. - code = res.get('returnCode') # returnCode can be missing. - if code != '0000': + 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: + 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']) + 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, - }) + 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. @@ -475,28 +494,34 @@ def set_device_controls(self, device_id, values): `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': '', - }) + 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'): + 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'] + res = self.post( + "rti/rtiControl", + { + "cmd": category, + "cmdOpt": "Get", + "value": key, + "deviceId": device_id, + "workId": gen_uuid(), + "data": "", + }, + ) + return res["returnData"] diff --git a/wideq/dishwasher.py b/wideq/dishwasher.py index f912eff..1297b64 100644 --- a/wideq/dishwasher.py +++ b/wideq/dishwasher.py @@ -7,56 +7,58 @@ class DishWasherState(enum.Enum): """The state of the dishwasher device.""" - INITIAL = '@DW_STATE_INITIAL_W' - RUNNING = '@DW_STATE_RUNNING_W' + + 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' + 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' + "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' + + 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', + "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', + "Haeavy": "Heavy", } class DishWasherDevice(Device): """A higher-level interface for a dishwasher.""" - def poll(self) -> Optional['DishWasherStatus']: + def poll(self) -> Optional["DishWasherStatus"]: """Poll the device's current state. Monitoring must be started first with `monitor_start`. @@ -65,7 +67,7 @@ def poll(self) -> Optional['DishWasherStatus']: is not yet available. """ # Abort if monitoring has not started yet. - if not hasattr(self, 'mon'): + if not hasattr(self, "mon"): return None data = self.mon.poll() @@ -91,7 +93,8 @@ def __init__(self, dishwasher: DishWasherDevice, data: dict): def state(self) -> DishWasherState: """Get the state of the dishwasher.""" return DishWasherState( - lookup_enum('State', self.data, self.dishwasher)) + lookup_enum("State", self.data, self.dishwasher) + ) @property def readable_state(self) -> str: @@ -101,8 +104,8 @@ def readable_state(self) -> str: @property def process(self) -> Optional[DishWasherProcess]: """Get the process of the dishwasher.""" - process = lookup_enum('Process', self.data, self.dishwasher) - if process and process != '-': + process = lookup_enum("Process", self.data, self.dishwasher) + if process and process != "-": return DishWasherProcess(process) else: return None @@ -123,27 +126,28 @@ def is_on(self) -> bool: @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'])) + 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'])) + 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'])) + 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) + course = lookup_reference("Course", self.data, self.dishwasher) if course in DISHWASHER_COURSE_MAP: return DISHWASHER_COURSE_MAP[course] else: @@ -152,9 +156,9 @@ def course(self) -> str: @property def smart_course(self) -> str: """Get the current smart course.""" - return lookup_reference('SmartCourse', self.data, self.dishwasher) + 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) + return lookup_reference("Error", self.data, self.dishwasher) diff --git a/wideq/dryer.py b/wideq/dryer.py index 0b25203..49a3757 100644 --- a/wideq/dryer.py +++ b/wideq/dryer.py @@ -8,84 +8,84 @@ class DryerState(enum.Enum): """The state of the dryer device.""" - COOLING = '@WM_STATE_COOLING_W' - END = '@WM_STATE_END_W' - ERROR = '@WM_STATE_ERROR_W' - DRYING = '@WM_STATE_DRYING_W' - INITIAL = '@WM_STATE_INITIAL_W' - OFF = '@WM_STATE_POWER_OFF_W' - PAUSE = '@WM_STATE_PAUSE_W' - RUNNING = '@WM_STATE_RUNNING_W' - SMART_DIAGNOSIS = '@WM_STATE_SMART_DIAGNOSIS_W' - WRINKLE_CARE = '@WM_STATE_WRINKLECARE_W' + COOLING = "@WM_STATE_COOLING_W" + END = "@WM_STATE_END_W" + ERROR = "@WM_STATE_ERROR_W" + DRYING = "@WM_STATE_DRYING_W" + INITIAL = "@WM_STATE_INITIAL_W" + OFF = "@WM_STATE_POWER_OFF_W" + PAUSE = "@WM_STATE_PAUSE_W" + RUNNING = "@WM_STATE_RUNNING_W" + SMART_DIAGNOSIS = "@WM_STATE_SMART_DIAGNOSIS_W" + WRINKLE_CARE = "@WM_STATE_WRINKLECARE_W" UNKNOWN = _UNKNOWN class DryLevel(enum.Enum): """Represents the dry level setting of the dryer.""" - CUPBOARD = '@WM_DRY27_DRY_LEVEL_CUPBOARD_W' - DAMP = '@WM_DRY27_DRY_LEVEL_DAMP_W' - EXTRA = '@WM_DRY27_DRY_LEVEL_EXTRA_W' - IRON = '@WM_DRY27_DRY_LEVEL_IRON_W' - LESS = '@WM_DRY27_DRY_LEVEL_LESS_W' - MORE = '@WM_DRY27_DRY_LEVEL_MORE_W' - NORMAL = '@WM_DRY27_DRY_LEVEL_NORMAL_W' - OFF = '-' - VERY = '@WM_DRY27_DRY_LEVEL_VERY_W' + CUPBOARD = "@WM_DRY27_DRY_LEVEL_CUPBOARD_W" + DAMP = "@WM_DRY27_DRY_LEVEL_DAMP_W" + EXTRA = "@WM_DRY27_DRY_LEVEL_EXTRA_W" + IRON = "@WM_DRY27_DRY_LEVEL_IRON_W" + LESS = "@WM_DRY27_DRY_LEVEL_LESS_W" + MORE = "@WM_DRY27_DRY_LEVEL_MORE_W" + NORMAL = "@WM_DRY27_DRY_LEVEL_NORMAL_W" + OFF = "-" + VERY = "@WM_DRY27_DRY_LEVEL_VERY_W" UNKNOWN = _UNKNOWN class DryerError(enum.Enum): """A dryer error.""" - ERROR_AE = '@WM_US_DRYER_ERROR_AE_W' - ERROR_CE1 = '@WM_US_DRYER_ERROR_CE1_W' - ERROR_DE4 = '@WM_WW_FL_ERROR_DE4_W' - ERROR_DOOR = '@WM_US_DRYER_ERROR_DE_W' - ERROR_DRAINMOTOR = '@WM_US_DRYER_ERROR_OE_W' - ERROR_EMPTYWATER = '@WM_US_DRYER_ERROR_EMPTYWATER_W' - ERROR_F1 = '@WM_US_DRYER_ERROR_F1_W' - ERROR_LE1 = '@WM_US_DRYER_ERROR_LE1_W' - ERROR_LE2 = '@WM_US_DRYER_ERROR_LE2_W' - ERROR_NOFILTER = '@WM_US_DRYER_ERROR_NOFILTER_W' - ERROR_NP = '@WM_US_DRYER_ERROR_NP_GAS_W' - ERROR_PS = '@WM_US_DRYER_ERROR_PS_W' - ERROR_TE1 = '@WM_US_DRYER_ERROR_TE1_W' - ERROR_TE2 = '@WM_US_DRYER_ERROR_TE2_W' - ERROR_TE5 = '@WM_US_DRYER_ERROR_TE5_W' - ERROR_TE6 = '@WM_US_DRYER_ERROR_TE6_W' + ERROR_AE = "@WM_US_DRYER_ERROR_AE_W" + ERROR_CE1 = "@WM_US_DRYER_ERROR_CE1_W" + ERROR_DE4 = "@WM_WW_FL_ERROR_DE4_W" + ERROR_DOOR = "@WM_US_DRYER_ERROR_DE_W" + ERROR_DRAINMOTOR = "@WM_US_DRYER_ERROR_OE_W" + ERROR_EMPTYWATER = "@WM_US_DRYER_ERROR_EMPTYWATER_W" + ERROR_F1 = "@WM_US_DRYER_ERROR_F1_W" + ERROR_LE1 = "@WM_US_DRYER_ERROR_LE1_W" + ERROR_LE2 = "@WM_US_DRYER_ERROR_LE2_W" + ERROR_NOFILTER = "@WM_US_DRYER_ERROR_NOFILTER_W" + ERROR_NP = "@WM_US_DRYER_ERROR_NP_GAS_W" + ERROR_PS = "@WM_US_DRYER_ERROR_PS_W" + ERROR_TE1 = "@WM_US_DRYER_ERROR_TE1_W" + ERROR_TE2 = "@WM_US_DRYER_ERROR_TE2_W" + ERROR_TE5 = "@WM_US_DRYER_ERROR_TE5_W" + ERROR_TE6 = "@WM_US_DRYER_ERROR_TE6_W" UNKNOWN = _UNKNOWN class TempControl(enum.Enum): """Represents temperature control setting.""" - OFF = '-' - ULTRA_LOW = '@WM_DRY27_TEMP_ULTRA_LOW_W' - LOW = '@WM_DRY27_TEMP_LOW_W' - MEDIUM = '@WM_DRY27_TEMP_MEDIUM_W' - MID_HIGH = '@WM_DRY27_TEMP_MID_HIGH_W' - HIGH = '@WM_DRY27_TEMP_HIGH_W' + OFF = "-" + ULTRA_LOW = "@WM_DRY27_TEMP_ULTRA_LOW_W" + LOW = "@WM_DRY27_TEMP_LOW_W" + MEDIUM = "@WM_DRY27_TEMP_MEDIUM_W" + MID_HIGH = "@WM_DRY27_TEMP_MID_HIGH_W" + HIGH = "@WM_DRY27_TEMP_HIGH_W" UNKNOWN = _UNKNOWN class TimeDry(enum.Enum): """Represents a timed dry setting.""" - OFF = '-' - TWENTY = '20' - THIRTY = '30' - FOURTY = '40' - FIFTY = '50' - SIXTY = '60' + OFF = "-" + TWENTY = "20" + THIRTY = "30" + FOURTY = "40" + FIFTY = "50" + SIXTY = "60" UNKNOWN = _UNKNOWN class DryerDevice(Device): """A higher-level interface for a dryer.""" - def poll(self) -> Optional['DryerStatus']: + def poll(self) -> Optional["DryerStatus"]: """Poll the device's current state. Monitoring must be started first with `monitor_start`. @@ -94,7 +94,7 @@ def poll(self) -> Optional['DryerStatus']: not yet available. """ # Abort if monitoring has not started yet. - if not hasattr(self, 'mon'): + if not hasattr(self, "mon"): return None data = self.mon.poll() @@ -121,34 +121,34 @@ def get_bit(self, key: str, index: int) -> str: bit_index = 2 ** index mode = bin(bit_value & bit_index) if mode == bin(0): - return 'OFF' + return "OFF" else: - return 'ON' + return "ON" @property def state(self) -> DryerState: """Get the state of the dryer.""" - return DryerState(lookup_enum('State', self.data, self.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)) + 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)) + 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)) + 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)) + return TimeDry(lookup_enum("TimeDry", self.data, self.dryer)) @property def is_on(self) -> bool: @@ -158,27 +158,28 @@ def is_on(self) -> bool: @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'])) + 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'])) + 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) + 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) + 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) + return lookup_reference("Error", self.data, self.dryer) diff --git a/wideq/refrigerator.py b/wideq/refrigerator.py index 7dff39f..b6847e6 100644 --- a/wideq/refrigerator.py +++ b/wideq/refrigerator.py @@ -38,18 +38,16 @@ 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) + """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) + """Set the freezer temperature in Celsius.""" + value = self.model.enum_value("TempFreezer", str(temp)) + self._set_control("REFT", value) - def poll(self) -> Optional['RefrigeratorStatus']: + def poll(self) -> Optional["RefrigeratorStatus"]: """Poll the device's current state. Monitoring must be started first with `monitor_start`. @@ -58,7 +56,7 @@ def poll(self) -> Optional['RefrigeratorStatus']: status is not yet available. """ # Abort if monitoring has not started yet. - if not hasattr(self, 'mon'): + if not hasattr(self, "mon"): return None data = self.mon.poll() @@ -82,59 +80,59 @@ def __init__(self, refrigerator: RefrigeratorDevice, data: dict): @property def temp_refrigerator_c(self): - temp = lookup_enum('TempRefrigerator', self.data, self.refrigerator) + 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) + 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) + 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) + 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) + mode = lookup_enum("SmartSavingMode", self.data, self.refrigerator) return SmartSavingMode(mode) @property def door_opened(self): - state = lookup_enum('DoorOpenState', self.data, self.refrigerator) + state = lookup_enum("DoorOpenState", self.data, self.refrigerator) return state == "OPEN" @property def temp_unit(self): - return lookup_enum('TempUnit', self.data, self.refrigerator) + return lookup_enum("TempUnit", self.data, self.refrigerator) @property def energy_saving_enabled(self): mode = lookup_enum( - 'SmartSavingModeStatus', self.data, self.refrigerator + "SmartSavingModeStatus", self.data, self.refrigerator ) - return mode == 'ON' + return mode == "ON" @property def locked(self): - status = lookup_enum('LockingStatus', self.data, self.refrigerator) + status = lookup_enum("LockingStatus", self.data, self.refrigerator) return status == "LOCK" @property def active_saving_status(self): - return self.data['ActiveSavingStatus'] + return self.data["ActiveSavingStatus"] @property def eco_enabled(self): - eco = lookup_enum('EcoFriendly', self.data, self.refrigerator) + 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'] + return self.data["WaterFilterUsedMonth"] diff --git a/wideq/util.py b/wideq/util.py index 200ecf0..88dae23 100644 --- a/wideq/util.py +++ b/wideq/util.py @@ -3,7 +3,7 @@ from .client import Device, DeviceType -T = TypeVar('T', bound=Device) +T = TypeVar("T", bound=Device) def lookup_enum(attr: str, data: dict, device: T): @@ -27,13 +27,12 @@ def lookup_reference(attr: str, data: dict, device: T) -> str: """ value = device.model.reference_name(attr, data[attr]) if value is None: - return 'Off' + return "Off" return value def device_classes(): - """The mapping of every Device subclass related to the DeviceType enum - """ + """The mapping of every Device subclass related to the DeviceType enum""" from .ac import ACDevice from .dryer import DryerDevice from .dishwasher import DishWasherDevice diff --git a/wideq/washer.py b/wideq/washer.py index 974d3be..2a793ea 100644 --- a/wideq/washer.py +++ b/wideq/washer.py @@ -8,36 +8,36 @@ 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' + 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']: + def poll(self) -> Optional["WasherStatus"]: """Poll the device's current state. Monitoring must be started first with `monitor_start`. @@ -46,7 +46,7 @@ def poll(self) -> Optional['WasherStatus']: not yet available. """ # Abort if monitoring has not started yet. - if not hasattr(self, 'mon'): + if not hasattr(self, "mon"): return None data = self.mon.poll() @@ -71,12 +71,12 @@ def __init__(self, washer: WasherDevice, data: dict): @property def state(self) -> WasherState: """Get the state of the washer.""" - return WasherState(lookup_enum('State', self.data, self.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)) + return WasherState(lookup_enum("PreState", self.data, self.washer)) @property def is_on(self) -> bool: @@ -86,15 +86,16 @@ def is_on(self) -> bool: @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'])) + 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'])) + 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. @@ -104,20 +105,20 @@ def _lookup_reference(self, attr: str) -> str: """ value = self.washer.model.reference_name(attr, self.data[attr]) if value is None: - return 'Off' + return "Off" return value @property def course(self) -> str: """Get the current course.""" - return lookup_reference('APCourse', self.data, self.washer) + 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) + 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) + return lookup_reference("Error", self.data, self.washer) From 4612ca07d674a8259bb3cbd74970517708aab361 Mon Sep 17 00:00:00 2001 From: Adrian Sampson Date: Sat, 28 Nov 2020 17:01:05 -0500 Subject: [PATCH 2/5] Expand Black discussion in README --- README.md | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index b82e5ac..cabff02 100644 --- a/README.md +++ b/README.md @@ -31,10 +31,14 @@ You can also specify one of several other commands: * `ac-config `: Print out some configuration information about an AC device. Development -------- -To ensure consistent formatting across pull requests, install the precommit hooks to auto format your code using `pre-commit install`. +----------- + +This project uses the [Black][] code formatting tool. Before submitting pull requests, please run Black to ensure consistent formatting. + +If you like, you can install a git hook to automatically run Black and flake8 every time you commit. Install the [pre-commit][] tool and type `pre-commit install` to use it. -The code will be auto-formatted by `black` to ensure consistent style. +[black]: https://github.com/psf/black +[pre-commit]: https://pre-commit.com/ Credits ------- From 717391dae8deea0eb90f001ad4c1e75a9cdf3b60 Mon Sep 17 00:00:00 2001 From: Adrian Sampson Date: Sat, 28 Nov 2020 17:01:22 -0500 Subject: [PATCH 3/5] Reorg links --- README.md | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index cabff02..b0e11df 100644 --- a/README.md +++ b/README.md @@ -37,9 +37,6 @@ This project uses the [Black][] code formatting tool. Before submitting pull req If you like, you can install a git hook to automatically run Black and flake8 every time you commit. Install the [pre-commit][] tool and type `pre-commit install` to use it. -[black]: https://github.com/psf/black -[pre-commit]: https://pre-commit.com/ - Credits ------- @@ -50,3 +47,5 @@ I also made a [Home Assistant component][hass-smartthinq] that uses wideq. [hass-smartthinq]: https://github.com/sampsyo/hass-smartthinq [adrian]: https://github.com/sampsyo [mit]: https://opensource.org/licenses/MIT +[black]: https://github.com/psf/black +[pre-commit]: https://pre-commit.com/ From 6c2232a7dbcf9d0989a57cfc9e5782345aa5d7b1 Mon Sep 17 00:00:00 2001 From: Adrian Sampson Date: Sat, 28 Nov 2020 17:02:08 -0500 Subject: [PATCH 4/5] pre-commit is not a dependency --- pyproject.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ff12792..c27db83 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,5 +1,5 @@ [build-system] -requires = ["flit", "pre-commit"] +requires = ["flit"] build-backend = "flit.buildapi" [tool.flit.metadata] @@ -31,4 +31,4 @@ exclude = ''' | build | dist )/ -''' \ No newline at end of file +''' From 8319b2ef973cebd479165ef9e61be94380195dd5 Mon Sep 17 00:00:00 2001 From: Adrian Sampson Date: Sat, 28 Nov 2020 17:05:17 -0500 Subject: [PATCH 5/5] Run Black on example.py --- example.py | 162 +++++++++++++++++++++++++++++++---------------------- 1 file changed, 94 insertions(+), 68 deletions(-) diff --git a/example.py b/example.py index efe516f..d72a1d6 100755 --- a/example.py +++ b/example.py @@ -10,7 +10,7 @@ import logging from typing import List -STATE_FILE = 'wideq_state.json' +STATE_FILE = "wideq_state.json" LOGGER = logging.getLogger("wideq.example") @@ -20,9 +20,9 @@ def authenticate(gateway): """ login_url = gateway.oauth_url() - print('Log in here:') + print("Log in here:") print(login_url) - print('Then paste the URL where the browser is redirected:') + print("Then paste the URL where the browser is redirected:") callback_url = input() return wideq.Auth.from_url(gateway, callback_url) @@ -31,7 +31,7 @@ def ls(client): """List the user's devices.""" for device in client.devices: - print('{0.id}: {0.name} ({0.type.name} {0.model_id})'.format(device)) + print("{0.id}: {0.name} ({0.type.name} {0.model_id})".format(device)) def gen_mon(client, device_id): @@ -46,27 +46,33 @@ def gen_mon(client, device_id): try: while True: time.sleep(1) - print('Polling...') + print("Polling...") data = mon.poll() if data: try: res = model.decode_monitor(data) except ValueError: - print('status data: {!r}'.format(data)) + print("status data: {!r}".format(data)) else: for key, value in res.items(): try: desc = model.value(key) except KeyError: - print('- {}: {}'.format(key, value)) + print("- {}: {}".format(key, value)) if isinstance(desc, wideq.EnumValue): - print('- {}: {}'.format( - key, desc.options.get(value, value) - )) + print( + "- {}: {}".format( + key, desc.options.get(value, value) + ) + ) elif isinstance(desc, wideq.RangeValue): - print('- {0}: {1} ({2.min}-{2.max})'.format( - key, value, desc, - )) + print( + "- {0}: {1} ({2.min}-{2.max})".format( + key, + value, + desc, + ) + ) except KeyboardInterrupt: pass @@ -80,7 +86,7 @@ def ac_mon(ac): try: ac.monitor_start() except wideq.core.NotConnectedError: - print('Device not available.') + print("Device not available.") return try: @@ -89,18 +95,16 @@ def ac_mon(ac): state = ac.poll() if state: print( - '{1}; ' - '{0.mode.name}; ' - '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' + "{1}; " + "{0.mode.name}; " + "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" ) ) else: - print('no state. Wait 1 more second.') + print("no state. Wait 1 more second.") except KeyboardInterrupt: pass @@ -121,8 +125,8 @@ def mon(client, device_id): class UserError(Exception): - """A user-visible command-line error. - """ + """A user-visible command-line error.""" + def __init__(self, msg): self.msg = msg @@ -147,11 +151,13 @@ def set_temp(client, device_id, temp): ac.set_fahrenheit(int(temp)) elif device.type == wideq.client.DeviceType.REFRIGERATOR: refrigerator = wideq.RefrigeratorDevice( - client, _force_device(client, device_id)) + client, _force_device(client, device_id) + ) refrigerator.set_temp_refrigerator_c(int(temp)) else: raise UserError( - 'set-temp only suported for AC or refrigerator devices') + "set-temp only suported for AC or refrigerator devices" + ) def set_temp_freezer(client, device_id, temp): @@ -161,18 +167,20 @@ def set_temp_freezer(client, device_id, temp): if device.type == wideq.client.DeviceType.REFRIGERATOR: refrigerator = wideq.RefrigeratorDevice( - client, _force_device(client, device_id)) + client, _force_device(client, device_id) + ) refrigerator.set_temp_freezer_c(int(temp)) else: raise UserError( - 'set-temp-freezer only suported for refrigerator devices') + "set-temp-freezer only suported for refrigerator devices" + ) def turn(client, device_id, on_off): """Turn on/off an AC device.""" ac = wideq.ACDevice(client, _force_device(client, device_id)) - ac.set_on(on_off == 'on') + ac.set_on(on_off == "on") def ac_config(client, device_id): @@ -190,26 +198,30 @@ def ac_config(client, device_id): EXAMPLE_COMMANDS = { - 'ls': ls, - 'mon': mon, - 'set-temp': set_temp, - 'set-temp-freezer': set_temp_freezer, - 'turn': turn, - 'ac-config': ac_config, + "ls": ls, + "mon": mon, + "set-temp": set_temp, + "set-temp-freezer": set_temp_freezer, + "turn": turn, + "ac-config": ac_config, } def example_command(client, cmd, args): func = EXAMPLE_COMMANDS.get(cmd) if not func: - LOGGER.error("Invalid command: '%s'.\n" - "Use one of: %s", cmd, ', '.join(EXAMPLE_COMMANDS)) + LOGGER.error( + "Invalid command: '%s'.\n" "Use one of: %s", + cmd, + ", ".join(EXAMPLE_COMMANDS), + ) return func(client, *args) -def example(country: str, language: str, verbose: bool, - cmd: str, args: List[str]) -> None: +def example( + country: str, language: str, verbose: bool, cmd: str, args: List[str] +) -> None: if verbose: wideq.set_log_level(logging.DEBUG) @@ -220,8 +232,9 @@ def example(country: str, language: str, verbose: bool, state = json.load(f) except IOError: state = {} - LOGGER.debug("No state file found (tried: '%s')", - os.path.abspath(STATE_FILE)) + LOGGER.debug( + "No state file found (tried: '%s')", os.path.abspath(STATE_FILE) + ) client = wideq.Client.load(state) if country: @@ -240,7 +253,7 @@ def example(country: str, language: str, verbose: bool, break except wideq.NotLoggedInError: - LOGGER.info('Session expired.') + LOGGER.info("Session expired.") client.refresh() except UserError as exc: @@ -249,54 +262,67 @@ def example(country: str, language: str, verbose: bool, # 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() -> None: - """The main command-line entry point. - """ + """The main command-line entry point.""" parser = argparse.ArgumentParser( - description='Interact with the LG SmartThinQ API.' + description="Interact with the LG SmartThinQ API." + ) + parser.add_argument( + "cmd", + metavar="CMD", + nargs="?", + default="ls", + help=f'one of: {", ".join(EXAMPLE_COMMANDS)}', + ) + parser.add_argument( + "args", metavar="ARGS", nargs="*", help="subcommand arguments" ) - parser.add_argument('cmd', metavar='CMD', nargs='?', default='ls', - help=f'one of: {", ".join(EXAMPLE_COMMANDS)}') - parser.add_argument('args', metavar='ARGS', nargs='*', - help='subcommand arguments') parser.add_argument( - '--country', '-c', - help=f'country code for account (default: {wideq.DEFAULT_COUNTRY})', - default=wideq.DEFAULT_COUNTRY + "--country", + "-c", + help=f"country code for account (default: {wideq.DEFAULT_COUNTRY})", + default=wideq.DEFAULT_COUNTRY, ) parser.add_argument( - '--language', '-l', - help=f'language code for the API (default: {wideq.DEFAULT_LANGUAGE})', - default=wideq.DEFAULT_LANGUAGE + "--language", + "-l", + help=f"language code for the API (default: {wideq.DEFAULT_LANGUAGE})", + default=wideq.DEFAULT_LANGUAGE, ) parser.add_argument( - '--verbose', '-v', - help='verbose mode to help debugging', - action='store_true', default=False + "--verbose", + "-v", + help="verbose mode to help debugging", + action="store_true", + default=False, ) args = parser.parse_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) + 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) + 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__': +if __name__ == "__main__": main()