mirror of
https://github.com/Andre0512/pyhOn.git
synced 2026-06-21 22:48:11 +02:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| cf137dd176 |
@@ -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
@@ -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
@@ -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:
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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:
|
||||||
|
|||||||
@@ -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
@@ -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
@@ -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)
|
||||||
|
|||||||
@@ -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
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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",
|
||||||
|
|||||||
Reference in New Issue
Block a user