Compare commits

..

1 Commits

Author SHA1 Message Date
Andre Basche cf137dd176 Fix dependencies 2023-11-20 15:29:52 +01:00
18 changed files with 48 additions and 97 deletions
+1 -1
View File
@@ -4,6 +4,6 @@ disable=missing-docstring
[FORMAT] [FORMAT]
max-args=7 max-args=6
max-attributes=8 max-attributes=8
max-line-length=88 max-line-length=88
+6 -3
View File
@@ -38,7 +38,9 @@ def get_arguments() -> Dict[str, Any]:
"translate", help="language (de, en, fr...)", metavar="LANGUAGE" "translate", help="language (de, en, fr...)", metavar="LANGUAGE"
) )
translation.add_argument("--json", help="print as json", action="store_true") translation.add_argument("--json", help="print as json", action="store_true")
parser.add_argument("-i", "--import", help="import pyhon data", nargs="?") parser.add_argument(
"-i", "--import", help="import pyhon data", nargs="?", default=Path().cwd()
)
return vars(parser.parse_args()) return vars(parser.parse_args())
@@ -71,8 +73,9 @@ async def main() -> None:
if language := args.get("translate"): if language := args.get("translate"):
await translate(language, json_output=args.get("json", "")) await translate(language, json_output=args.get("json", ""))
return return
test_data_path = Path(path) if (path := args.get("import", "")) else None async with Hon(
async with Hon(*get_login_data(args), test_data_path=test_data_path) as hon: *get_login_data(args), test_data_path=Path(args.get("import", ""))
) as hon:
for device in hon.appliances: for device in hon.appliances:
if args.get("export"): if args.get("export"):
anonymous = args.get("anonymous", False) anonymous = args.get("anonymous", False)
+4 -2
View File
@@ -72,10 +72,12 @@ class HonAppliance:
return self.info[item] return self.info[item]
@overload @overload
def get(self, item: str, default: None = None) -> Any: ... def get(self, item: str, default: None = None) -> Any:
...
@overload @overload
def get(self, item: str, default: T) -> T: ... def get(self, item: str, default: T) -> T:
...
def get(self, item: str, default: Optional[T] = None) -> Any: def get(self, item: str, default: Optional[T] = None) -> Any:
try: try:
+5 -5
View File
@@ -7,10 +7,10 @@ class Appliance(ApplianceBase):
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data) data = super().attributes(data)
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
data["parameters"]["temp"].value = 0 data["parameters"]["temp"].value = "0"
data["parameters"]["onOffStatus"].value = 0 data["parameters"]["onOffStatus"].value = "0"
data["parameters"]["remoteCtrValid"].value = 0 data["parameters"]["remoteCtrValid"].value = "0"
data["parameters"]["remainingTimeMM"].value = 0 data["parameters"]["remainingTimeMM"].value = "0"
data["active"] = data["parameters"]["onOffStatus"].value == 1 data["active"] = data["parameters"]["onOffStatus"] == "1"
return data return data
+1 -1
View File
@@ -7,7 +7,7 @@ from pyhon.parameter.base import HonParameter
class Appliance(ApplianceBase): class Appliance(ApplianceBase):
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data) data = super().attributes(data)
parameter = data.get("parameters", {}).get("onOffStatus") parameter = data["parameters"]["onOffStatus"]
is_class = isinstance(parameter, HonParameter) is_class = isinstance(parameter, HonParameter)
data["active"] = parameter.value == 1 if is_class else parameter == 1 data["active"] = parameter.value == 1 if is_class else parameter == 1
return data return data
+1 -1
View File
@@ -109,7 +109,7 @@ class HonCommandLoader:
categories: Optional[Dict[str, "HonCommand"]] = None, categories: Optional[Dict[str, "HonCommand"]] = None,
category_name: str = "", category_name: str = "",
) -> Optional[HonCommand]: ) -> Optional[HonCommand]:
"""Try to create HonCommand object""" """Try to crate HonCommand object"""
if not isinstance(data, dict): if not isinstance(data, dict):
self._additional_data[command_name] = data self._additional_data[command_name] = data
return None return None
+2 -9
View File
@@ -89,11 +89,8 @@ class HonCommand:
def parameter_value(self) -> Dict[str, Union[str, float]]: def parameter_value(self) -> Dict[str, Union[str, float]]:
return {n: p.value for n, p in self._parameters.items()} return {n: p.value for n, p in self._parameters.items()}
def _load_parameters(self, attributes: Dict[str, Dict[str, Any] | Any]) -> None: def _load_parameters(self, attributes: Dict[str, Dict[str, Any]]) -> None:
for key, items in attributes.items(): 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(): for name, data in items.items():
self._create_parameters(data, name, key) self._create_parameters(data, name, key)
for rule in self._rules: for rule in self._rules:
@@ -147,11 +144,7 @@ class HonCommand:
self.appliance.sync_command_to_params(self.name) self.appliance.sync_command_to_params(self.name)
try: try:
result = await self.api.send_command( result = await self.api.send_command(
self._appliance, self._appliance, self._name, params, ancillary_params
self._name,
params,
ancillary_params,
self._category_name,
) )
if not result: if not result:
_LOGGER.error(result) _LOGGER.error(result)
+2 -18
View File
@@ -24,16 +24,12 @@ class HonAPI:
email: str = "", email: str = "",
password: str = "", password: str = "",
anonymous: bool = False, anonymous: bool = False,
mobile_id: str = "",
refresh_token: str = "",
session: Optional[ClientSession] = None, session: Optional[ClientSession] = None,
) -> None: ) -> None:
super().__init__() super().__init__()
self._email: str = email self._email: str = email
self._password: str = password self._password: str = password
self._anonymous: bool = anonymous self._anonymous: bool = anonymous
self._mobile_id: str = mobile_id
self._refresh_token: str = refresh_token
self._hon_handler: Optional[HonConnectionHandler] = None self._hon_handler: Optional[HonConnectionHandler] = None
self._hon_anonymous_handler: Optional[HonAnonymousConnectionHandler] = None self._hon_anonymous_handler: Optional[HonAnonymousConnectionHandler] = None
self._session: Optional[ClientSession] = session self._session: Optional[ClientSession] = session
@@ -73,10 +69,7 @@ class HonAPI:
).create() ).create()
if not self._anonymous: if not self._anonymous:
self._hon_handler = await HonConnectionHandler( self._hon_handler = await HonConnectionHandler(
self._email, self._email, self._password, self._session
self._password,
session=self._session,
mobile_id=self._mobile_id,
).create() ).create()
return self return self
@@ -197,7 +190,6 @@ class HonAPI:
command: str, command: str,
parameters: Dict[str, Any], parameters: Dict[str, Any],
ancillary_parameters: Dict[str, Any], ancillary_parameters: Dict[str, Any],
program_name: str = "",
) -> bool: ) -> bool:
now: str = datetime.utcnow().isoformat() now: str = datetime.utcnow().isoformat()
data: Dict[str, Any] = { data: Dict[str, Any] = {
@@ -216,8 +208,6 @@ class HonAPI:
"parameters": parameters, "parameters": parameters,
"applianceType": appliance.appliance_type, "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" url: str = f"{const.API_URL}/commands/v1/send"
async with self._hon.post(url, json=data) as response: async with self._hon.post(url, json=data) as response:
json_data: Dict[str, Any] = await response.json() json_data: Dict[str, Any] = await response.json()
@@ -329,12 +319,6 @@ class TestAPI(HonAPI):
command: str, command: str,
parameters: Dict[str, Any], parameters: Dict[str, Any],
ancillary_parameters: Dict[str, Any], ancillary_parameters: Dict[str, Any],
program_name: str = "",
) -> bool: ) -> bool:
_LOGGER.info( _LOGGER.info("%s - %s", str(parameters), str(ancillary_parameters))
"%s - %s - %s",
str(parameters),
str(ancillary_parameters),
str(program_name),
)
return True return True
+2 -4
View File
@@ -199,7 +199,7 @@ class HonAuth:
if access_token := re.findall("access_token=(.*?)&", text): if access_token := re.findall("access_token=(.*?)&", text):
self._auth.access_token = access_token[0] self._auth.access_token = access_token[0]
if refresh_token := re.findall("refresh_token=(.*?)&", text): if refresh_token := re.findall("refresh_token=(.*?)&", text):
self._auth.refresh_token = parse.unquote(refresh_token[0]) self._auth.refresh_token = refresh_token[0]
if id_token := re.findall("id_token=(.*?)&", text): if id_token := re.findall("id_token=(.*?)&", text):
self._auth.id_token = id_token[0] self._auth.id_token = id_token[0]
return bool(access_token and refresh_token and id_token) return bool(access_token and refresh_token and id_token)
@@ -264,9 +264,7 @@ class HonAuth:
except exceptions.HonNoAuthenticationNeeded: except exceptions.HonNoAuthenticationNeeded:
return return
async def refresh(self, refresh_token: str = "") -> bool: async def refresh(self) -> bool:
if refresh_token:
self._auth.refresh_token = refresh_token
params = { params = {
"client_id": const.CLIENT_ID, "client_id": const.CLIENT_ID,
"refresh_token": self._auth.refresh_token, "refresh_token": self._auth.refresh_token,
+3 -2
View File
@@ -1,15 +1,16 @@
import secrets
from typing import Dict from typing import Dict
from pyhon import const from pyhon import const
class HonDevice: class HonDevice:
def __init__(self, mobile_id: str = "") -> None: def __init__(self) -> None:
self._app_version: str = const.APP_VERSION self._app_version: str = const.APP_VERSION
self._os_version: int = const.OS_VERSION self._os_version: int = const.OS_VERSION
self._os: str = const.OS self._os: str = const.OS
self._device_model: str = const.DEVICE_MODEL self._device_model: str = const.DEVICE_MODEL
self._mobile_id: str = mobile_id or const.MOBILE_ID self._mobile_id: str = secrets.token_hex(8)
@property @property
def app_version(self) -> str: def app_version(self) -> str:
+3 -16
View File
@@ -19,18 +19,12 @@ _LOGGER = logging.getLogger(__name__)
class HonConnectionHandler(ConnectionHandler): class HonConnectionHandler(ConnectionHandler):
def __init__( def __init__(
self, self, email: str, password: str, session: Optional[aiohttp.ClientSession] = None
email: str,
password: str,
session: Optional[aiohttp.ClientSession] = None,
mobile_id: str = "",
refresh_token: str = "",
) -> None: ) -> None:
super().__init__(session=session) super().__init__(session=session)
self._device: HonDevice = HonDevice(mobile_id) self._device: HonDevice = HonDevice()
self._email: str = email self._email: str = email
self._password: str = password self._password: str = password
self._refresh_token: str = refresh_token
if not self._email: if not self._email:
raise HonAuthenticationError("An email address must be specified") raise HonAuthenticationError("An email address must be specified")
if not self._password: if not self._password:
@@ -49,17 +43,10 @@ class HonConnectionHandler(ConnectionHandler):
async def create(self) -> Self: async def create(self) -> Self:
await super().create() await super().create()
self._auth = HonAuth( self._auth = HonAuth(self.session, self._email, self._password, self._device)
self.session,
self._email,
self._password,
self._device,
)
return self return self
async def _check_headers(self, headers: Dict[str, str]) -> Dict[str, str]: 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): if not (self.auth.cognito_token and self.auth.id_token):
await self.auth.authenticate() await self.auth.authenticate()
headers["cognito-token"] = self.auth.cognito_token headers["cognito-token"] = self.auth.cognito_token
+4 -5
View File
@@ -6,9 +6,8 @@ CLIENT_ID = (
"3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9." "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9."
"HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6" "HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6"
) )
APP_VERSION = "2.6.5" APP_VERSION = "2.3.5"
OS_VERSION = 999 OS_VERSION = 31
OS = "android" OS = "android"
DEVICE_MODEL = "pyhOn" DEVICE_MODEL = "exynos9820"
USER_AGENT = "Chrome/999.999.999.999" USER_AGENT = "Chrome/110.0.5481.153"
MOBILE_ID = "pyhOn"
+3 -15
View File
@@ -21,8 +21,6 @@ class Hon:
email: Optional[str] = "", email: Optional[str] = "",
password: Optional[str] = "", password: Optional[str] = "",
session: Optional[ClientSession] = None, session: Optional[ClientSession] = None,
mobile_id: str = "",
refresh_token: str = "",
test_data_path: Optional[Path] = None, test_data_path: Optional[Path] = None,
): ):
self._email: Optional[str] = email self._email: Optional[str] = email
@@ -31,8 +29,6 @@ class Hon:
self._appliances: List[HonAppliance] = [] self._appliances: List[HonAppliance] = []
self._api: Optional[HonAPI] = None self._api: Optional[HonAPI] = None
self._test_data_path: Path = test_data_path or Path().cwd() 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: async def __aenter__(self) -> Self:
return await self.create() return await self.create()
@@ -65,11 +61,7 @@ class Hon:
async def create(self) -> Self: async def create(self) -> Self:
self._api = await HonAPI( self._api = await HonAPI(
self.email, self.email, self.password, session=self._session
self.password,
session=self._session,
mobile_id=self._mobile_id,
refresh_token=self._refresh_token,
).create() ).create()
await self.setup() await self.setup()
return self return self
@@ -111,12 +103,8 @@ class Hon:
) )
await self._create_appliance(appliance, self.api) await self._create_appliance(appliance, self.api)
if ( if (
self._test_data_path test_data := self._test_data_path / "hon-test-data" / "test_data"
and ( ).exists() or (test_data := test_data / "test_data").exists():
test_data := self._test_data_path / "hon-test-data" / "test_data"
).exists()
or (test_data := test_data / "..").exists()
):
api = TestAPI(test_data) api = TestAPI(test_data)
for appliance in await api.load_appliances(): for appliance in await api.load_appliances():
await self._create_appliance(appliance, api) await self._create_appliance(appliance, api)
+1 -6
View File
@@ -69,9 +69,4 @@ class HonParameterRange(HonParameter):
@property @property
def values(self) -> List[str]: def values(self) -> List[str]:
result = [] return [str(i) for i in range(int(self.min), int(self.max) + 1, int(self.step))]
i = self.min
while i <= self.max:
result.append(str(i))
i += self.step
return result
+2 -1
View File
@@ -14,7 +14,8 @@ if TYPE_CHECKING:
class Callback(Protocol): # pylint: disable=too-few-public-methods class Callback(Protocol): # pylint: disable=too-few-public-methods
def __call__( def __call__(
self, url: str | URL, *args: Any, **kwargs: Any self, url: str | URL, *args: Any, **kwargs: Any
) -> aiohttp.client._RequestContextManager: ... ) -> aiohttp.client._RequestContextManager:
...
Parameter = Union[ Parameter = Union[
+1 -1
View File
@@ -1,3 +1,3 @@
aiohttp>=3.8.6 aiohttp>=3.8
yarl>=1.8 yarl>=1.8
typing-extensions>=4.8 typing-extensions>=4.8
+5 -5
View File
@@ -1,5 +1,5 @@
black>=22.12 black>=23.11
flake8>=6.0 flake8>=6.1
mypy>=0.991 mypy>=1.7
pylint>=2.15 pylint>=3.0
setuptools>=62.3 setuptools>=68.2
+2 -2
View File
@@ -7,7 +7,7 @@ with open("README.md", "r", encoding="utf-8") as f:
setup( setup(
name="pyhOn", name="pyhOn",
version="0.16.1", version="0.15.12",
author="Andre Basche", author="Andre Basche",
description="Control hOn devices with python", description="Control hOn devices with python",
long_description=long_description, long_description=long_description,
@@ -21,7 +21,7 @@ setup(
packages=find_packages(), packages=find_packages(),
include_package_data=True, include_package_data=True,
python_requires=">=3.10", python_requires=">=3.10",
install_requires=["aiohttp>=3.8.6", "typing-extensions>=4.8", "yarl>=1.8"], install_requires=["aiohttp>=3.8", "typing-extensions>=4.8", "yarl>=1.8"],
classifiers=[ classifiers=[
"Development Status :: 4 - Beta", "Development Status :: 4 - Beta",
"Environment :: Console", "Environment :: Console",