mirror of
https://github.com/Andre0512/pyhOn.git
synced 2024-12-22 18:55:32 +00:00
More type hints
This commit is contained in:
parent
834f25a639
commit
461a247ad3
|
@ -1,34 +1,49 @@
|
|||
from typing import Optional, Dict, Any, List, TYPE_CHECKING
|
||||
|
||||
from pyhon.parameter import (
|
||||
HonParameterFixed,
|
||||
HonParameterEnum,
|
||||
HonParameterRange,
|
||||
HonParameterProgram,
|
||||
HonParameter,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pyhon import HonAPI
|
||||
from pyhon.appliance import HonAppliance
|
||||
|
||||
|
||||
class HonCommand:
|
||||
def __init__(
|
||||
self, name: str, attributes, connector, device, programs=None, program_name=""
|
||||
self,
|
||||
name: str,
|
||||
attributes: Dict[str, Any],
|
||||
api: "HonAPI",
|
||||
appliance: "HonAppliance",
|
||||
programs: Optional[Dict[str, "HonCommand"]] = None,
|
||||
program_name: str = "",
|
||||
):
|
||||
self._connector = connector
|
||||
self._device = device
|
||||
self._name = name
|
||||
self._programs = programs or {}
|
||||
self._program_name = program_name
|
||||
self._description = attributes.get("description", "")
|
||||
self._parameters = self._create_parameters(attributes.get("parameters", {}))
|
||||
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", {})
|
||||
)
|
||||
|
||||
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)
|
||||
|
@ -41,38 +56,41 @@ class HonCommand:
|
|||
return result
|
||||
|
||||
@property
|
||||
def parameters(self):
|
||||
def parameters(self) -> Dict[str, HonParameter]:
|
||||
return self._parameters
|
||||
|
||||
@property
|
||||
def ancillary_parameters(self):
|
||||
def ancillary_parameters(self) -> Dict[str, str | float]:
|
||||
return {
|
||||
key: parameter.value
|
||||
for key, parameter in self._ancillary_parameters.items()
|
||||
}
|
||||
|
||||
async def send(self):
|
||||
async def send(self) -> bool:
|
||||
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
|
||||
return await self._api.send_command(
|
||||
self._appliance, self._name, parameters, self.ancillary_parameters
|
||||
)
|
||||
|
||||
@property
|
||||
def programs(self):
|
||||
def programs(self) -> Dict[str, "HonCommand"]:
|
||||
if self._programs is None:
|
||||
return {}
|
||||
return self._programs
|
||||
|
||||
@property
|
||||
def program(self):
|
||||
def program(self) -> str:
|
||||
return self._program_name
|
||||
|
||||
@program.setter
|
||||
def program(self, program):
|
||||
self._device.commands[self._name] = self._programs[program]
|
||||
def program(self, program: str) -> None:
|
||||
self._appliance.commands[self._name] = self.programs[program]
|
||||
|
||||
def _get_settings_keys(self, command=None):
|
||||
command = command or self
|
||||
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):
|
||||
|
@ -82,7 +100,7 @@ class HonCommand:
|
|||
return keys
|
||||
|
||||
@property
|
||||
def setting_keys(self):
|
||||
def setting_keys(self) -> List[str]:
|
||||
if not self._programs:
|
||||
return self._get_settings_keys()
|
||||
result = [
|
||||
|
@ -93,10 +111,10 @@ class HonCommand:
|
|||
return list(set(result + ["program"]))
|
||||
|
||||
@property
|
||||
def settings(self):
|
||||
def settings(self) -> Dict[str, HonParameter]:
|
||||
"""Parameters with typology enum and range"""
|
||||
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
|
||||
}
|
||||
|
|
20
pyhon/hon.py
20
pyhon/hon.py
|
@ -1,5 +1,6 @@
|
|||
import asyncio
|
||||
from typing import List, Optional, Dict, Any
|
||||
from types import TracebackType
|
||||
from typing import List, Optional, Dict, Any, Type
|
||||
|
||||
from aiohttp import ClientSession
|
||||
from typing_extensions import Self
|
||||
|
@ -16,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
|
||||
|
@ -52,12 +58,12 @@ 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 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)
|
||||
await self._create_appliance(appliance)
|
||||
|
||||
async def close(self):
|
||||
await self._api.close()
|
||||
async def close(self) -> None:
|
||||
await self.api.close()
|
||||
|
|
|
@ -1,4 +1,10 @@
|
|||
def str_to_float(string):
|
||||
from typing import Dict, Any, List, TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pyhon.commands import HonCommand
|
||||
|
||||
|
||||
def str_to_float(string: str | float) -> float:
|
||||
try:
|
||||
return int(string)
|
||||
except ValueError:
|
||||
|
@ -6,43 +12,44 @@ def str_to_float(string):
|
|||
|
||||
|
||||
class HonParameter:
|
||||
def __init__(self, key, attributes):
|
||||
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
|
||||
self._key = key
|
||||
self._category = attributes.get("category")
|
||||
self._typology = attributes.get("typology")
|
||||
self._mandatory = attributes.get("mandatory")
|
||||
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):
|
||||
def key(self) -> str:
|
||||
return self._key
|
||||
|
||||
@property
|
||||
def value(self):
|
||||
def value(self) -> str | float:
|
||||
return self._value if self._value is not None else "0"
|
||||
|
||||
@property
|
||||
def category(self):
|
||||
def category(self) -> str:
|
||||
return self._category
|
||||
|
||||
@property
|
||||
def typology(self):
|
||||
def typology(self) -> str:
|
||||
return self._typology
|
||||
|
||||
@property
|
||||
def mandatory(self):
|
||||
def mandatory(self) -> int:
|
||||
return self._mandatory
|
||||
|
||||
|
||||
class HonParameterFixed(HonParameter):
|
||||
def __init__(self, key, attributes):
|
||||
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
|
||||
super().__init__(key, attributes)
|
||||
self._value = attributes.get("fixedValue", None)
|
||||
|
||||
def __repr__(self):
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.__class__} (<{self.key}> fixed)"
|
||||
|
||||
@property
|
||||
def value(self):
|
||||
def value(self) -> str | float:
|
||||
return self._value if self._value is not None else "0"
|
||||
|
||||
@value.setter
|
||||
|
@ -52,35 +59,35 @@ class HonParameterFixed(HonParameter):
|
|||
|
||||
|
||||
class HonParameterRange(HonParameter):
|
||||
def __init__(self, key, attributes):
|
||||
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
|
||||
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
|
||||
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):
|
||||
def min(self) -> float:
|
||||
return self._min
|
||||
|
||||
@property
|
||||
def max(self):
|
||||
def max(self) -> float:
|
||||
return self._max
|
||||
|
||||
@property
|
||||
def step(self):
|
||||
def step(self) -> float:
|
||||
return self._step
|
||||
|
||||
@property
|
||||
def value(self):
|
||||
def value(self) -> float:
|
||||
return self._value if self._value is not None else self._min
|
||||
|
||||
@value.setter
|
||||
def value(self, value):
|
||||
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
|
||||
|
@ -91,25 +98,25 @@ class HonParameterRange(HonParameter):
|
|||
|
||||
|
||||
class HonParameterEnum(HonParameter):
|
||||
def __init__(self, key, attributes):
|
||||
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 = attributes.get("enumValues")
|
||||
self._values: List[str] = attributes.get("enumValues", [])
|
||||
|
||||
def __repr__(self):
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.__class__} (<{self.key}> {self.values})"
|
||||
|
||||
@property
|
||||
def values(self):
|
||||
def values(self) -> List[str]:
|
||||
return [str(value) for value in self._values]
|
||||
|
||||
@property
|
||||
def value(self):
|
||||
def value(self) -> str | float:
|
||||
return self._value if self._value is not None else self.values[0]
|
||||
|
||||
@value.setter
|
||||
def value(self, value):
|
||||
def value(self, value: str) -> None:
|
||||
if value in self.values:
|
||||
self._value = value
|
||||
else:
|
||||
|
@ -119,26 +126,25 @@ class HonParameterEnum(HonParameter):
|
|||
class HonParameterProgram(HonParameterEnum):
|
||||
_FILTER = ["iot_recipe", "iot_guided"]
|
||||
|
||||
def __init__(self, key, command):
|
||||
def __init__(self, key: str, command: "HonCommand") -> None:
|
||||
super().__init__(key, {})
|
||||
self._command = command
|
||||
self._value = command.program
|
||||
self._values = command.programs
|
||||
self._typology = "enum"
|
||||
self._filter = ""
|
||||
self._value: str = command.program
|
||||
self._values: List[str] = list(command.programs)
|
||||
self._typology: str = "enum"
|
||||
|
||||
@property
|
||||
def value(self):
|
||||
def value(self) -> str | float:
|
||||
return self._value
|
||||
|
||||
@value.setter
|
||||
def value(self, value):
|
||||
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):
|
||||
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)
|
||||
|
|
Loading…
Reference in a new issue