Compare commits

...

11 Commits

Author SHA1 Message Date
04f19c4609 Fix error when sending commands 2023-04-17 00:07:22 +02:00
69be63df2a Bump version to v0.8.0b5 2023-04-16 13:57:40 +02:00
6c44aa895d Disable zones for devices with just 1 zone, fixes #11 2023-04-16 13:33:21 +02:00
8372c75e30 expose ancillary parameter in settings 2023-04-16 13:11:56 +02:00
5b91747ec1 Add dish washer 2023-04-16 02:46:30 +02:00
8da2018302 Set fixed values 2023-04-16 02:44:20 +02:00
03187745bf Split up parameters 2023-04-16 01:43:37 +02:00
461a247ad3 More type hints 2023-04-16 01:36:10 +02:00
834f25a639 Remove filters, filter out recies #9 2023-04-16 00:40:59 +02:00
46ff9be4a2 Fix code depts 2023-04-15 23:02:37 +02:00
a1618bb18a Fix missing zone attribute 2023-04-15 22:25:34 +02:00
22 changed files with 291 additions and 253 deletions

View File

@ -25,7 +25,7 @@ jobs:
run: | run: |
python -m pip install --upgrade pip python -m pip install --upgrade pip
python -m pip install -r requirements.txt python -m pip install -r requirements.txt
python -m pip install flake8 pylint black mypy python -m pip install -r requirements_dev.txt
- name: Lint with flake8 - name: Lint with flake8
run: | run: |
# stop the build if there are Python syntax errors or undefined names # stop the build if there are Python syntax errors or undefined names

1
.gitignore vendored
View File

@ -4,3 +4,4 @@ __pycache__/
dist/ dist/
**/*.egg-info/ **/*.egg-info/
test* test*
build/

View File

@ -6,7 +6,7 @@
[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/pyhOn)](https://www.python.org/) [![PyPI - Python Version](https://img.shields.io/pypi/pyversions/pyhOn)](https://www.python.org/)
[![PyPI - License](https://img.shields.io/pypi/l/pyhOn)](https://github.com/Andre0512/pyhOn/blob/main/LICENSE) [![PyPI - License](https://img.shields.io/pypi/l/pyhOn)](https://github.com/Andre0512/pyhOn/blob/main/LICENSE)
[![PyPI - Downloads](https://img.shields.io/pypi/dm/pyhOn)](https://pypistats.org/packages/pyhon) [![PyPI - Downloads](https://img.shields.io/pypi/dm/pyhOn)](https://pypistats.org/packages/pyhon)
Control your Haier appliances with python! Control your Haier, Candy and Hoover appliances with python!
The idea behind this library is, to make the use of all available commands as simple as possible. The idea behind this library is, to make the use of all available commands as simple as possible.
## Installation ## Installation

View File

@ -5,7 +5,7 @@ from typing import TYPE_CHECKING
from pyhon import helper from pyhon import helper
from pyhon.commands import HonCommand from pyhon.commands import HonCommand
from pyhon.parameter import HonParameterFixed from pyhon.parameter.fixed import HonParameterFixed
if TYPE_CHECKING: if TYPE_CHECKING:
from pyhon import HonAPI from pyhon import HonAPI
@ -24,7 +24,7 @@ class HonAppliance:
self._commands: Dict = {} self._commands: Dict = {}
self._statistics: Dict = {} self._statistics: Dict = {}
self._attributes: Dict = {} self._attributes: Dict = {}
self._zone = zone self._zone: int = zone
try: try:
self._extra = importlib.import_module( self._extra = importlib.import_module(
@ -106,6 +106,10 @@ class HonAppliance:
def info(self): def info(self):
return self._info return self._info
@property
def zone(self) -> int:
return self._zone
async def _recover_last_command_states(self, commands): async def _recover_last_command_states(self, commands):
command_history = await self._api.command_history(self) command_history = await self._api.command_history(self)
for name, command in commands.items(): for name, command in commands.items():
@ -120,8 +124,8 @@ class HonAppliance:
if last is None: if last is None:
continue continue
parameters = command_history[last].get("command", {}).get("parameters", {}) parameters = command_history[last].get("command", {}).get("parameters", {})
if command._multi and parameters.get("program"): if command.programs and parameters.get("program"):
command.set_program(parameters.pop("program").split(".")[-1].lower()) command.program = parameters.pop("program").split(".")[-1].lower()
command = self.commands[name] command = self.commands[name]
for key, data in command.settings.items(): for key, data in command.settings.items():
if ( if (
@ -145,7 +149,12 @@ class HonAppliance:
for program, attr2 in attr.items(): for program, attr2 in attr.items():
program = program.split(".")[-1].lower() program = program.split(".")[-1].lower()
cmd = HonCommand( cmd = HonCommand(
command, attr2, self._api, self, multi=multi, program=program command,
attr2,
self._api,
self,
programs=multi,
program_name=program,
) )
multi[program] = cmd multi[program] = cmd
commands[command] = cmd commands[command] = cmd
@ -166,7 +175,9 @@ class HonAppliance:
def parameters(self): def parameters(self):
result = {} result = {}
for name, command in self._commands.items(): for name, command in self._commands.items():
for key, parameter in command.parameters.items(): for key, parameter in (
command.parameters | command.ancillary_parameters
).items():
result.setdefault(name, {})[key] = parameter.value result.setdefault(name, {})[key] = parameter.value
return result return result

9
pyhon/appliances/dw.py Normal file
View File

@ -0,0 +1,9 @@
class Appliance:
def data(self, data):
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED":
data["attributes"]["parameters"]["machMode"] = "0"
data["active"] = bool(data.get("attributes", {}).get("activity"))
return data
def settings(self, settings):
return settings

View File

@ -1,23 +0,0 @@
from pyhon.parameter import HonParameterEnum
class Appliance:
_FILTERS = {
"default": "^(?!iot_(?:recipe|guided))\\S+$",
"recipe": "iot_recipe_",
"guided": "iot_guided_",
}
def __init__(self):
filters = list(self._FILTERS.values())
data = {"defaultValue": filters[0], "enumValues": filters}
self._program_filter = HonParameterEnum("program_filter", data)
def data(self, data):
return data
def settings(self, settings):
settings["program_filter"] = self._program_filter
value = self._FILTERS[self._program_filter.value]
settings["startProgram.program"].filter = value
return settings

View File

@ -1,32 +1,47 @@
from pyhon.parameter import ( from typing import Optional, Dict, Any, List, TYPE_CHECKING
HonParameterFixed,
HonParameterEnum, from pyhon.parameter.base import HonParameter
HonParameterRange, from pyhon.parameter.enum import HonParameterEnum
HonParameterProgram, from pyhon.parameter.fixed import HonParameterFixed
) from pyhon.parameter.program import HonParameterProgram
from pyhon.parameter.range import HonParameterRange
if TYPE_CHECKING:
from pyhon import HonAPI
from pyhon.appliance import HonAppliance
class HonCommand: class HonCommand:
def __init__(self, name, attributes, connector, device, multi=None, program=""): def __init__(
self._connector = connector self,
self._device = device name: str,
self._name = name attributes: Dict[str, Any],
self._multi = multi or {} api: "HonAPI",
self._program = program appliance: "HonAppliance",
self._description = attributes.get("description", "") programs: Optional[Dict[str, "HonCommand"]] = None,
self._parameters = self._create_parameters(attributes.get("parameters", {})) program_name: str = "",
self._ancillary_parameters = self._create_parameters( ):
self._api: HonAPI = api
self._appliance: "HonAppliance" = appliance
self._name: str = name
self._programs: Optional[Dict[str, "HonCommand"]] = programs or {}
self._program_name: str = program_name
self._description: str = attributes.get("description", "")
self._parameters: Dict[str, HonParameter] = self._create_parameters(
attributes.get("parameters", {})
)
self._ancillary_parameters: Dict[str, HonParameter] = self._create_parameters(
attributes.get("ancillaryParameters", {}) attributes.get("ancillaryParameters", {})
) )
def __repr__(self): def __repr__(self) -> str:
return f"{self._name} command" return f"{self._name} command"
def _create_parameters(self, parameters): def _create_parameters(self, parameters: Dict) -> Dict[str, HonParameter]:
result = {} result: Dict[str, HonParameter] = {}
for parameter, attributes in parameters.items(): for parameter, attributes in parameters.items():
if parameter == "zoneMap" and self._device.zone: if parameter == "zoneMap" and self._appliance.zone:
attributes["default"] = self._device.zone attributes["default"] = self._appliance.zone
match attributes.get("typology"): match attributes.get("typology"):
case "range": case "range":
result[parameter] = HonParameterRange(parameter, attributes) result[parameter] = HonParameterRange(parameter, attributes)
@ -34,39 +49,46 @@ class HonCommand:
result[parameter] = HonParameterEnum(parameter, attributes) result[parameter] = HonParameterEnum(parameter, attributes)
case "fixed": case "fixed":
result[parameter] = HonParameterFixed(parameter, attributes) result[parameter] = HonParameterFixed(parameter, attributes)
if self._multi: if self._programs:
result["program"] = HonParameterProgram("program", self) result["program"] = HonParameterProgram("program", self)
return result return result
@property @property
def parameters(self): def parameters(self) -> Dict[str, HonParameter]:
return self._parameters return self._parameters
@property @property
def ancillary_parameters(self): def ancillary_parameters(self) -> Dict[str, HonParameter]:
return { return self._ancillary_parameters
key: parameter.value
for key, parameter in self._ancillary_parameters.items()
}
async def send(self): async def send(self) -> bool:
parameters = { params = {k: v.value for k, v in self._parameters.items()}
name: parameter.value for name, parameter in self._parameters.items() ancillary_params = {k: v.value for k, v in self._ancillary_parameters.items()}
} return await self._api.send_command(
return await self._connector.send_command( self._appliance, self._name, params, ancillary_params
self._device, self._name, parameters, self.ancillary_parameters
) )
def get_programs(self): @property
return self._multi def programs(self) -> Dict[str, "HonCommand"]:
if self._programs is None:
return {}
return self._programs
def set_program(self, program): @property
self._device.commands[self._name] = self._multi[program] def program(self) -> str:
return self._program_name
def _get_settings_keys(self, command=None): @program.setter
command = command or self def program(self, program: str) -> None:
self._appliance.commands[self._name] = self.programs[program]
def _get_settings_keys(self, command: Optional["HonCommand"] = None) -> List[str]:
if command is None:
command = self
keys = [] keys = []
for key, parameter in command._parameters.items(): for key, parameter in (
command._parameters | command._ancillary_parameters
).items():
if isinstance(parameter, HonParameterFixed): if isinstance(parameter, HonParameterFixed):
continue continue
if key not in keys: if key not in keys:
@ -74,19 +96,22 @@ class HonCommand:
return keys return keys
@property @property
def setting_keys(self): def setting_keys(self) -> List[str]:
if not self._multi: if not self._programs:
return self._get_settings_keys() return self._get_settings_keys()
result = [ result = [
key for cmd in self._multi.values() for key in self._get_settings_keys(cmd) key
for cmd in self._programs.values()
for key in self._get_settings_keys(cmd)
] ]
return list(set(result + ["program"])) return list(set(result + ["program"]))
@property @property
def settings(self): def settings(self) -> Dict[str, HonParameter]:
"""Parameters with typology enum and range""" """Parameters with typology enum and range"""
return { return {
s: self._parameters.get(s) s: param
for s in self.setting_keys for s in self.setting_keys
if self._parameters.get(s) is not None if (param := self._parameters.get(s)) is not None
or (param := self._ancillary_parameters.get(s)) is not None
} }

View File

@ -2,15 +2,15 @@ import json
import logging import logging
from datetime import datetime from datetime import datetime
from typing import Dict, Optional from typing import Dict, Optional
from typing_extensions import Self
from aiohttp import ClientSession from aiohttp import ClientSession
from typing_extensions import Self
from pyhon import const, exceptions from pyhon import const, exceptions
from pyhon.appliance import HonAppliance from pyhon.appliance import HonAppliance
from pyhon.connection.auth import HonAuth from pyhon.connection.auth import HonAuth
from pyhon.connection.handler.hon import HonConnectionHandler
from pyhon.connection.handler.anonym import HonAnonymousConnectionHandler from pyhon.connection.handler.anonym import HonAnonymousConnectionHandler
from pyhon.connection.handler.hon import HonConnectionHandler
_LOGGER = logging.getLogger() _LOGGER = logging.getLogger()

View File

@ -6,8 +6,7 @@ import urllib
from contextlib import suppress from contextlib import suppress
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timedelta from datetime import datetime, timedelta
from pprint import pformat from typing import Dict, Optional
from typing import Dict, Optional, List
from urllib import parse from urllib import parse
from urllib.parse import quote from urllib.parse import quote

View File

@ -1,7 +1,7 @@
import logging import logging
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Optional, Callable, Dict, Any from typing import Optional, Callable, Dict
import aiohttp import aiohttp
from typing_extensions import Self from typing_extensions import Self

View File

@ -49,7 +49,7 @@ def create_command(commands, concat=False):
for name, command in commands.items(): for name, command in commands.items():
if not concat: if not concat:
result[name] = {} result[name] = {}
for parameter, data in command.parameters.items(): for parameter, data in command.settings.items():
if data.typology == "enum": if data.typology == "enum":
value = data.values value = data.values
elif data.typology == "range": elif data.typology == "range":

View File

@ -1,9 +1,9 @@
import asyncio import asyncio
import copy from types import TracebackType
from typing import List, Optional, Dict, Any from typing import List, Optional, Dict, Any, Type
from typing_extensions import Self
from aiohttp import ClientSession from aiohttp import ClientSession
from typing_extensions import Self
from pyhon import HonAPI, exceptions from pyhon import HonAPI, exceptions
from pyhon.appliance import HonAppliance from pyhon.appliance import HonAppliance
@ -17,10 +17,15 @@ class Hon:
self._appliances: List[HonAppliance] = [] self._appliances: List[HonAppliance] = []
self._api: Optional[HonAPI] = None self._api: Optional[HonAPI] = None
async def __aenter__(self): async def __aenter__(self) -> Self:
return await self.create() return await self.create()
async def __aexit__(self, exc_type, exc_val, exc_tb): async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
await self.close() await self.close()
@property @property
@ -53,12 +58,13 @@ class Hon:
) )
self._appliances.append(appliance) self._appliances.append(appliance)
async def setup(self): async def setup(self) -> None:
appliance: Dict appliance: Dict
for appliance in (await self._api.load_appliances())["payload"]["appliances"]: for appliance in (await self.api.load_appliances())["payload"]["appliances"]:
for zone in range(int(appliance.get("zone", "0"))): if (zones := int(appliance.get("zone", "0"))) > 1:
await self._create_appliance(appliance.copy(), zone=zone + 1) for zone in range(zones):
await self._create_appliance(appliance.copy(), zone=zone + 1)
await self._create_appliance(appliance) await self._create_appliance(appliance)
async def close(self): async def close(self) -> None:
await self._api.close() await self.api.close()

View File

@ -1,157 +0,0 @@
import re
def str_to_float(string):
try:
return int(string)
except ValueError:
return float(str(string).replace(",", "."))
class HonParameter:
def __init__(self, key, attributes):
self._key = key
self._category = attributes.get("category")
self._typology = attributes.get("typology")
self._mandatory = attributes.get("mandatory")
self._value = ""
@property
def key(self):
return self._key
@property
def value(self):
return self._value if self._value is not None else "0"
@property
def category(self):
return self._category
@property
def typology(self):
return self._typology
@property
def mandatory(self):
return self._mandatory
class HonParameterFixed(HonParameter):
def __init__(self, key, attributes):
super().__init__(key, attributes)
self._value = attributes.get("fixedValue", None)
def __repr__(self):
return f"{self.__class__} (<{self.key}> fixed)"
@property
def value(self):
return self._value if self._value is not None else "0"
@value.setter
def value(self, value):
if not value == self._value:
raise ValueError("Can't change fixed value")
class HonParameterRange(HonParameter):
def __init__(self, key, attributes):
super().__init__(key, attributes)
self._min = str_to_float(attributes["minimumValue"])
self._max = str_to_float(attributes["maximumValue"])
self._step = str_to_float(attributes["incrementValue"])
self._default = str_to_float(attributes.get("defaultValue", self._min))
self._value = self._default
def __repr__(self):
return f"{self.__class__} (<{self.key}> [{self._min} - {self._max}])"
@property
def min(self):
return self._min
@property
def max(self):
return self._max
@property
def step(self):
return self._step
@property
def value(self):
return self._value if self._value is not None else self._min
@value.setter
def value(self, value):
value = str_to_float(value)
if self._min <= value <= self._max and not value % self._step:
self._value = value
else:
raise ValueError(
f"Allowed: min {self._min} max {self._max} step {self._step}"
)
class HonParameterEnum(HonParameter):
def __init__(self, key, attributes):
super().__init__(key, attributes)
self._default = attributes.get("defaultValue")
self._value = self._default or "0"
self._values = attributes.get("enumValues")
def __repr__(self):
return f"{self.__class__} (<{self.key}> {self.values})"
@property
def values(self):
return [str(value) for value in self._values]
@property
def value(self):
return self._value if self._value is not None else self.values[0]
@value.setter
def value(self, value):
if value in self.values:
self._value = value
else:
raise ValueError(f"Allowed values {self._value}")
class HonParameterProgram(HonParameterEnum):
def __init__(self, key, command):
super().__init__(key, {})
self._command = command
self._value = command._program
self._values = command._multi
self._typology = "enum"
self._filter = ""
@property
def value(self):
return self._value
@value.setter
def value(self, value):
if value in self.values:
self._command.set_program(value)
else:
raise ValueError(f"Allowed values {self._values}")
@property
def filter(self):
return self._filter
@filter.setter
def filter(self, filter):
self._filter = filter
@property
def values(self):
values = []
for value in self._values:
if not self._filter or re.findall(self._filter, str(value)):
values.append(str(value))
return sorted(values)

View File

30
pyhon/parameter/base.py Normal file
View File

@ -0,0 +1,30 @@
from typing import Dict, Any
class HonParameter:
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
self._key = key
self._category: str = attributes.get("category", "")
self._typology: str = attributes.get("typology", "")
self._mandatory: int = attributes.get("mandatory", 0)
self._value: str | float = ""
@property
def key(self) -> str:
return self._key
@property
def value(self) -> str | float:
return self._value if self._value is not None else "0"
@property
def category(self) -> str:
return self._category
@property
def typology(self) -> str:
return self._typology
@property
def mandatory(self) -> int:
return self._mandatory

29
pyhon/parameter/enum.py Normal file
View File

@ -0,0 +1,29 @@
from typing import Dict, Any, List
from pyhon.parameter.base import HonParameter
class HonParameterEnum(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
super().__init__(key, attributes)
self._default = attributes.get("defaultValue")
self._value = self._default or "0"
self._values: List[str] = attributes.get("enumValues", [])
def __repr__(self) -> str:
return f"{self.__class__} (<{self.key}> {self.values})"
@property
def values(self) -> List[str]:
return [str(value) for value in self._values]
@property
def value(self) -> str | float:
return self._value if self._value is not None else self.values[0]
@value.setter
def value(self, value: str) -> None:
if value in self.values:
self._value = value
else:
raise ValueError(f"Allowed values {self._value}")

21
pyhon/parameter/fixed.py Normal file
View File

@ -0,0 +1,21 @@
from typing import Dict, Any
from pyhon.parameter.base import HonParameter
class HonParameterFixed(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
super().__init__(key, attributes)
self._value = attributes.get("fixedValue", None)
def __repr__(self) -> str:
return f"{self.__class__} (<{self.key}> fixed)"
@property
def value(self) -> str | float:
return self._value if self._value is not None else "0"
@value.setter
def value(self, value: str | float) -> None:
# Fixed values seems being not so fixed as thought
self._value = value

View File

@ -0,0 +1,33 @@
from typing import List, TYPE_CHECKING
from pyhon.parameter.enum import HonParameterEnum
if TYPE_CHECKING:
from pyhon.commands import HonCommand
class HonParameterProgram(HonParameterEnum):
_FILTER = ["iot_recipe", "iot_guided"]
def __init__(self, key: str, command: "HonCommand") -> None:
super().__init__(key, {})
self._command = command
self._value: str = command.program
self._values: List[str] = list(command.programs)
self._typology: str = "enum"
@property
def value(self) -> str | float:
return self._value
@value.setter
def value(self, value: str) -> None:
if value in self.values:
self._command.program = value
else:
raise ValueError(f"Allowed values {self._values}")
@property
def values(self) -> List[str]:
values = [v for v in self._values if all(f not in v for f in self._FILTER)]
return sorted(values)

49
pyhon/parameter/range.py Normal file
View File

@ -0,0 +1,49 @@
from typing import Dict, Any
from pyhon.parameter.base import HonParameter
def str_to_float(string: str | float) -> float:
try:
return int(string)
except ValueError:
return float(str(string).replace(",", "."))
class HonParameterRange(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
super().__init__(key, attributes)
self._min: float = str_to_float(attributes["minimumValue"])
self._max: float = str_to_float(attributes["maximumValue"])
self._step: float = str_to_float(attributes["incrementValue"])
self._default: float = str_to_float(attributes.get("defaultValue", self._min))
self._value: float = self._default
def __repr__(self):
return f"{self.__class__} (<{self.key}> [{self._min} - {self._max}])"
@property
def min(self) -> float:
return self._min
@property
def max(self) -> float:
return self._max
@property
def step(self) -> float:
return self._step
@property
def value(self) -> float:
return self._value if self._value is not None else self._min
@value.setter
def value(self, value: float) -> None:
value = str_to_float(value)
if self._min <= value <= self._max and not value % self._step:
self._value = value
else:
raise ValueError(
f"Allowed: min {self._min} max {self._max} step {self._step}"
)

View File

@ -1 +1,2 @@
aiohttp aiohttp==3.8.4
yarl==1.8.2

4
requirements_dev.txt Normal file
View File

@ -0,0 +1,4 @@
black==23.3.0
flake8==6.0.0
mypy==1.2.0
pylint==2.17.2

View File

@ -7,7 +7,7 @@ with open("README.md", "r") as f:
setup( setup(
name="pyhOn", name="pyhOn",
version="0.8.0b3", version="0.8.0b6",
author="Andre Basche", author="Andre Basche",
description="Control hOn devices with python", description="Control hOn devices with python",
long_description=long_description, long_description=long_description,