v/wideq
1
0
Fork 0
mirror of https://github.com/no2chem/wideq.git synced 2025-05-15 14:50:28 -07:00
Code Issues Projects Releases Wiki Activity

introduce black formatter

Browse Source
This commit is contained in:
Michael Wei 2020-11-21 19:46:57 +00:00
parent 6e4f9d6484
commit fb3c755c2e
18 changed files with 807 additions and 695 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.flake8 Normal file
View File

@ -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

10
.pre-commit-config.yaml Normal file
View File

@ -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

5
README.md
View File

@ -30,6 +30,11 @@ You can also specify one of several other commands:
* `turn <ID> <ONOFF>`: Turn an AC device on or off. Use "on" or "off" as the second argument.
* `ac-config <ID>`: 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
-------

18
pyproject.toml
View File

@ -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
)/
'''

181
tests/test_client.py
View File

@ -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'
"type": "Bit",
},
'TimeBsOn': {
'_comment':
'오전 12시 30분은 0030, 오후12시30분은 1230 ,오후 4시30분은 1630 off는 0 ',
'type': 'String'
},
'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")

57
tests/test_core.py
View File

@ -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")

39
tests/test_dishwasher.py
View File

@ -2,8 +2,11 @@ import json
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 @@ POLL_DATA = {
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 @@ class DishWasherStatusTest(unittest.TestCase):
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)

86
tests/test_dryer.py
View File

@ -4,47 +4,55 @@ from unittest import mock
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 @@ class DryerStatusTest(unittest.TestCase):
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)

76
tests/test_washer.py
View File

@ -6,48 +6,50 @@ from wideq.washer import WasherDevice, WasherState, WasherStatus
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 @@ class WasherStatusTest(unittest.TestCase):
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)

2
wideq/__init__.py
View File

@ -8,4 +8,4 @@ from .dryer import * # noqa
from .refrigerator import * # noqa
from .washer import * # noqa
__version__ = '1.5.0'
__version__ = "1.5.0"

223
wideq/ac.py
View File

@ -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 @@ class ACDevice(Device):
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 @@ class ACDevice(Device):
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 @@ class ACDevice(Device):
@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 @@ class ACDevice(Device):
# 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 @@ class ACDevice(Device):
# 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 @@ class ACDevice(Device):
`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 @@ class ACDevice(Device):
"""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 @@ class ACDevice(Device):
"""
# 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 @@ class ACDevice(Device):
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 @@ class ACStatus(object):
@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 @@ class ACStatus(object):
@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 @@ class ACStatus(object):
@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):

188
wideq/client.py
View File

@ -14,7 +14,7 @@ from . import core
#: Represents an unknown enum value.
_UNKNOWN = 'Unknown'
_UNKNOWN = "Unknown"
LOGGER = logging.getLogger("wideq.client")
@ -53,7 +53,7 @@ class Monitor(object):
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 @@ class Monitor(object):
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 @@ class Client(object):
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 @@ class Client(object):
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 @@ class Client(object):
"""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 @@ class Client(object):
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 @@ class Client(object):
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 @@ class DeviceInfo(object):
@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 @@ class ModelInfo(object):
`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 @@ class ModelInfo(object):
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 @@ class Device(object):
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 @@ class Device(object):
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):

271
wideq/core.py
View File

@ -12,17 +12,17 @@ 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'
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 @@ LOGGER = get_wideq_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 @@ class Gateway(object):
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 @@ class Auth(object):
@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 @@ class Session(object):
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 @@ class Session(object):
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 @@ class Session(object):
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 @@ class Session(object):
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 @@ class Session(object):
`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"]

86
wideq/dishwasher.py
View File

@ -7,56 +7,58 @@ 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'
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 @@ class DishWasherDevice(Device):
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 @@ class DishWasherStatus(object):
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 @@ class DishWasherStatus(object):
@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 @@ class DishWasherStatus(object):
@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 @@ class DishWasherStatus(object):
@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)

129
wideq/dryer.py
View File

@ -8,84 +8,84 @@ from .util import lookup_enum, lookup_reference
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 @@ class DryerDevice(Device):
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 @@ class DryerStatus(object):
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 @@ class DryerStatus(object):
@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)

44
wideq/refrigerator.py
View File

@ -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 @@ class RefrigeratorDevice(Device):
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 @@ class RefrigeratorStatus(object):
@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"]

7
wideq/util.py
View File

@ -3,7 +3,7 @@ from typing import TypeVar
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

75
wideq/washer.py
View File

@ -8,36 +8,36 @@ from .util import lookup_enum, lookup_reference
class WasherState(enum.Enum):
"""The state of the washer device."""
ADD_DRAIN = '@WM_STATE_ADD_DRAIN_W'
COMPLETE = '@WM_STATE_COMPLETE_W'
DETECTING = '@WM_STATE_DETECTING_W'
DETERGENT_AMOUNT = '@WM_STATE_DETERGENT_AMOUNT_W'
DRYING = '@WM_STATE_DRYING_W'
END = '@WM_STATE_END_W'
ERROR_AUTO_OFF = '@WM_STATE_ERROR_AUTO_OFF_W'
FRESH_CARE = '@WM_STATE_FRESHCARE_W'
FROZEN_PREVENT_INITIAL = '@WM_STATE_FROZEN_PREVENT_INITIAL_W'
FROZEN_PREVENT_PAUSE = '@WM_STATE_FROZEN_PREVENT_PAUSE_W'
FROZEN_PREVENT_RUNNING = '@WM_STATE_FROZEN_PREVENT_RUNNING_W'
INITIAL = '@WM_STATE_INITIAL_W'
OFF = '@WM_STATE_POWER_OFF_W'
PAUSE = '@WM_STATE_PAUSE_W'
PRE_WASH = '@WM_STATE_PREWASH_W'
RESERVE = '@WM_STATE_RESERVE_W'
RINSING = '@WM_STATE_RINSING_W'
RINSE_HOLD = '@WM_STATE_RINSE_HOLD_W'
RUNNING = '@WM_STATE_RUNNING_W'
SMART_DIAGNOSIS = '@WM_STATE_SMART_DIAG_W'
SMART_DIAGNOSIS_DATA = '@WM_STATE_SMART_DIAGDATA_W'
SPINNING = '@WM_STATE_SPINNING_W'
TCL_ALARM_NORMAL = 'TCL_ALARM_NORMAL'
TUBCLEAN_COUNT_ALARM = '@WM_STATE_TUBCLEAN_COUNT_ALRAM_W'
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 @@ class WasherDevice(Device):
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 @@ class WasherStatus(object):
@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 @@ class WasherStatus(object):
@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 @@ class WasherStatus(object):
"""
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)