mirror of
https://github.com/no2chem/wideq.git
synced 2025-05-16 07:10:09 -07:00
Adding a washer device.
This commit is contained in:
parent
c0b1db0bd4
commit
89967d0d76
62
tests/test_washer.py
Normal file
62
tests/test_washer.py
Normal file
@ -0,0 +1,62 @@
|
|||||||
|
import json
|
||||||
|
import unittest
|
||||||
|
|
||||||
|
from wideq.client import Client, DeviceInfo
|
||||||
|
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',
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class WasherStatusTest(unittest.TestCase):
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
super().setUp()
|
||||||
|
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.washer = WasherDevice(self.client, self.device_info)
|
||||||
|
|
||||||
|
def test_properties(self):
|
||||||
|
status = WasherStatus(self.washer, POLL_DATA)
|
||||||
|
self.assertEqual(WasherState.RINSING, status.state)
|
||||||
|
self.assertEqual(WasherState.RUNNING, status.previous_state)
|
||||||
|
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)
|
123
wideq/washer.py
Normal file
123
wideq/washer.py
Normal file
@ -0,0 +1,123 @@
|
|||||||
|
import enum
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
from .client import Device
|
||||||
|
|
||||||
|
|
||||||
|
class WasherState(enum.Enum):
|
||||||
|
"""The state of the washer device."""
|
||||||
|
|
||||||
|
ADD_DRAIN = '@WM_STATE_ADD_DRAIN_W'
|
||||||
|
DETECTING = '@WM_STATE_DETECTING_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'
|
||||||
|
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'
|
||||||
|
TUBCLEAN_COUNT_ALARM = '@WM_STATE_TUBCLEAN_COUNT_ALRAM_W'
|
||||||
|
|
||||||
|
|
||||||
|
class WasherDevice(Device):
|
||||||
|
"""A higher-level interface for a washer."""
|
||||||
|
|
||||||
|
def poll(self) -> Optional['WasherStatus']:
|
||||||
|
"""Poll the device's current state.
|
||||||
|
|
||||||
|
Monitoring must be started first with `monitor_start`.
|
||||||
|
|
||||||
|
:returns: Either a `WasherStatus` instance or `None` if the status is
|
||||||
|
not yet available.
|
||||||
|
"""
|
||||||
|
# Abort if monitoring has not started yet.
|
||||||
|
if not hasattr(self, 'mon'):
|
||||||
|
return None
|
||||||
|
|
||||||
|
res = self.mon.poll_json()
|
||||||
|
if res:
|
||||||
|
return WasherStatus(self, res)
|
||||||
|
else:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
class WasherStatus(object):
|
||||||
|
"""Higher-level information about a washer's current status.
|
||||||
|
|
||||||
|
:param washer: The WasherDevice instance.
|
||||||
|
:param data: JSON data from the API.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, washer: WasherDevice, data: dict):
|
||||||
|
self.washer = washer
|
||||||
|
self.data = data
|
||||||
|
|
||||||
|
def _lookup_enum(self, attr: str) -> str:
|
||||||
|
"""Looks up an enum value for the provided attr.
|
||||||
|
|
||||||
|
:param attr: The attribute to lookup in the enum.
|
||||||
|
:returns: The enum value.
|
||||||
|
"""
|
||||||
|
return self.washer.model.enum_name(attr, self.data[attr])
|
||||||
|
|
||||||
|
@property
|
||||||
|
def state(self) -> WasherState:
|
||||||
|
"""Get the state of the washer."""
|
||||||
|
return WasherState(self._lookup_enum('State'))
|
||||||
|
|
||||||
|
@property
|
||||||
|
def previous_state(self) -> WasherState:
|
||||||
|
"""Get the previous state of the washer."""
|
||||||
|
return WasherState(self._lookup_enum('PreState'))
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_on(self) -> bool:
|
||||||
|
"""Check if the washer is on or not."""
|
||||||
|
return self.state != WasherState.OFF
|
||||||
|
|
||||||
|
@property
|
||||||
|
def remaining_time(self) -> int:
|
||||||
|
"""Get the remaining time in minutes."""
|
||||||
|
return (int(self.data['Remain_Time_H']) * 60 +
|
||||||
|
int(self.data['Remain_Time_M']))
|
||||||
|
|
||||||
|
@property
|
||||||
|
def initial_time(self) -> int:
|
||||||
|
"""Get the initial time in minutes."""
|
||||||
|
return (
|
||||||
|
int(self.data['Initial_Time_H']) * 60 +
|
||||||
|
int(self.data['Initial_Time_M']))
|
||||||
|
|
||||||
|
def _lookup_reference(self, attr: str) -> str:
|
||||||
|
"""Look up a reference value for the provided attribute.
|
||||||
|
|
||||||
|
:param attr: The attribute to find the value for.
|
||||||
|
:returns: The looked up value.
|
||||||
|
"""
|
||||||
|
value = self.washer.model.reference_name(attr, self.data[attr])
|
||||||
|
if value is None:
|
||||||
|
return 'Off'
|
||||||
|
return value
|
||||||
|
|
||||||
|
@property
|
||||||
|
def course(self) -> str:
|
||||||
|
"""Get the current course."""
|
||||||
|
return self._lookup_reference('APCourse')
|
||||||
|
|
||||||
|
@property
|
||||||
|
def smart_course(self) -> str:
|
||||||
|
"""Get the current smart course."""
|
||||||
|
return self._lookup_reference('SmartCourse')
|
||||||
|
|
||||||
|
@property
|
||||||
|
def error(self) -> str:
|
||||||
|
"""Get the current error."""
|
||||||
|
return self._lookup_reference('Error')
|
Loading…
x
Reference in New Issue
Block a user