Issue with sync_command #16
6
.github/workflows/python-check.yml
vendored
6
.github/workflows/python-check.yml
vendored
@ -34,9 +34,9 @@ jobs:
|
|||||||
- name: Type check with mypy
|
- name: Type check with mypy
|
||||||
run: |
|
run: |
|
||||||
mypy pyhon/
|
mypy pyhon/
|
||||||
# - name: Analysing the code with pylint
|
- name: Analysing the code with pylint
|
||||||
# run: |
|
run: |
|
||||||
# pylint --max-line-length 88 $(git ls-files '*.py')
|
pylint $(git ls-files '*.py')
|
||||||
- name: Check black style
|
- name: Check black style
|
||||||
run: |
|
run: |
|
||||||
black . --check
|
black . --check
|
||||||
|
7
.pylintrc
Normal file
7
.pylintrc
Normal file
@ -0,0 +1,7 @@
|
|||||||
|
[MESSAGES CONTROL]
|
||||||
|
|
||||||
|
disable=C,R
|
||||||
|
|
||||||
|
[FORMAT]
|
||||||
|
|
||||||
|
max-line-length=88
|
@ -30,13 +30,13 @@ def get_arguments() -> Dict[str, Any]:
|
|||||||
export.add_argument("--zip", help="create zip archive", action="store_true")
|
export.add_argument("--zip", help="create zip archive", action="store_true")
|
||||||
export.add_argument("--anonymous", help="anonymize data", action="store_true")
|
export.add_argument("--anonymous", help="anonymize data", action="store_true")
|
||||||
export.add_argument("directory", nargs="?", default=Path().cwd())
|
export.add_argument("directory", nargs="?", default=Path().cwd())
|
||||||
translate = subparser.add_parser(
|
translation = subparser.add_parser(
|
||||||
"translate", help="print available translation keys"
|
"translate", help="print available translation keys"
|
||||||
)
|
)
|
||||||
translate.add_argument(
|
translation.add_argument(
|
||||||
"translate", help="language (de, en, fr...)", metavar="LANGUAGE"
|
"translate", help="language (de, en, fr...)", metavar="LANGUAGE"
|
||||||
)
|
)
|
||||||
translate.add_argument("--json", help="print as json", action="store_true")
|
translation.add_argument("--json", help="print as json", action="store_true")
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"-i", "--import", help="import pyhon data", nargs="?", default=Path().cwd()
|
"-i", "--import", help="import pyhon data", nargs="?", default=Path().cwd()
|
||||||
)
|
)
|
||||||
|
@ -1,7 +1,6 @@
|
|||||||
from typing import Any, Dict
|
from typing import Any, Dict
|
||||||
|
|
||||||
from pyhon.appliances.base import ApplianceBase
|
from pyhon.appliances.base import ApplianceBase
|
||||||
from pyhon.parameter.program import HonParameterProgram
|
|
||||||
|
|
||||||
|
|
||||||
class Appliance(ApplianceBase):
|
class Appliance(ApplianceBase):
|
||||||
|
@ -1,14 +1,15 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
from contextlib import suppress
|
from contextlib import suppress
|
||||||
from copy import copy
|
from copy import copy
|
||||||
from typing import Dict, Any, Optional, TYPE_CHECKING, List, Collection
|
from typing import Dict, Any, Optional, TYPE_CHECKING, List
|
||||||
|
|
||||||
from pyhon.commands import HonCommand
|
from pyhon.commands import HonCommand
|
||||||
|
from pyhon.exceptions import NoAuthenticationException
|
||||||
from pyhon.parameter.fixed import HonParameterFixed
|
from pyhon.parameter.fixed import HonParameterFixed
|
||||||
from pyhon.parameter.program import HonParameterProgram
|
from pyhon.parameter.program import HonParameterProgram
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from pyhon import HonAPI, exceptions
|
from pyhon import HonAPI
|
||||||
from pyhon.appliance import HonAppliance
|
from pyhon.appliance import HonAppliance
|
||||||
|
|
||||||
|
|
||||||
@ -29,7 +30,7 @@ class HonCommandLoader:
|
|||||||
def api(self) -> "HonAPI":
|
def api(self) -> "HonAPI":
|
||||||
"""api connection object"""
|
"""api connection object"""
|
||||||
if self._api is None:
|
if self._api is None:
|
||||||
raise exceptions.NoAuthenticationException("Missing hOn login")
|
raise NoAuthenticationException("Missing hOn login")
|
||||||
return self._api
|
return self._api
|
||||||
|
|
||||||
@property
|
@property
|
||||||
|
@ -192,3 +192,7 @@ class HonCommand:
|
|||||||
else:
|
else:
|
||||||
result[name] = parameter
|
result[name] = parameter
|
||||||
return result
|
return result
|
||||||
|
|
||||||
|
def reset(self) -> None:
|
||||||
|
for parameter in self._parameters.values():
|
||||||
|
parameter.reset()
|
||||||
|
@ -21,7 +21,7 @@ class HonDevice:
|
|||||||
return self._os_version
|
return self._os_version
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def os(self) -> str:
|
def os_type(self) -> str:
|
||||||
return self._os
|
return self._os
|
||||||
|
|
||||||
@property
|
@property
|
||||||
@ -36,7 +36,7 @@ class HonDevice:
|
|||||||
result: Dict[str, str | int] = {
|
result: Dict[str, str | int] = {
|
||||||
"appVersion": self.app_version,
|
"appVersion": self.app_version,
|
||||||
"mobileId": self.mobile_id,
|
"mobileId": self.mobile_id,
|
||||||
"os": self.os,
|
"os": self.os_type,
|
||||||
"osVersion": self.os_version,
|
"osVersion": self.os_version,
|
||||||
"deviceModel": self.device_model,
|
"deviceModel": self.device_model,
|
||||||
}
|
}
|
||||||
|
@ -2,7 +2,7 @@ import logging
|
|||||||
from collections.abc import AsyncIterator
|
from collections.abc import AsyncIterator
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
from types import TracebackType
|
from types import TracebackType
|
||||||
from typing import Optional, Dict, Type, Any, Protocol
|
from typing import Optional, Dict, Type, Any
|
||||||
|
|
||||||
import aiohttp
|
import aiohttp
|
||||||
from typing_extensions import Self
|
from typing_extensions import Self
|
||||||
@ -47,10 +47,11 @@ class ConnectionHandler:
|
|||||||
return self
|
return self
|
||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
def _intercept(
|
async def _intercept(
|
||||||
self, method: Callback, url: str | URL, *args: Any, **kwargs: Dict[str, Any]
|
self, method: Callback, url: str | URL, *args: Any, **kwargs: Dict[str, Any]
|
||||||
) -> AsyncIterator[aiohttp.ClientResponse]:
|
) -> AsyncIterator[aiohttp.ClientResponse]:
|
||||||
raise NotImplementedError
|
async with method(url, *args, **kwargs) as response:
|
||||||
|
yield response
|
||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def get(
|
async def get(
|
||||||
|
@ -95,11 +95,11 @@ class HonConnectionHandler(ConnectionHandler):
|
|||||||
try:
|
try:
|
||||||
await response.json()
|
await response.json()
|
||||||
yield response
|
yield response
|
||||||
except json.JSONDecodeError:
|
except json.JSONDecodeError as exc:
|
||||||
_LOGGER.warning(
|
_LOGGER.warning(
|
||||||
"%s - JsonDecodeError %s - %s",
|
"%s - JsonDecodeError %s - %s",
|
||||||
response.request_info.url,
|
response.request_info.url,
|
||||||
response.status,
|
response.status,
|
||||||
await response.text(),
|
await response.text(),
|
||||||
)
|
)
|
||||||
raise HonAuthenticationError("Decode Error")
|
raise HonAuthenticationError("Decode Error") from exc
|
||||||
|
@ -7,14 +7,21 @@ if TYPE_CHECKING:
|
|||||||
class HonParameter:
|
class HonParameter:
|
||||||
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
|
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
|
||||||
self._key = key
|
self._key = key
|
||||||
self._category: str = attributes.get("category", "")
|
self._attributes = attributes
|
||||||
self._typology: str = attributes.get("typology", "")
|
self._category: str = ""
|
||||||
self._mandatory: int = attributes.get("mandatory", 0)
|
self._typology: str = ""
|
||||||
|
self._mandatory: int = 0
|
||||||
self._value: str | float = ""
|
self._value: str | float = ""
|
||||||
self._group: str = group
|
self._group: str = group
|
||||||
self._triggers: Dict[
|
self._triggers: Dict[
|
||||||
str, List[Tuple[Callable[["HonRule"], None], "HonRule"]]
|
str, List[Tuple[Callable[["HonRule"], None], "HonRule"]]
|
||||||
] = {}
|
] = {}
|
||||||
|
self._set_attributes()
|
||||||
|
|
||||||
|
def _set_attributes(self) -> None:
|
||||||
|
self._category = self._attributes.get("category", "")
|
||||||
|
self._typology = self._attributes.get("typology", "")
|
||||||
|
self._mandatory = self._attributes.get("mandatory", 0)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def key(self) -> str:
|
def key(self) -> str:
|
||||||
@ -85,3 +92,6 @@ class HonParameter:
|
|||||||
param[rule.param_key] = rule.param_data.get("defaultValue", "")
|
param[rule.param_key] = rule.param_data.get("defaultValue", "")
|
||||||
|
|
||||||
return result
|
return result
|
||||||
|
|
||||||
|
def reset(self) -> None:
|
||||||
|
self._set_attributes()
|
||||||
|
@ -10,12 +10,19 @@ def clean_value(value: str | float) -> str:
|
|||||||
class HonParameterEnum(HonParameter):
|
class HonParameterEnum(HonParameter):
|
||||||
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
|
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
|
||||||
super().__init__(key, attributes, group)
|
super().__init__(key, attributes, group)
|
||||||
self._default = attributes.get("defaultValue")
|
self._default: str | float = ""
|
||||||
self._value = self._default or "0"
|
self._value: str | float = ""
|
||||||
self._values: List[str] = attributes.get("enumValues", [])
|
self._values: List[str] = []
|
||||||
|
self._set_attributes()
|
||||||
if self._default and clean_value(self._default.strip("[]")) not in self.values:
|
if self._default and clean_value(self._default.strip("[]")) not in self.values:
|
||||||
self._values.append(self._default)
|
self._values.append(self._default)
|
||||||
|
|
||||||
|
def _set_attributes(self) -> None:
|
||||||
|
super()._set_attributes()
|
||||||
|
self._default = self._attributes.get("defaultValue", "")
|
||||||
|
self._value = self._default or "0"
|
||||||
|
self._values = self._attributes.get("enumValues", [])
|
||||||
|
|
||||||
def __repr__(self) -> str:
|
def __repr__(self) -> str:
|
||||||
return f"{self.__class__} (<{self.key}> {self.values})"
|
return f"{self.__class__} (<{self.key}> {self.values})"
|
||||||
|
|
||||||
|
@ -6,7 +6,12 @@ from pyhon.parameter.base import HonParameter
|
|||||||
class HonParameterFixed(HonParameter):
|
class HonParameterFixed(HonParameter):
|
||||||
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
|
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
|
||||||
super().__init__(key, attributes, group)
|
super().__init__(key, attributes, group)
|
||||||
self._value = attributes.get("fixedValue", None)
|
self._value: str | float = ""
|
||||||
|
self._set_attributes()
|
||||||
|
|
||||||
|
def _set_attributes(self) -> None:
|
||||||
|
super()._set_attributes()
|
||||||
|
self._value = self._attributes.get("fixedValue", "")
|
||||||
|
|
||||||
def __repr__(self) -> str:
|
def __repr__(self) -> str:
|
||||||
return f"{self.__class__} (<{self.key}> fixed)"
|
return f"{self.__class__} (<{self.key}> fixed)"
|
||||||
|
@ -37,7 +37,7 @@ class HonParameterProgram(HonParameterEnum):
|
|||||||
|
|
||||||
@values.setter
|
@values.setter
|
||||||
def values(self, values: List[str]) -> None:
|
def values(self, values: List[str]) -> None:
|
||||||
return
|
raise ValueError("Cant set values {values}")
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def ids(self) -> Dict[int, str]:
|
def ids(self) -> Dict[int, str]:
|
||||||
|
@ -7,11 +7,20 @@ from pyhon.parameter.base import HonParameter
|
|||||||
class HonParameterRange(HonParameter):
|
class HonParameterRange(HonParameter):
|
||||||
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
|
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
|
||||||
super().__init__(key, attributes, group)
|
super().__init__(key, attributes, group)
|
||||||
self._min: float = str_to_float(attributes["minimumValue"])
|
self._min: float = 0
|
||||||
self._max: float = str_to_float(attributes["maximumValue"])
|
self._max: float = 0
|
||||||
self._step: float = str_to_float(attributes["incrementValue"])
|
self._step: float = 0
|
||||||
self._default: float = str_to_float(attributes.get("defaultValue", self.min))
|
self._default: float = 0
|
||||||
self._value: float = self._default
|
self._value: float = 0
|
||||||
|
self._set_attributes()
|
||||||
|
|
||||||
|
def _set_attributes(self) -> None:
|
||||||
|
super()._set_attributes()
|
||||||
|
self._min = str_to_float(self._attributes["minimumValue"])
|
||||||
|
self._max = str_to_float(self._attributes["maximumValue"])
|
||||||
|
self._step = str_to_float(self._attributes["incrementValue"])
|
||||||
|
self._default = str_to_float(self._attributes.get("defaultValue", self.min))
|
||||||
|
self._value = self._default
|
||||||
|
|
||||||
def __repr__(self) -> str:
|
def __repr__(self) -> str:
|
||||||
return f"{self.__class__} (<{self.key}> [{self.min} - {self.max}])"
|
return f"{self.__class__} (<{self.key}> [{self.min} - {self.max}])"
|
||||||
|
Loading…
Reference in New Issue
Block a user