Compare commits

...

40 Commits

Author SHA1 Message Date
Andre Basche 6162cd3f98 Bump version 2024-02-11 05:11:21 +01:00
Andre Basche e225a850bc Fix checks 2024-02-11 05:11:21 +01:00
Andre Basche c4e8038652 Fix problems with refresh token 2024-02-11 05:11:21 +01:00
Andre Basche 767cbe35de Set refresh token 2024-02-11 05:11:21 +01:00
Andre Basche df27729c2e Change mobile id 2024-02-11 05:11:21 +01:00
Andre Basche a5595f47bf Fix another issue with data path 2024-02-11 05:11:21 +01:00
Andre Basche 29c936b045 Fix path for #22 2024-02-11 05:11:21 +01:00
Andre Basche 2a74000ff3 Update requirements 2024-02-11 05:11:21 +01:00
Andre Basche 8a46bd4b53 Fix for float steps #22 2024-02-11 05:11:21 +01:00
Andre Basche 3acd12f345 Update README.md 2024-02-02 23:49:49 +01:00
Andre Basche 327d4a1814 Update README.md 2024-01-15 10:36:18 +01:00
Andre Basche 4dc60c29ee Fix crash in loading attributs Andre0512/hon#134 2024-01-11 01:21:38 +01:00
Andre Basche 7bc9e718a0 Merge pull request #17 from MiguelAngelLV/main
Use class attribute to check active
2023-11-26 13:43:44 +01:00
Mguel Ángel d4e5793186 Fix check active oven 2023-11-26 13:36:41 +01:00
Andre Basche e9f2bb9f4f Send program name Andre0512/hon#124 2023-11-21 02:23:53 +01:00
Andre Basche ea81e2891f Bump version 2023-11-20 17:51:32 +01:00
Andre Basche c71f8f17f5 Fix error in wh 2023-11-20 17:50:28 +01:00
Andre Basche 27d974abba Fix dependencies 2023-11-20 16:38:39 +01:00
Andre Basche ab18e45f97 Add python3.12 support 2023-11-19 23:53:21 +01:00
Andre Basche e44b9b6773 Update mypy 2023-10-12 22:26:12 +02:00
Andre Basche 5c650d722b Bump version 2023-10-12 22:16:02 +02:00
Vadym 6ae40481e3 Issue with sync_command (#16)
* Added water heater appliance. Added ability to send only mandatory parameters

* fixed build

* formatting

* cleanup

* cleanup

* reformatting

* Added ability to send specific parameters. Useful in case the command has many not mandatory parameters and you want to send only one/few

* cleanup

* Fixed code style

* sync_command - fixed typos, skip to sync(actually reset) parameters of different types. Improved WaterHeater appliance

* cleanup

* cleanup

* clean code style

* check if base parameter is mandatory

* Reverted back sync_command, send mandatory parameters beside with specified

---------

Co-authored-by: Vadym Melnychuk <vme@primexm.com>
2023-10-12 16:43:41 +02:00
Andre Basche ff8ae160bb Fix empty prStr 2023-10-06 01:29:38 +02:00
Andre Basche 658e80a8f4 Change dependencies to variable major version 2023-10-02 03:21:51 +02:00
Andre Basche ffba85bf0d Bump version 2023-10-02 01:38:54 +02:00
Andre Basche 10c8d961c4 Support new style rules hon#112 2023-10-02 01:38:40 +02:00
Andre Basche 61dd470588 Set versions of dependant packages to 'compatible releases' 2023-07-27 19:16:23 +02:00
Andre Basche 1ed81c2a77 Simplify get favorites 2023-07-24 02:33:45 +02:00
Andre Basche e4dc3cb1d0 Next try to add py.typed in package 2023-07-24 01:47:45 +02:00
Andre Basche 2523069ce9 Fix false name caused by chatgpt's wrong advice 2023-07-23 23:18:09 +02:00
Andre Basche eeb458cb1b Add py.typed into package 2023-07-23 22:54:46 +02:00
Andre Basche 2764700bc7 Bump version 2023-07-23 21:56:16 +02:00
Andre Basche e6c796e822 Improve type hints 2023-07-23 21:55:42 +02:00
Andre Basche 454f2d8916 Use equal mypy cnofig as home assistant 2023-07-22 12:39:50 +02:00
Andre Basche 59ca8b6caf Not loading favorite if base program renamed 2023-07-22 11:53:39 +02:00
Andre Basche 44c55c681d Update requirements 2023-07-20 23:55:40 +02:00
Andre Basche cfee10df5f Improve logging for test api 2023-07-20 23:52:46 +02:00
Andre Basche e0774677eb Add and apply some mypy rules 2023-07-20 23:52:07 +02:00
Andre Basche fc60d15e60 Fix error for fridge without quickmode 2023-07-19 23:55:37 +02:00
Andre Basche 8ef8c0405d Fix empty value in settings 2023-07-19 19:52:21 +02:00
26 changed files with 239 additions and 103 deletions
+2 -2
View File
@@ -13,7 +13,7 @@ jobs:
strategy:
fail-fast: false
matrix:
python-version: ["3.10", "3.11"]
python-version: ["3.10", "3.11", "3.12"]
steps:
- uses: actions/checkout@v3
@@ -34,7 +34,7 @@ jobs:
mypy pyhon/
- name: Analysing the code with pylint
run: |
pylint $(git ls-files '*.py')
pylint $(git ls-files '*.py')
- name: Check black style
run: |
black . --check
+1 -1
View File
@@ -4,6 +4,6 @@ disable=missing-docstring
[FORMAT]
max-args=6
max-args=7
max-attributes=8
max-line-length=88
+1
View File
@@ -0,0 +1 @@
include pyhon/py.typed
+23 -8
View File
@@ -1,9 +1,24 @@
[mypy]
check_untyped_defs = True
disallow_any_generics = True
disallow_untyped_defs = True
disallow_any_unimported = True
no_implicit_optional = True
warn_return_any = True
show_error_codes = True
warn_unused_ignores = True
check_untyped_defs = true
disallow_any_generics = true
disallow_any_unimported = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
disable_error_code = annotation-unchecked
enable_error_code = ignore-without-code, redundant-self, truthy-iterable
follow_imports = silent
local_partial_types = true
no_implicit_optional = true
no_implicit_reexport = true
show_error_codes = true
strict_concatenate = false
strict_equality = true
warn_incomplete_stub = true
warn_redundant_casts = true
warn_return_any = true
warn_unreachable = true
warn_unused_configs = true
warn_unused_ignores = true
+3 -6
View File
@@ -38,9 +38,7 @@ def get_arguments() -> Dict[str, Any]:
"translate", help="language (de, en, fr...)", metavar="LANGUAGE"
)
translation.add_argument("--json", help="print as json", action="store_true")
parser.add_argument(
"-i", "--import", help="import pyhon data", nargs="?", default=Path().cwd()
)
parser.add_argument("-i", "--import", help="import pyhon data", nargs="?")
return vars(parser.parse_args())
@@ -73,9 +71,8 @@ async def main() -> None:
if language := args.get("translate"):
await translate(language, json_output=args.get("json", ""))
return
async with Hon(
*get_login_data(args), test_data_path=Path(args.get("import", ""))
) as hon:
test_data_path = Path(path) if (path := args.get("import", "")) else None
async with Hon(*get_login_data(args), test_data_path=test_data_path) as hon:
for device in hon.appliances:
if args.get("export"):
anonymous = args.get("anonymous", False)
+40 -17
View File
@@ -3,7 +3,7 @@ import logging
import re
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, Dict, Any, TYPE_CHECKING, List
from typing import Optional, Dict, Any, TYPE_CHECKING, List, TypeVar, overload
from pyhon import diagnose, exceptions
from pyhon.appliances.base import ApplianceBase
@@ -20,6 +20,8 @@ if TYPE_CHECKING:
_LOGGER = logging.getLogger(__name__)
T = TypeVar("T")
# pylint: disable=too-many-public-methods,too-many-instance-attributes
class HonAppliance:
@@ -49,24 +51,33 @@ class HonAppliance:
except ModuleNotFoundError:
self._extra = None
def _get_nested_item(self, item: str) -> Any:
result: List[Any] | Dict[str, Any] = self.data
for key in item.split("."):
if all(k in "0123456789" for k in key) and isinstance(result, list):
result = result[int(key)]
elif isinstance(result, dict):
result = result[key]
return result
def __getitem__(self, item: str) -> Any:
if self._zone:
item += f"Z{self._zone}"
if "." in item:
result = self.data
for key in item.split("."):
if all(k in "0123456789" for k in key) and isinstance(result, list):
result = result[int(key)]
else:
result = result[key]
return result
return self._get_nested_item(item)
if item in self.data:
return self.data[item]
if item in self.attributes["parameters"]:
return self.attributes["parameters"][item].value
return self.info[item]
def get(self, item: str, default: Any = None) -> Any:
@overload
def get(self, item: str, default: None = None) -> Any: ...
@overload
def get(self, item: str, default: T) -> T: ...
def get(self, item: str, default: Optional[T] = None) -> Any:
try:
return self[item]
except (KeyError, IndexError):
@@ -81,15 +92,15 @@ class HonAppliance:
@property
def appliance_model_id(self) -> str:
return self._info.get("applianceModelId", "")
return str(self._info.get("applianceModelId", ""))
@property
def appliance_type(self) -> str:
return self._info.get("applianceTypeName", "")
return str(self._info.get("applianceTypeName", ""))
@property
def mac_address(self) -> str:
return self.info.get("macAddress", "")
return str(self.info.get("macAddress", ""))
@property
def unique_id(self) -> str:
@@ -125,11 +136,11 @@ class HonAppliance:
@property
def model_id(self) -> int:
return self._info.get("applianceModelId", 0)
return int(self._info.get("applianceModelId", 0))
@property
def options(self) -> Dict[str, Any]:
return self._appliance_model.get("options", {})
return dict(self._appliance_model.get("options", {}))
@property
def commands(self) -> Dict[str, HonCommand]:
@@ -250,7 +261,9 @@ class HonAppliance:
if not (command := self.commands.get(command_name)):
return
for key in command.setting_keys:
if (new := self.attributes.get("parameters", {}).get(key)) is None:
if (
new := self.attributes.get("parameters", {}).get(key)
) is None or new.value == "":
continue
setting = command.settings[key]
try:
@@ -262,7 +275,12 @@ class HonAppliance:
_LOGGER.info("Can't set %s - %s", key, error)
continue
def sync_command(self, main: str, target: Optional[List[str] | str] = None) -> None:
def sync_command(
self,
main: str,
target: Optional[List[str] | str] = None,
to_sync: Optional[List[str] | bool] = None,
) -> None:
base: Optional[HonCommand] = self.commands.get(main)
if not base:
return
@@ -272,7 +290,12 @@ class HonAppliance:
for name, target_param in data.parameters.items():
if not (base_param := base.parameters.get(name)):
return
continue
if to_sync and (
(isinstance(to_sync, list) and name not in to_sync)
or not base_param.mandatory
):
continue
self.sync_parameter(base_param, target_param)
def sync_parameter(self, main: Parameter, target: Parameter) -> None:
+5 -5
View File
@@ -7,10 +7,10 @@ class Appliance(ApplianceBase):
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data)
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
data["parameters"]["temp"].value = "0"
data["parameters"]["onOffStatus"].value = "0"
data["parameters"]["remoteCtrValid"].value = "0"
data["parameters"]["remainingTimeMM"].value = "0"
data["parameters"]["temp"].value = 0
data["parameters"]["onOffStatus"].value = 0
data["parameters"]["remoteCtrValid"].value = 0
data["parameters"]["remainingTimeMM"].value = 0
data["active"] = data["parameters"]["onOffStatus"] == "1"
data["active"] = data["parameters"]["onOffStatus"].value == 1
return data
+2 -2
View File
@@ -10,12 +10,12 @@ class Appliance(ApplianceBase):
data["modeZ1"] = "holiday"
elif data["parameters"]["intelligenceMode"] == "1":
data["modeZ1"] = "auto_set"
elif data["parameters"]["quickModeZ1"] == "1":
elif data["parameters"].get("quickModeZ1") == "1":
data["modeZ1"] = "super_cool"
else:
data["modeZ1"] = "no_mode"
if data["parameters"]["quickModeZ2"] == "1":
if data["parameters"].get("quickModeZ2") == "1":
data["modeZ2"] = "super_freeze"
elif data["parameters"]["intelligenceMode"] == "1":
data["modeZ2"] = "auto_set"
+7 -2
View File
@@ -1,11 +1,16 @@
from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase
from pyhon.parameter.base import HonParameter
class Appliance(ApplianceBase):
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data)
data["active"] = data["parameters"]["onOffStatus"] == "1"
parameter = data.get("parameters", {}).get("onOffStatus")
is_class = isinstance(parameter, HonParameter)
data["active"] = parameter.value == 1 if is_class else parameter == 1
return data
def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]:
return settings
+41 -19
View File
@@ -109,7 +109,7 @@ class HonCommandLoader:
categories: Optional[Dict[str, "HonCommand"]] = None,
category_name: str = "",
) -> Optional[HonCommand]:
"""Try to crate HonCommand object"""
"""Try to create HonCommand object"""
if not isinstance(data, dict):
self._additional_data[command_name] = data
return None
@@ -184,22 +184,44 @@ class HonCommandLoader:
def _add_favourites(self) -> None:
"""Patch program categories with favourites"""
for favourite in self._favourites:
name = favourite.get("favouriteName", {})
command = favourite.get("command", {})
command_name = command.get("commandName", "")
program_name = self._clean_name(command.get("programName", ""))
base: HonCommand = copy(
self.commands[command_name].categories[program_name]
)
for data in command.values():
if isinstance(data, str):
name, command_name, base = self._get_favourite_info(favourite)
if not base:
continue
base_command: HonCommand = copy(base)
self._update_base_command_with_data(base_command, favourite)
self._update_base_command_with_favourite(base_command)
self._update_program_categories(command_name, name, base_command)
def _get_favourite_info(
self, favourite: Dict[str, Any]
) -> tuple[str, str, HonCommand | None]:
name: str = favourite.get("favouriteName", {})
command = favourite.get("command", {})
command_name: str = command.get("commandName", "")
program_name = self._clean_name(command.get("programName", ""))
base_command = self.commands[command_name].categories.get(program_name)
return name, command_name, base_command
def _update_base_command_with_data(
self, base_command: HonCommand, command: Dict[str, Any]
) -> None:
for data in command.values():
if isinstance(data, str):
continue
for key, value in data.items():
if not (parameter := base_command.parameters.get(key)):
continue
for key, value in data.items():
if parameter := base.parameters.get(key):
with suppress(ValueError):
parameter.value = value
extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom")
base.parameters.update(favourite=extra_param)
if isinstance(program := base.parameters["program"], HonParameterProgram):
program.set_value(name)
self.commands[command_name].categories[name] = base
with suppress(ValueError):
parameter.value = value
def _update_base_command_with_favourite(self, base_command: HonCommand) -> None:
extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom")
base_command.parameters.update(favourite=extra_param)
def _update_program_categories(
self, command_name: str, name: str, base_command: HonCommand
) -> None:
program = base_command.parameters["program"]
if isinstance(program, HonParameterProgram):
program.set_value(name)
self.commands[command_name].categories[name] = base_command
+17 -6
View File
@@ -89,8 +89,11 @@ class HonCommand:
def parameter_value(self) -> Dict[str, Union[str, float]]:
return {n: p.value for n, p in self._parameters.items()}
def _load_parameters(self, attributes: Dict[str, Dict[str, Any]]) -> None:
def _load_parameters(self, attributes: Dict[str, Dict[str, Any] | Any]) -> None:
for key, items in attributes.items():
if not isinstance(items, dict):
_LOGGER.info("Loading Attributes - Skipping %s", str(items))
continue
for name, data in items.items():
self._create_parameters(data, name, key)
for rule in self._rules:
@@ -102,10 +105,12 @@ class HonCommand:
if name == "zoneMap" and self._appliance.zone:
data["default"] = self._appliance.zone
if data.get("category") == "rule":
if "fixedValue" not in data:
_LOGGER.error("Rule not supported: %s", data)
else:
if "fixedValue" in data:
self._rules.append(HonRuleSet(self, data["fixedValue"]))
elif "enumValues" in data:
self._rules.append(HonRuleSet(self, data["enumValues"]))
else:
_LOGGER.warning("Rule not supported: %s", data)
match data.get("typology"):
case "range":
self._parameters[name] = HonParameterRange(name, data, parameter)
@@ -130,17 +135,23 @@ class HonCommand:
async def send_specific(self, param_names: List[str]) -> bool:
params: Dict[str, str | float] = {}
for key, parameter in self._parameters.items():
if key in param_names:
if key in param_names or parameter.mandatory:
params[key] = parameter.value
return await self.send_parameters(params)
async def send_parameters(self, params: Dict[str, str | float]) -> bool:
ancillary_params = self.parameter_groups.get("ancillaryParameters", {})
ancillary_params.pop("programRules", None)
if "prStr" in params:
params["prStr"] = self._category_name.upper()
self.appliance.sync_command_to_params(self.name)
try:
result = await self.api.send_command(
self._appliance, self._name, params, ancillary_params
self._appliance,
self._name,
params,
ancillary_params,
self._category_name,
)
if not result:
_LOGGER.error(result)
+24 -5
View File
@@ -24,12 +24,16 @@ class HonAPI:
email: str = "",
password: str = "",
anonymous: bool = False,
mobile_id: str = "",
refresh_token: str = "",
session: Optional[ClientSession] = None,
) -> None:
super().__init__()
self._email: str = email
self._password: str = password
self._anonymous: bool = anonymous
self._mobile_id: str = mobile_id
self._refresh_token: str = refresh_token
self._hon_handler: Optional[HonConnectionHandler] = None
self._hon_anonymous_handler: Optional[HonAnonymousConnectionHandler] = None
self._session: Optional[ClientSession] = session
@@ -69,7 +73,10 @@ class HonAPI:
).create()
if not self._anonymous:
self._hon_handler = await HonConnectionHandler(
self._email, self._password, self._session
self._email,
self._password,
session=self._session,
mobile_id=self._mobile_id,
).create()
return self
@@ -190,6 +197,7 @@ class HonAPI:
command: str,
parameters: Dict[str, Any],
ancillary_parameters: Dict[str, Any],
program_name: str = "",
) -> bool:
now: str = datetime.utcnow().isoformat()
data: Dict[str, Any] = {
@@ -208,6 +216,8 @@ class HonAPI:
"parameters": parameters,
"applianceType": appliance.appliance_type,
}
if command == "startProgram" and program_name:
data.update({"programName": program_name.upper()})
url: str = f"{const.API_URL}/commands/v1/send"
async with self._hon.post(url, json=data) as response:
json_data: Dict[str, Any] = await response.json()
@@ -278,10 +288,12 @@ class TestAPI(HonAPI):
async def load_appliances(self) -> List[Dict[str, Any]]:
result = []
for appliance in self._path.glob("*/"):
with open(
appliance / "appliance_data.json", "r", encoding="utf-8"
) as json_file:
result.append(json.loads(json_file.read()))
file = appliance / "appliance_data.json"
with open(file, "r", encoding="utf-8") as json_file:
try:
result.append(json.loads(json_file.read()))
except json.decoder.JSONDecodeError as error:
_LOGGER.error("%s - %s", str(file), error)
return result
async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]:
@@ -317,5 +329,12 @@ class TestAPI(HonAPI):
command: str,
parameters: Dict[str, Any],
ancillary_parameters: Dict[str, Any],
program_name: str = "",
) -> bool:
_LOGGER.info(
"%s - %s - %s",
str(parameters),
str(ancillary_parameters),
str(program_name),
)
return True
+4 -2
View File
@@ -199,7 +199,7 @@ class HonAuth:
if access_token := re.findall("access_token=(.*?)&", text):
self._auth.access_token = access_token[0]
if refresh_token := re.findall("refresh_token=(.*?)&", text):
self._auth.refresh_token = refresh_token[0]
self._auth.refresh_token = parse.unquote(refresh_token[0])
if id_token := re.findall("id_token=(.*?)&", text):
self._auth.id_token = id_token[0]
return bool(access_token and refresh_token and id_token)
@@ -264,7 +264,9 @@ class HonAuth:
except exceptions.HonNoAuthenticationNeeded:
return
async def refresh(self) -> bool:
async def refresh(self, refresh_token: str = "") -> bool:
if refresh_token:
self._auth.refresh_token = refresh_token
params = {
"client_id": const.CLIENT_ID,
"refresh_token": self._auth.refresh_token,
+2 -3
View File
@@ -1,16 +1,15 @@
import secrets
from typing import Dict
from pyhon import const
class HonDevice:
def __init__(self) -> None:
def __init__(self, mobile_id: str = "") -> None:
self._app_version: str = const.APP_VERSION
self._os_version: int = const.OS_VERSION
self._os: str = const.OS
self._device_model: str = const.DEVICE_MODEL
self._mobile_id: str = secrets.token_hex(8)
self._mobile_id: str = mobile_id or const.MOBILE_ID
@property
def app_version(self) -> str:
+16 -3
View File
@@ -19,12 +19,18 @@ _LOGGER = logging.getLogger(__name__)
class HonConnectionHandler(ConnectionHandler):
def __init__(
self, email: str, password: str, session: Optional[aiohttp.ClientSession] = None
self,
email: str,
password: str,
session: Optional[aiohttp.ClientSession] = None,
mobile_id: str = "",
refresh_token: str = "",
) -> None:
super().__init__(session=session)
self._device: HonDevice = HonDevice()
self._device: HonDevice = HonDevice(mobile_id)
self._email: str = email
self._password: str = password
self._refresh_token: str = refresh_token
if not self._email:
raise HonAuthenticationError("An email address must be specified")
if not self._password:
@@ -43,10 +49,17 @@ class HonConnectionHandler(ConnectionHandler):
async def create(self) -> Self:
await super().create()
self._auth = HonAuth(self.session, self._email, self._password, self._device)
self._auth = HonAuth(
self.session,
self._email,
self._password,
self._device,
)
return self
async def _check_headers(self, headers: Dict[str, str]) -> Dict[str, str]:
if self._refresh_token:
await self.auth.refresh(self._refresh_token)
if not (self.auth.cognito_token and self.auth.id_token):
await self.auth.authenticate()
headers["cognito-token"] = self.auth.cognito_token
+5 -4
View File
@@ -6,8 +6,9 @@ CLIENT_ID = (
"3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9."
"HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6"
)
APP_VERSION = "2.1.2"
OS_VERSION = 31
APP_VERSION = "2.6.5"
OS_VERSION = 999
OS = "android"
DEVICE_MODEL = "exynos9820"
USER_AGENT = "Chrome/110.0.5481.153"
DEVICE_MODEL = "pyhOn"
USER_AGENT = "Chrome/999.999.999.999"
MOBILE_ID = "pyhOn"
+15 -3
View File
@@ -21,6 +21,8 @@ class Hon:
email: Optional[str] = "",
password: Optional[str] = "",
session: Optional[ClientSession] = None,
mobile_id: str = "",
refresh_token: str = "",
test_data_path: Optional[Path] = None,
):
self._email: Optional[str] = email
@@ -29,6 +31,8 @@ class Hon:
self._appliances: List[HonAppliance] = []
self._api: Optional[HonAPI] = None
self._test_data_path: Path = test_data_path or Path().cwd()
self._mobile_id: str = mobile_id
self._refresh_token: str = refresh_token
async def __aenter__(self) -> Self:
return await self.create()
@@ -61,7 +65,11 @@ class Hon:
async def create(self) -> Self:
self._api = await HonAPI(
self.email, self.password, session=self._session
self.email,
self.password,
session=self._session,
mobile_id=self._mobile_id,
refresh_token=self._refresh_token,
).create()
await self.setup()
return self
@@ -103,8 +111,12 @@ class Hon:
)
await self._create_appliance(appliance, self.api)
if (
test_data := self._test_data_path / "hon-test-data" / "test_data"
).exists() or (test_data := test_data / "test_data").exists():
self._test_data_path
and (
test_data := self._test_data_path / "hon-test-data" / "test_data"
).exists()
or (test_data := test_data / "..").exists()
):
api = TestAPI(test_data)
for appliance in await api.load_appliances():
await self._create_appliance(appliance, api)
+3 -2
View File
@@ -68,8 +68,9 @@ class HonParameter:
self._triggers.setdefault(value, []).append((func, data))
def check_trigger(self, value: str | float) -> None:
if str(value) in self._triggers:
for trigger in self._triggers[str(value)]:
triggers = {str(k).lower(): v for k, v in self._triggers.items()}
if str(value).lower() in triggers:
for trigger in triggers[str(value)]:
func, args = trigger
func(args)
+1 -1
View File
@@ -18,7 +18,7 @@ class HonParameterFixed(HonParameter):
@property
def value(self) -> str | float:
return self._value if self._value is not None else "0"
return self._value if self._value != "" else "0"
@value.setter
def value(self, value: str | float) -> None:
+6 -1
View File
@@ -69,4 +69,9 @@ class HonParameterRange(HonParameter):
@property
def values(self) -> List[str]:
return [str(i) for i in range(int(self.min), int(self.max) + 1, int(self.step))]
result = []
i = self.min
while i < self.max:
i += self.step
result.append(str(i))
return result
View File
+9
View File
@@ -56,6 +56,11 @@ class HonRuleSet:
extra[trigger_key] = trigger_value
for extra_key, extra_data in param_data.items():
self._parse_conditions(param_key, extra_key, extra_data, extra)
else:
param_data = {"typology": "fixed", "fixedValue": param_data}
self._create_rule(
param_key, trigger_key, trigger_value, param_data, extra
)
def _create_rule(
self,
@@ -102,6 +107,10 @@ class HonRuleSet:
param.values = [str(value)]
param.value = str(value)
elif isinstance(param, HonParameterRange):
if float(value) < param.min:
param.min = float(value)
elif float(value) > param.max:
param.max = float(value)
param.value = float(value)
return
param.value = str(value)
+1 -2
View File
@@ -14,8 +14,7 @@ if TYPE_CHECKING:
class Callback(Protocol): # pylint: disable=too-few-public-methods
def __call__(
self, url: str | URL, *args: Any, **kwargs: Any
) -> aiohttp.client._RequestContextManager:
...
) -> aiohttp.client._RequestContextManager: ...
Parameter = Union[
+3 -3
View File
@@ -1,3 +1,3 @@
aiohttp==3.8.4
yarl==1.8.2
typing-extensions==4.7.1
aiohttp>=3.8.6
yarl>=1.8
typing-extensions>=4.8
+5 -4
View File
@@ -1,4 +1,5 @@
black==23.7.0
flake8==6.0.0
mypy==1.4.1
pylint==2.17.4
black>=22.12
flake8>=6.0
mypy>=0.991
pylint>=2.15
setuptools>=62.3
+3 -2
View File
@@ -7,7 +7,7 @@ with open("README.md", "r", encoding="utf-8") as f:
setup(
name="pyhOn",
version="0.15.0",
version="0.16.0",
author="Andre Basche",
description="Control hOn devices with python",
long_description=long_description,
@@ -21,7 +21,7 @@ setup(
packages=find_packages(),
include_package_data=True,
python_requires=">=3.10",
install_requires=["aiohttp==3.8.4", "typing-extensions==4.7.1"],
install_requires=["aiohttp>=3.8.6", "typing-extensions>=4.8", "yarl>=1.8"],
classifiers=[
"Development Status :: 4 - Beta",
"Environment :: Console",
@@ -30,6 +30,7 @@ setup(
"Operating System :: OS Independent",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules",
],
entry_points={