Compare commits
28 Commits
Author | SHA1 | Date | |
---|---|---|---|
b011d98e07 | |||
ad864286fc | |||
13cff8caa0 | |||
5fc6245806 | |||
1dad0e14b8 | |||
b04c601ad6 | |||
3ec0f5a006 | |||
78bc85132f | |||
191928287f | |||
c0aab8b99d | |||
b37715d0ca | |||
411effd814 | |||
04f19c4609 | |||
a68dcac379 | |||
69be63df2a | |||
6c44aa895d | |||
8372c75e30 | |||
40cc0013a5 | |||
f2aa3dc8fd | |||
e887371bec | |||
5b91747ec1 | |||
8da2018302 | |||
03187745bf | |||
461a247ad3 | |||
834f25a639 | |||
46ff9be4a2 | |||
a1618bb18a | |||
18b0ecdd68 |
2
.github/workflows/python-check.yml
vendored
2
.github/workflows/python-check.yml
vendored
@ -25,7 +25,7 @@ jobs:
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
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
|
||||
run: |
|
||||
# stop the build if there are Python syntax errors or undefined names
|
||||
|
1
.gitignore
vendored
1
.gitignore
vendored
@ -4,3 +4,4 @@ __pycache__/
|
||||
dist/
|
||||
**/*.egg-info/
|
||||
test*
|
||||
build/
|
||||
|
@ -6,7 +6,7 @@
|
||||
[](https://www.python.org/)
|
||||
[](https://github.com/Andre0512/pyhOn/blob/main/LICENSE)
|
||||
[](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.
|
||||
|
||||
## Installation
|
||||
|
0
pyhon/__main__.py
Executable file → Normal file
0
pyhon/__main__.py
Executable file → Normal file
@ -5,7 +5,7 @@ from typing import TYPE_CHECKING
|
||||
|
||||
from pyhon import helper
|
||||
from pyhon.commands import HonCommand
|
||||
from pyhon.parameter import HonParameterFixed
|
||||
from pyhon.parameter.fixed import HonParameterFixed
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pyhon import HonAPI
|
||||
@ -24,7 +24,7 @@ class HonAppliance:
|
||||
self._commands: Dict = {}
|
||||
self._statistics: Dict = {}
|
||||
self._attributes: Dict = {}
|
||||
self._zone = zone
|
||||
self._zone: int = zone
|
||||
|
||||
try:
|
||||
self._extra = importlib.import_module(
|
||||
@ -106,6 +106,10 @@ class HonAppliance:
|
||||
def info(self):
|
||||
return self._info
|
||||
|
||||
@property
|
||||
def zone(self) -> int:
|
||||
return self._zone
|
||||
|
||||
async def _recover_last_command_states(self, commands):
|
||||
command_history = await self._api.command_history(self)
|
||||
for name, command in commands.items():
|
||||
@ -120,8 +124,8 @@ class HonAppliance:
|
||||
if last is None:
|
||||
continue
|
||||
parameters = command_history[last].get("command", {}).get("parameters", {})
|
||||
if command._multi and parameters.get("program"):
|
||||
command.set_program(parameters.pop("program").split(".")[-1].lower())
|
||||
if command.programs and parameters.get("program"):
|
||||
command.program = parameters.pop("program").split(".")[-1].lower()
|
||||
command = self.commands[name]
|
||||
for key, data in command.settings.items():
|
||||
if (
|
||||
@ -145,7 +149,12 @@ class HonAppliance:
|
||||
for program, attr2 in attr.items():
|
||||
program = program.split(".")[-1].lower()
|
||||
cmd = HonCommand(
|
||||
command, attr2, self._api, self, multi=multi, program=program
|
||||
command,
|
||||
attr2,
|
||||
self._api,
|
||||
self,
|
||||
programs=multi,
|
||||
program_name=program,
|
||||
)
|
||||
multi[program] = cmd
|
||||
commands[command] = cmd
|
||||
@ -166,7 +175,9 @@ class HonAppliance:
|
||||
def parameters(self):
|
||||
result = {}
|
||||
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
|
||||
return result
|
||||
|
||||
|
9
pyhon/appliances/dw.py
Normal file
9
pyhon/appliances/dw.py
Normal 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
|
@ -1,23 +1,14 @@
|
||||
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):
|
||||
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED":
|
||||
data["attributes"]["parameters"]["temp"] = "0"
|
||||
data["attributes"]["parameters"]["onOffStatus"] = "0"
|
||||
data["attributes"]["parameters"]["remoteCtrValid"] = "0"
|
||||
data["attributes"]["parameters"]["remainingTimeMM"] = "0"
|
||||
|
||||
data["active"] = data["attributes"]["parameters"]["onOffStatus"] == "1"
|
||||
|
||||
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
|
||||
|
@ -1,3 +1,6 @@
|
||||
from pyhon.parameter.fixed import HonParameterFixed
|
||||
|
||||
|
||||
class Appliance:
|
||||
def data(self, data):
|
||||
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED":
|
||||
@ -7,4 +10,7 @@ class Appliance:
|
||||
return data
|
||||
|
||||
def settings(self, settings):
|
||||
dry_level = settings["startProgram.dryLevel"]
|
||||
if isinstance(dry_level, HonParameterFixed) and dry_level.value == "11":
|
||||
settings.pop("startProgram.dryLevel", None)
|
||||
return settings
|
||||
|
@ -1,32 +1,47 @@
|
||||
from pyhon.parameter import (
|
||||
HonParameterFixed,
|
||||
HonParameterEnum,
|
||||
HonParameterRange,
|
||||
HonParameterProgram,
|
||||
)
|
||||
from typing import Optional, Dict, Any, List, TYPE_CHECKING
|
||||
|
||||
from pyhon.parameter.base import HonParameter
|
||||
from pyhon.parameter.enum import HonParameterEnum
|
||||
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:
|
||||
def __init__(self, name, attributes, connector, device, multi=None, program=""):
|
||||
self._connector = connector
|
||||
self._device = device
|
||||
self._name = name
|
||||
self._multi = multi or {}
|
||||
self._program = program
|
||||
self._description = attributes.get("description", "")
|
||||
self._parameters = self._create_parameters(attributes.get("parameters", {}))
|
||||
self._ancillary_parameters = self._create_parameters(
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
attributes: Dict[str, Any],
|
||||
api: "HonAPI",
|
||||
appliance: "HonAppliance",
|
||||
programs: Optional[Dict[str, "HonCommand"]] = None,
|
||||
program_name: str = "",
|
||||
):
|
||||
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", {})
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
def __repr__(self) -> str:
|
||||
return f"{self._name} command"
|
||||
|
||||
def _create_parameters(self, parameters):
|
||||
result = {}
|
||||
def _create_parameters(self, parameters: Dict) -> Dict[str, HonParameter]:
|
||||
result: Dict[str, HonParameter] = {}
|
||||
for parameter, attributes in parameters.items():
|
||||
if parameter == "zoneMap" and self._device.zone:
|
||||
attributes["default"] = self._device.zone
|
||||
if parameter == "zoneMap" and self._appliance.zone:
|
||||
attributes["default"] = self._appliance.zone
|
||||
match attributes.get("typology"):
|
||||
case "range":
|
||||
result[parameter] = HonParameterRange(parameter, attributes)
|
||||
@ -34,59 +49,66 @@ class HonCommand:
|
||||
result[parameter] = HonParameterEnum(parameter, attributes)
|
||||
case "fixed":
|
||||
result[parameter] = HonParameterFixed(parameter, attributes)
|
||||
if self._multi:
|
||||
if self._programs:
|
||||
result["program"] = HonParameterProgram("program", self)
|
||||
return result
|
||||
|
||||
@property
|
||||
def parameters(self):
|
||||
def parameters(self) -> Dict[str, HonParameter]:
|
||||
return self._parameters
|
||||
|
||||
@property
|
||||
def ancillary_parameters(self):
|
||||
return {
|
||||
key: parameter.value
|
||||
for key, parameter in self._ancillary_parameters.items()
|
||||
}
|
||||
def ancillary_parameters(self) -> Dict[str, HonParameter]:
|
||||
return self._ancillary_parameters
|
||||
|
||||
async def send(self):
|
||||
parameters = {
|
||||
name: parameter.value for name, parameter in self._parameters.items()
|
||||
}
|
||||
return await self._connector.send_command(
|
||||
self._device, self._name, parameters, self.ancillary_parameters
|
||||
async def send(self) -> bool:
|
||||
params = {k: v.value for k, v in self._parameters.items()}
|
||||
ancillary_params = {k: v.value for k, v in self._ancillary_parameters.items()}
|
||||
return await self._api.send_command(
|
||||
self._appliance, self._name, params, ancillary_params
|
||||
)
|
||||
|
||||
def get_programs(self):
|
||||
return self._multi
|
||||
@property
|
||||
def programs(self) -> Dict[str, "HonCommand"]:
|
||||
if self._programs is None:
|
||||
return {}
|
||||
return self._programs
|
||||
|
||||
def set_program(self, program):
|
||||
self._device.commands[self._name] = self._multi[program]
|
||||
@property
|
||||
def program(self) -> str:
|
||||
return self._program_name
|
||||
|
||||
def _get_settings_keys(self, command=None):
|
||||
command = command or self
|
||||
@program.setter
|
||||
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 = []
|
||||
for key, parameter in command._parameters.items():
|
||||
if isinstance(parameter, HonParameterFixed):
|
||||
continue
|
||||
for key, parameter in (
|
||||
command._parameters | command._ancillary_parameters
|
||||
).items():
|
||||
if key not in keys:
|
||||
keys.append(key)
|
||||
return keys
|
||||
|
||||
@property
|
||||
def setting_keys(self):
|
||||
if not self._multi:
|
||||
def setting_keys(self) -> List[str]:
|
||||
if not self._programs:
|
||||
return self._get_settings_keys()
|
||||
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"]))
|
||||
|
||||
@property
|
||||
def settings(self):
|
||||
"""Parameters with typology enum and range"""
|
||||
def settings(self) -> Dict[str, HonParameter]:
|
||||
return {
|
||||
s: self._parameters.get(s)
|
||||
s: param
|
||||
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
|
||||
}
|
||||
|
@ -2,15 +2,15 @@ import json
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Dict, Optional
|
||||
from typing_extensions import Self
|
||||
|
||||
from aiohttp import ClientSession
|
||||
from typing_extensions import Self
|
||||
|
||||
from pyhon import const, exceptions
|
||||
from pyhon.appliance import HonAppliance
|
||||
from pyhon.connection.auth import HonAuth
|
||||
from pyhon.connection.handler.hon import HonConnectionHandler
|
||||
from pyhon.connection.handler.anonym import HonAnonymousConnectionHandler
|
||||
from pyhon.connection.handler.hon import HonConnectionHandler
|
||||
|
||||
_LOGGER = logging.getLogger()
|
||||
|
||||
@ -74,13 +74,14 @@ class HonAPI:
|
||||
"applianceType": appliance.appliance_type,
|
||||
"code": appliance.info["code"],
|
||||
"applianceModelId": appliance.appliance_model_id,
|
||||
"firmwareId": appliance.info["eepromId"],
|
||||
"macAddress": appliance.mac_address,
|
||||
"fwVersion": appliance.info["fwVersion"],
|
||||
"os": const.OS,
|
||||
"appVersion": const.APP_VERSION,
|
||||
"series": appliance.info["series"],
|
||||
}
|
||||
if firmware_id := appliance.info.get("eepromId"):
|
||||
params["firmwareId"] = firmware_id
|
||||
url: str = f"{const.API_URL}/commands/v1/retrieve"
|
||||
async with self._hon.get(url, params=params) as response:
|
||||
result: Dict = (await response.json()).get("payload", {})
|
||||
|
@ -6,8 +6,7 @@ import urllib
|
||||
from contextlib import suppress
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta
|
||||
from pprint import pformat
|
||||
from typing import Dict, Optional, List
|
||||
from typing import Dict, Optional
|
||||
from urllib import parse
|
||||
from urllib.parse import quote
|
||||
|
||||
@ -229,7 +228,10 @@ class HonAuth:
|
||||
except json.JSONDecodeError:
|
||||
await self._error_logger(response)
|
||||
return False
|
||||
self._cognito_token = json_data["cognitoUser"]["Token"]
|
||||
self._cognito_token = json_data.get("cognitoUser", {}).get("Token", "")
|
||||
if not self._cognito_token:
|
||||
_LOGGER.error(json_data)
|
||||
raise exceptions.HonAuthenticationError()
|
||||
return True
|
||||
|
||||
async def authenticate(self) -> None:
|
||||
|
@ -1,7 +1,7 @@
|
||||
import logging
|
||||
from collections.abc import AsyncIterator
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import Optional, Callable, Dict, Any
|
||||
from typing import Optional, Callable, Dict
|
||||
|
||||
import aiohttp
|
||||
from typing_extensions import Self
|
||||
|
@ -49,7 +49,7 @@ def create_command(commands, concat=False):
|
||||
for name, command in commands.items():
|
||||
if not concat:
|
||||
result[name] = {}
|
||||
for parameter, data in command.parameters.items():
|
||||
for parameter, data in command.settings.items():
|
||||
if data.typology == "enum":
|
||||
value = data.values
|
||||
elif data.typology == "range":
|
||||
|
28
pyhon/hon.py
28
pyhon/hon.py
@ -1,9 +1,9 @@
|
||||
import asyncio
|
||||
import copy
|
||||
from typing import List, Optional, Dict, Any
|
||||
from typing_extensions import Self
|
||||
from types import TracebackType
|
||||
from typing import List, Optional, Dict, Any, Type
|
||||
|
||||
from aiohttp import ClientSession
|
||||
from typing_extensions import Self
|
||||
|
||||
from pyhon import HonAPI, exceptions
|
||||
from pyhon.appliance import HonAppliance
|
||||
@ -17,10 +17,15 @@ class Hon:
|
||||
self._appliances: List[HonAppliance] = []
|
||||
self._api: Optional[HonAPI] = None
|
||||
|
||||
async def __aenter__(self):
|
||||
async def __aenter__(self) -> Self:
|
||||
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()
|
||||
|
||||
@property
|
||||
@ -53,12 +58,13 @@ class Hon:
|
||||
)
|
||||
self._appliances.append(appliance)
|
||||
|
||||
async def setup(self):
|
||||
async def setup(self) -> None:
|
||||
appliance: Dict
|
||||
for appliance in (await self._api.load_appliances())["payload"]["appliances"]:
|
||||
for zone in range(int(appliance.get("zone", "0"))):
|
||||
await self._create_appliance(appliance.copy(), zone=zone + 1)
|
||||
for appliance in (await self.api.load_appliances())["payload"]["appliances"]:
|
||||
if (zones := int(appliance.get("zone", "0"))) > 1:
|
||||
for zone in range(zones):
|
||||
await self._create_appliance(appliance.copy(), zone=zone + 1)
|
||||
await self._create_appliance(appliance)
|
||||
|
||||
async def close(self):
|
||||
await self._api.close()
|
||||
async def close(self) -> None:
|
||||
await self.api.close()
|
||||
|
@ -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)
|
0
pyhon/parameter/__init__.py
Normal file
0
pyhon/parameter/__init__.py
Normal file
30
pyhon/parameter/base.py
Normal file
30
pyhon/parameter/base.py
Normal 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
29
pyhon/parameter/enum.py
Normal 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}")
|
25
pyhon/parameter/fixed.py
Normal file
25
pyhon/parameter/fixed.py
Normal file
@ -0,0 +1,25 @@
|
||||
from typing import Dict, Any, List
|
||||
|
||||
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
|
||||
|
||||
@property
|
||||
def values(self) -> List[str]:
|
||||
return list(str(self.value))
|
33
pyhon/parameter/program.py
Normal file
33
pyhon/parameter/program.py
Normal file
@ -0,0 +1,33 @@
|
||||
from typing import List, TYPE_CHECKING, Dict
|
||||
|
||||
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._programs: Dict[str, "HonCommand"] = 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._programs if all(f not in v for f in self._FILTER)]
|
||||
return sorted(values)
|
53
pyhon/parameter/range.py
Normal file
53
pyhon/parameter/range.py
Normal file
@ -0,0 +1,53 @@
|
||||
from typing import Dict, Any, List
|
||||
|
||||
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}"
|
||||
)
|
||||
|
||||
@property
|
||||
def values(self) -> List[str]:
|
||||
return [str(i) for i in range(int(self.min), int(self.max) + 1, int(self.step))]
|
@ -1 +1,2 @@
|
||||
aiohttp
|
||||
aiohttp==3.8.4
|
||||
yarl==1.8.2
|
||||
|
4
requirements_dev.txt
Normal file
4
requirements_dev.txt
Normal file
@ -0,0 +1,4 @@
|
||||
black==23.3.0
|
||||
flake8==6.0.0
|
||||
mypy==1.2.0
|
||||
pylint==2.17.2
|
Reference in New Issue
Block a user