Compare commits
2 Commits
Author | SHA1 | Date |
---|---|---|
Andre Basche | 84000fb76c | 1 year ago |
Andre Basche | 26528a7310 | 1 year ago |
@ -1,9 +0,0 @@
|
||||
[MESSAGES CONTROL]
|
||||
|
||||
disable=missing-docstring
|
||||
|
||||
[FORMAT]
|
||||
|
||||
max-args=6
|
||||
max-attributes=8
|
||||
max-line-length=88
|
@ -1 +0,0 @@
|
||||
include pyhon/py.typed
|
@ -1,24 +0,0 @@
|
||||
[mypy]
|
||||
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
|
@ -1,25 +1,15 @@
|
||||
from typing import Dict, Any, TYPE_CHECKING
|
||||
|
||||
from pyhon.parameter.program import HonParameterProgram
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pyhon.appliance import HonAppliance
|
||||
|
||||
|
||||
class ApplianceBase:
|
||||
def __init__(self, appliance: "HonAppliance"):
|
||||
def __init__(self, appliance):
|
||||
self.parent = appliance
|
||||
|
||||
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def attributes(self, data):
|
||||
program_name = "No Program"
|
||||
if program := int(str(data.get("parameters", {}).get("prCode", "0"))):
|
||||
if program := int(data["parameters"].get("prCode", "0")):
|
||||
if start_cmd := self.parent.settings.get("startProgram.program"):
|
||||
if isinstance(start_cmd, HonParameterProgram) and (
|
||||
ids := start_cmd.ids
|
||||
):
|
||||
if ids := start_cmd.ids:
|
||||
program_name = ids.get(program, program_name)
|
||||
data["programName"] = program_name
|
||||
return data
|
||||
|
||||
def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def settings(self, settings):
|
||||
return settings
|
||||
|
@ -1,13 +1,10 @@
|
||||
# pylint: disable=duplicate-code
|
||||
from typing import Any, Dict
|
||||
|
||||
from pyhon.appliances.base import ApplianceBase
|
||||
|
||||
|
||||
class Appliance(ApplianceBase):
|
||||
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def attributes(self, data):
|
||||
data = super().attributes(data)
|
||||
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
|
||||
data["parameters"]["machMode"].value = "0"
|
||||
if data["lastConnEvent"]["category"] == "DISCONNECTED":
|
||||
data["parameters"]["machMode"] = "0"
|
||||
data["active"] = bool(data.get("activity"))
|
||||
return data
|
||||
|
@ -1,16 +1,19 @@
|
||||
from typing import Any, Dict
|
||||
|
||||
from pyhon.appliances.base import ApplianceBase
|
||||
|
||||
|
||||
class Appliance(ApplianceBase):
|
||||
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def attributes(self, data):
|
||||
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
|
||||
if data["lastConnEvent"]["category"] == "DISCONNECTED":
|
||||
data["parameters"]["temp"] = "0"
|
||||
data["parameters"]["onOffStatus"] = "0"
|
||||
data["parameters"]["remoteCtrValid"] = "0"
|
||||
data["parameters"]["remainingTimeMM"] = "0"
|
||||
|
||||
data["active"] = data["parameters"]["onOffStatus"] == "1"
|
||||
|
||||
if program := int(data["parameters"]["prCode"]):
|
||||
ids = self.parent.settings["startProgram.program"].ids
|
||||
data["programName"] = ids.get(program, "")
|
||||
|
||||
data["active"] = data["parameters"]["onOffStatus"].value == 1
|
||||
return data
|
||||
|
@ -1,17 +1,14 @@
|
||||
# pylint: disable=duplicate-code
|
||||
from typing import Dict, Any
|
||||
|
||||
from pyhon.appliances.base import ApplianceBase
|
||||
|
||||
|
||||
class Appliance(ApplianceBase):
|
||||
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def attributes(self, data):
|
||||
data = super().attributes(data)
|
||||
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
|
||||
data["parameters"]["machMode"].value = "0"
|
||||
if data["lastConnEvent"]["category"] == "DISCONNECTED":
|
||||
data["parameters"]["machMode"] = "0"
|
||||
data["active"] = bool(data.get("activity"))
|
||||
data["pause"] = data["parameters"]["machMode"] == "3"
|
||||
return data
|
||||
|
||||
def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def settings(self, settings):
|
||||
return settings
|
||||
|
@ -1,16 +0,0 @@
|
||||
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)
|
||||
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
|
@ -1,17 +1,14 @@
|
||||
# pylint: disable=duplicate-code
|
||||
from typing import Any, Dict
|
||||
|
||||
from pyhon.appliances.base import ApplianceBase
|
||||
|
||||
|
||||
class Appliance(ApplianceBase):
|
||||
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def attributes(self, data):
|
||||
data = super().attributes(data)
|
||||
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
|
||||
data["parameters"]["machMode"].value = "0"
|
||||
if data["lastConnEvent"]["category"] == "DISCONNECTED":
|
||||
data["parameters"]["machMode"] = "0"
|
||||
data["active"] = bool(data.get("activity"))
|
||||
data["pause"] = data["parameters"]["machMode"] == "3"
|
||||
return data
|
||||
|
||||
def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def settings(self, settings):
|
||||
return settings
|
||||
|
@ -1,58 +0,0 @@
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional, Final, Dict
|
||||
|
||||
from pyhon.helper import str_to_float
|
||||
|
||||
|
||||
class HonAttribute:
|
||||
_LOCK_TIMEOUT: Final = 10
|
||||
|
||||
def __init__(self, data: Dict[str, str] | str):
|
||||
self._value: str = ""
|
||||
self._last_update: Optional[datetime] = None
|
||||
self._lock_timestamp: Optional[datetime] = None
|
||||
self.update(data)
|
||||
|
||||
@property
|
||||
def value(self) -> float | str:
|
||||
"""Attribute value"""
|
||||
try:
|
||||
return str_to_float(self._value)
|
||||
except ValueError:
|
||||
return self._value
|
||||
|
||||
@value.setter
|
||||
def value(self, value: str) -> None:
|
||||
self._value = value
|
||||
|
||||
@property
|
||||
def last_update(self) -> Optional[datetime]:
|
||||
"""Timestamp of last api update"""
|
||||
return self._last_update
|
||||
|
||||
@property
|
||||
def lock(self) -> bool:
|
||||
"""Shows if value changes are forbidden"""
|
||||
if not self._lock_timestamp:
|
||||
return False
|
||||
lock_until = self._lock_timestamp + timedelta(seconds=self._LOCK_TIMEOUT)
|
||||
return lock_until >= datetime.utcnow()
|
||||
|
||||
def update(self, data: Dict[str, str] | str, shield: bool = False) -> bool:
|
||||
if self.lock and not shield:
|
||||
return False
|
||||
if shield:
|
||||
self._lock_timestamp = datetime.utcnow()
|
||||
if isinstance(data, str):
|
||||
self.value = data
|
||||
return True
|
||||
self.value = data.get("parNewVal", "")
|
||||
if last_update := data.get("lastUpdate"):
|
||||
try:
|
||||
self._last_update = datetime.fromisoformat(last_update)
|
||||
except ValueError:
|
||||
self._last_update = None
|
||||
return True
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self._value
|
@ -1,227 +0,0 @@
|
||||
import asyncio
|
||||
from contextlib import suppress
|
||||
from copy import copy
|
||||
from typing import Dict, Any, Optional, TYPE_CHECKING, List
|
||||
|
||||
from pyhon.commands import HonCommand
|
||||
from pyhon.exceptions import NoAuthenticationException
|
||||
from pyhon.parameter.fixed import HonParameterFixed
|
||||
from pyhon.parameter.program import HonParameterProgram
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pyhon import HonAPI
|
||||
from pyhon.appliance import HonAppliance
|
||||
|
||||
|
||||
class HonCommandLoader:
|
||||
"""Loads and parses hOn command data"""
|
||||
|
||||
def __init__(self, api: "HonAPI", appliance: "HonAppliance") -> None:
|
||||
self._api: "HonAPI" = api
|
||||
self._appliance: "HonAppliance" = appliance
|
||||
self._api_commands: Dict[str, Any] = {}
|
||||
self._favourites: List[Dict[str, Any]] = []
|
||||
self._command_history: List[Dict[str, Any]] = []
|
||||
self._commands: Dict[str, HonCommand] = {}
|
||||
self._appliance_data: Dict[str, Any] = {}
|
||||
self._additional_data: Dict[str, Any] = {}
|
||||
|
||||
@property
|
||||
def api(self) -> "HonAPI":
|
||||
"""api connection object"""
|
||||
if self._api is None:
|
||||
raise NoAuthenticationException("Missing hOn login")
|
||||
return self._api
|
||||
|
||||
@property
|
||||
def appliance(self) -> "HonAppliance":
|
||||
"""appliance object"""
|
||||
return self._appliance
|
||||
|
||||
@property
|
||||
def commands(self) -> Dict[str, HonCommand]:
|
||||
"""Get list of hon commands"""
|
||||
return self._commands
|
||||
|
||||
@property
|
||||
def appliance_data(self) -> Dict[str, Any]:
|
||||
"""Get command appliance data"""
|
||||
return self._appliance_data
|
||||
|
||||
@property
|
||||
def additional_data(self) -> Dict[str, Any]:
|
||||
"""Get command additional data"""
|
||||
return self._additional_data
|
||||
|
||||
async def load_commands(self) -> None:
|
||||
"""Trigger loading of command data"""
|
||||
await self._load_data()
|
||||
self._appliance_data = self._api_commands.pop("applianceModel", {})
|
||||
self._get_commands()
|
||||
self._add_favourites()
|
||||
self._recover_last_command_states()
|
||||
|
||||
async def _load_commands(self) -> None:
|
||||
self._api_commands = await self._api.load_commands(self._appliance)
|
||||
|
||||
async def _load_favourites(self) -> None:
|
||||
self._favourites = await self._api.load_favourites(self._appliance)
|
||||
|
||||
async def _load_command_history(self) -> None:
|
||||
self._command_history = await self._api.load_command_history(self._appliance)
|
||||
|
||||
async def _load_data(self) -> None:
|
||||
"""Callback parallel all relevant data"""
|
||||
await asyncio.gather(
|
||||
*[
|
||||
self._load_commands(),
|
||||
self._load_favourites(),
|
||||
self._load_command_history(),
|
||||
]
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _is_command(data: Dict[str, Any]) -> bool:
|
||||
"""Check if dict can be parsed as command"""
|
||||
return (
|
||||
data.get("description") is not None and data.get("protocolType") is not None
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _clean_name(category: str) -> str:
|
||||
"""Clean up category name"""
|
||||
if "PROGRAM" in category:
|
||||
return category.split(".")[-1].lower()
|
||||
return category
|
||||
|
||||
def _get_commands(self) -> None:
|
||||
"""Generates HonCommand dict from api data"""
|
||||
commands = []
|
||||
for name, data in self._api_commands.items():
|
||||
if command := self._parse_command(data, name):
|
||||
commands.append(command)
|
||||
self._commands = {c.name: c for c in commands}
|
||||
|
||||
def _parse_command(
|
||||
self,
|
||||
data: Dict[str, Any] | str,
|
||||
command_name: str,
|
||||
categories: Optional[Dict[str, "HonCommand"]] = None,
|
||||
category_name: str = "",
|
||||
) -> Optional[HonCommand]:
|
||||
"""Try to crate HonCommand object"""
|
||||
if not isinstance(data, dict):
|
||||
self._additional_data[command_name] = data
|
||||
return None
|
||||
if self._is_command(data):
|
||||
return HonCommand(
|
||||
command_name,
|
||||
data,
|
||||
self._appliance,
|
||||
category_name=category_name,
|
||||
categories=categories,
|
||||
)
|
||||
if category := self._parse_categories(data, command_name):
|
||||
return category
|
||||
return None
|
||||
|
||||
def _parse_categories(
|
||||
self, data: Dict[str, Any], command_name: str
|
||||
) -> Optional[HonCommand]:
|
||||
"""Parse categories and create reference to other"""
|
||||
categories: Dict[str, HonCommand] = {}
|
||||
for category, value in data.items():
|
||||
if command := self._parse_command(
|
||||
value, command_name, category_name=category, categories=categories
|
||||
):
|
||||
categories[self._clean_name(category)] = command
|
||||
if categories:
|
||||
# setParameters should be at first place
|
||||
if "setParameters" in categories:
|
||||
return categories["setParameters"]
|
||||
return list(categories.values())[0]
|
||||
return None
|
||||
|
||||
def _get_last_command_index(self, name: str) -> Optional[int]:
|
||||
"""Get index of last command execution"""
|
||||
return next(
|
||||
(
|
||||
index
|
||||
for (index, d) in enumerate(self._command_history)
|
||||
if d.get("command", {}).get("commandName") == name
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
def _set_last_category(
|
||||
self, command: HonCommand, name: str, parameters: Dict[str, Any]
|
||||
) -> HonCommand:
|
||||
"""Set category to last state"""
|
||||
if command.categories:
|
||||
if program := parameters.pop("program", None):
|
||||
command.category = self._clean_name(program)
|
||||
elif category := parameters.pop("category", None):
|
||||
command.category = category
|
||||
else:
|
||||
return command
|
||||
return self.commands[name]
|
||||
return command
|
||||
|
||||
def _recover_last_command_states(self) -> None:
|
||||
"""Set commands to last state"""
|
||||
for name, command in self.commands.items():
|
||||
if (last_index := self._get_last_command_index(name)) is None:
|
||||
continue
|
||||
last_command = self._command_history[last_index]
|
||||
parameters = last_command.get("command", {}).get("parameters", {})
|
||||
command = self._set_last_category(command, name, parameters)
|
||||
for key, data in command.settings.items():
|
||||
if parameters.get(key) is None:
|
||||
continue
|
||||
with suppress(ValueError):
|
||||
data.value = parameters.get(key)
|
||||
|
||||
def _add_favourites(self) -> None:
|
||||
"""Patch program categories with favourites"""
|
||||
for favourite in self._favourites:
|
||||
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
|
||||
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
|
@ -1,27 +1,21 @@
|
||||
import logging
|
||||
from collections.abc import AsyncIterator
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import Dict, Any
|
||||
|
||||
import aiohttp
|
||||
from yarl import URL
|
||||
from typing import Callable, Dict
|
||||
|
||||
from pyhon import const
|
||||
from pyhon.connection.handler.base import ConnectionHandler
|
||||
from pyhon.typedefs import Callback
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class HonAnonymousConnectionHandler(ConnectionHandler):
|
||||
_HEADERS: Dict[str, str] = ConnectionHandler._HEADERS | {"x-api-key": const.API_KEY}
|
||||
_HEADERS: Dict = ConnectionHandler._HEADERS | {"x-api-key": const.API_KEY}
|
||||
|
||||
@asynccontextmanager
|
||||
async def _intercept(
|
||||
self, method: Callback, url: str | URL, *args: Any, **kwargs: Dict[str, Any]
|
||||
) -> AsyncIterator[aiohttp.ClientResponse]:
|
||||
async def _intercept(self, method: Callable, *args, **kwargs) -> AsyncIterator:
|
||||
kwargs["headers"] = kwargs.pop("headers", {}) | self._HEADERS
|
||||
async with method(url, *args, **kwargs) as response:
|
||||
async with method(*args, **kwargs) as response:
|
||||
if response.status == 403:
|
||||
_LOGGER.error("Can't authenticate anymore")
|
||||
yield response
|
||||
|
@ -1,99 +0,0 @@
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, List, Tuple
|
||||
|
||||
from pyhon import printer
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pyhon.appliance import HonAppliance
|
||||
|
||||
|
||||
def anonymize_data(data: str) -> str:
|
||||
default_date = "1970-01-01T00:00:00.0Z"
|
||||
default_mac = "xx-xx-xx-xx-xx-xx"
|
||||
data = re.sub("[0-9A-Fa-f]{2}(-[0-9A-Fa-f]{2}){5}", default_mac, data)
|
||||
data = re.sub("[\\d-]{10}T[\\d:]{8}(.\\d+)?Z", default_date, data)
|
||||
for sensible in [
|
||||
"serialNumber",
|
||||
"code",
|
||||
"nickName",
|
||||
"mobileId",
|
||||
"PK",
|
||||
"SK",
|
||||
"lat",
|
||||
"lng",
|
||||
]:
|
||||
for match in re.findall(f'"{sensible}.*?":\\s"?(.+?)"?,?\\n', data):
|
||||
replace = re.sub("[a-z]", "x", match)
|
||||
replace = re.sub("[A-Z]", "X", replace)
|
||||
replace = re.sub("\\d", "1", replace)
|
||||
data = data.replace(match, replace)
|
||||
return data
|
||||
|
||||
|
||||
async def load_data(appliance: "HonAppliance", topic: str) -> Tuple[str, str]:
|
||||
return topic, await getattr(appliance.api, f"load_{topic}")(appliance)
|
||||
|
||||
|
||||
def write_to_json(data: str, topic: str, path: Path, anonymous: bool = False) -> Path:
|
||||
json_data = json.dumps(data, indent=4)
|
||||
if anonymous:
|
||||
json_data = anonymize_data(json_data)
|
||||
file = path / f"{topic}.json"
|
||||
with open(file, "w", encoding="utf-8") as json_file:
|
||||
json_file.write(json_data)
|
||||
return file
|
||||
|
||||
|
||||
async def appliance_data(
|
||||
appliance: "HonAppliance", path: Path, anonymous: bool = False
|
||||
) -> List[Path]:
|
||||
requests = [
|
||||
"commands",
|
||||
"attributes",
|
||||
"command_history",
|
||||
"statistics",
|
||||
"maintenance",
|
||||
"appliance_data",
|
||||
]
|
||||
path /= f"{appliance.appliance_type}_{appliance.model_id}".lower()
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
api_data = await asyncio.gather(*[load_data(appliance, name) for name in requests])
|
||||
return [write_to_json(data, topic, path, anonymous) for topic, data in api_data]
|
||||
|
||||
|
||||
async def zip_archive(
|
||||
appliance: "HonAppliance", path: Path, anonymous: bool = False
|
||||
) -> str:
|
||||
data = await appliance_data(appliance, path, anonymous)
|
||||
archive = data[0].parent
|
||||
shutil.make_archive(str(archive), "zip", archive)
|
||||
shutil.rmtree(archive)
|
||||
return f"{archive.stem}.zip"
|
||||
|
||||
|
||||
def yaml_export(appliance: "HonAppliance", anonymous: bool = False) -> str:
|
||||
data = {
|
||||
"attributes": appliance.attributes.copy(),
|
||||
"appliance": appliance.info,
|
||||
"statistics": appliance.statistics,
|
||||
"additional_data": appliance.additional_data,
|
||||
}
|
||||
data |= {n: c.parameter_groups for n, c in appliance.commands.items()}
|
||||
extra = {n: c.data for n, c in appliance.commands.items() if c.data}
|
||||
if extra:
|
||||
data |= {"extra_command_data": extra}
|
||||
if anonymous:
|
||||
for sensible in ["serialNumber", "coords"]:
|
||||
data.get("appliance", {}).pop(sensible, None)
|
||||
result = printer.pretty_print({"data": data})
|
||||
if commands := printer.create_commands(appliance.commands):
|
||||
result += printer.pretty_print({"commands": commands})
|
||||
if rules := printer.create_rules(appliance.commands):
|
||||
result += printer.pretty_print({"rules": rules})
|
||||
if anonymous:
|
||||
result = anonymize_data(result)
|
||||
return result
|
@ -1,5 +1,75 @@
|
||||
def str_to_float(string: str | float) -> float:
|
||||
try:
|
||||
return int(string)
|
||||
except ValueError:
|
||||
return float(str(string).replace(",", "."))
|
||||
def key_print(data, key="", start=True):
|
||||
result = ""
|
||||
if isinstance(data, list):
|
||||
for i, value in enumerate(data):
|
||||
result += key_print(value, key=f"{key}.{i}", start=False)
|
||||
elif isinstance(data, dict):
|
||||
for k, value in sorted(data.items()):
|
||||
result += key_print(value, key=k if start else f"{key}.{k}", start=False)
|
||||
else:
|
||||
result += f"{key}: {data}\n"
|
||||
return result
|
||||
|
||||
|
||||
# yaml.dump() would be done the same, but needs an additional dependency...
|
||||
def pretty_print(data, key="", intend=0, is_list=False, whitespace=" "):
|
||||
result = ""
|
||||
if isinstance(data, list):
|
||||
if key:
|
||||
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n"
|
||||
intend += 1
|
||||
for i, value in enumerate(data):
|
||||
result += pretty_print(
|
||||
value, intend=intend, is_list=True, whitespace=whitespace
|
||||
)
|
||||
elif isinstance(data, dict):
|
||||
if key:
|
||||
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n"
|
||||
intend += 1
|
||||
for i, (key, value) in enumerate(sorted(data.items())):
|
||||
if is_list and not i:
|
||||
result += pretty_print(
|
||||
value, key=key, intend=intend, is_list=True, whitespace=whitespace
|
||||
)
|
||||
elif is_list:
|
||||
result += pretty_print(
|
||||
value, key=key, intend=intend + 1, whitespace=whitespace
|
||||
)
|
||||
else:
|
||||
result += pretty_print(
|
||||
value, key=key, intend=intend, whitespace=whitespace
|
||||
)
|
||||
else:
|
||||
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n"
|
||||
return result
|
||||
|
||||
|
||||
def create_command(commands, concat=False):
|
||||
result = {}
|
||||
for name, command in commands.items():
|
||||
for parameter, data in command.available_settings.items():
|
||||
if data.typology == "enum":
|
||||
value = data.values
|
||||
elif data.typology == "range":
|
||||
value = {"min": data.min, "max": data.max, "step": data.step}
|
||||
else:
|
||||
continue
|
||||
if not concat:
|
||||
result.setdefault(name, {})[parameter] = value
|
||||
else:
|
||||
result[f"{name}.{parameter}"] = value
|
||||
return result
|
||||
|
||||
|
||||
def create_rules(commands, concat=False):
|
||||
result = {}
|
||||
for name, command in commands.items():
|
||||
for parameter, data in command.available_settings.items():
|
||||
value = data.triggers
|
||||
if not value:
|
||||
continue
|
||||
if not concat:
|
||||
result.setdefault(name, {})[parameter] = value
|
||||
else:
|
||||
result[f"{name}.{parameter}"] = value
|
||||
return result
|
||||
|
@ -1,87 +0,0 @@
|
||||
from typing import Dict, Any, TYPE_CHECKING, List
|
||||
|
||||
from pyhon.parameter.enum import HonParameterEnum
|
||||
from pyhon.parameter.range import HonParameterRange
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pyhon.commands import HonCommand
|
||||
|
||||
|
||||
def key_print(data: Any, key: str = "", start: bool = True) -> str:
|
||||
result = ""
|
||||
if isinstance(data, list):
|
||||
for i, value in enumerate(data):
|
||||
result += key_print(value, key=f"{key}.{i}", start=False)
|
||||
elif isinstance(data, dict):
|
||||
for k, value in sorted(data.items()):
|
||||
result += key_print(value, key=k if start else f"{key}.{k}", start=False)
|
||||
else:
|
||||
result += f"{key}: {data}\n"
|
||||
return result
|
||||
|
||||
|
||||
# yaml.dump() would be done the same, but needs an additional dependency...
|
||||
def pretty_print(
|
||||
data: Any,
|
||||
key: str = "",
|
||||
intend: int = 0,
|
||||
is_list: bool = False,
|
||||
whitespace: str = " ",
|
||||
) -> str:
|
||||
result = ""
|
||||
space = whitespace * intend
|
||||
if isinstance(data, (dict, list)) and key:
|
||||
result += f"{space}{'- ' if is_list else ''}{key}:\n"
|
||||
intend += 1
|
||||
if isinstance(data, list):
|
||||
for i, value in enumerate(data):
|
||||
result += pretty_print(
|
||||
value, intend=intend, is_list=True, whitespace=whitespace
|
||||
)
|
||||
elif isinstance(data, dict):
|
||||
for i, (list_key, value) in enumerate(sorted(data.items())):
|
||||
result += pretty_print(
|
||||
value,
|
||||
key=list_key,
|
||||
intend=intend + (is_list if i else 0),
|
||||
is_list=is_list and not i,
|
||||
whitespace=whitespace,
|
||||
)
|
||||
else:
|
||||
result += f"{space}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n"
|
||||
return result
|
||||
|
||||
|
||||
def create_commands(
|
||||
commands: Dict[str, "HonCommand"], concat: bool = False
|
||||
) -> Dict[str, Any]:
|
||||
result: Dict[str, Any] = {}
|
||||
for name, command in commands.items():
|
||||
for parameter, data in command.available_settings.items():
|
||||
if isinstance(data, HonParameterEnum):
|
||||
value: List[str] | Dict[str, str | float] = data.values
|
||||
elif isinstance(data, HonParameterRange):
|
||||
value = {"min": data.min, "max": data.max, "step": data.step}
|
||||
else:
|
||||
continue
|
||||
if not concat:
|
||||
result.setdefault(name, {})[parameter] = value
|
||||
else:
|
||||
result[f"{name}.{parameter}"] = value
|
||||
return result
|
||||
|
||||
|
||||
def create_rules(
|
||||
commands: Dict[str, "HonCommand"], concat: bool = False
|
||||
) -> Dict[str, Any]:
|
||||
result: Dict[str, Any] = {}
|
||||
for name, command in commands.items():
|
||||
for parameter, data in command.available_settings.items():
|
||||
value = data.triggers
|
||||
if not value:
|
||||
continue
|
||||
if not concat:
|
||||
result.setdefault(name, {})[parameter] = value
|
||||
else:
|
||||
result[f"{name}.{parameter}"] = value
|
||||
return result
|
@ -1,27 +0,0 @@
|
||||
from typing import Union, Any, TYPE_CHECKING, Protocol
|
||||
|
||||
import aiohttp
|
||||
from yarl import URL
|
||||
|
||||
if 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
|
||||
|
||||
|
||||
class Callback(Protocol): # pylint: disable=too-few-public-methods
|
||||
def __call__(
|
||||
self, url: str | URL, *args: Any, **kwargs: Any
|
||||
) -> aiohttp.client._RequestContextManager:
|
||||
...
|
||||
|
||||
|
||||
Parameter = Union[
|
||||
"HonParameter",
|
||||
"HonParameterRange",
|
||||
"HonParameterEnum",
|
||||
"HonParameterFixed",
|
||||
"HonParameterProgram",
|
||||
]
|
@ -1,3 +1,2 @@
|
||||
aiohttp>=3.8
|
||||
yarl>=1.8
|
||||
typing-extensions>=4.8
|
||||
aiohttp==3.8.4
|
||||
yarl==1.8.2
|
||||
|
@ -1,5 +1,4 @@
|
||||
black>=22.12
|
||||
flake8>=6.0
|
||||
mypy>=0.991
|
||||
pylint>=2.15
|
||||
setuptools>=62.3
|
||||
black==23.3.0
|
||||
flake8==6.0.0
|
||||
mypy==1.2.0
|
||||
pylint==2.17.2
|
||||
|
Loading…
Reference in New Issue