mirror of https://github.com/Andre0512/pyhOn.git
Compare commits
9 Commits
3acd12f345
...
6162cd3f98
Author | SHA1 | Date |
---|---|---|
Andre Basche | 6162cd3f98 | |
Andre Basche | e225a850bc | |
Andre Basche | c4e8038652 | |
Andre Basche | 767cbe35de | |
Andre Basche | df27729c2e | |
Andre Basche | a5595f47bf | |
Andre Basche | 29c936b045 | |
Andre Basche | 2a74000ff3 | |
Andre Basche | 8a46bd4b53 |
|
@ -4,6 +4,6 @@ disable=missing-docstring
|
|||
|
||||
[FORMAT]
|
||||
|
||||
max-args=6
|
||||
max-args=7
|
||||
max-attributes=8
|
||||
max-line-length=88
|
||||
|
|
|
@ -38,9 +38,7 @@ def get_arguments() -> Dict[str, Any]:
|
|||
"translate", help="language (de, en, fr...)", metavar="LANGUAGE"
|
||||
)
|
||||
translation.add_argument("--json", help="print as json", action="store_true")
|
||||
parser.add_argument(
|
||||
"-i", "--import", help="import pyhon data", nargs="?", default=Path().cwd()
|
||||
)
|
||||
parser.add_argument("-i", "--import", help="import pyhon data", nargs="?")
|
||||
return vars(parser.parse_args())
|
||||
|
||||
|
||||
|
@ -73,9 +71,8 @@ async def main() -> None:
|
|||
if language := args.get("translate"):
|
||||
await translate(language, json_output=args.get("json", ""))
|
||||
return
|
||||
async with Hon(
|
||||
*get_login_data(args), test_data_path=Path(args.get("import", ""))
|
||||
) as hon:
|
||||
test_data_path = Path(path) if (path := args.get("import", "")) else None
|
||||
async with Hon(*get_login_data(args), test_data_path=test_data_path) as hon:
|
||||
for device in hon.appliances:
|
||||
if args.get("export"):
|
||||
anonymous = args.get("anonymous", False)
|
||||
|
|
|
@ -72,12 +72,10 @@ class HonAppliance:
|
|||
return self.info[item]
|
||||
|
||||
@overload
|
||||
def get(self, item: str, default: None = None) -> Any:
|
||||
...
|
||||
def get(self, item: str, default: None = None) -> Any: ...
|
||||
|
||||
@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:
|
||||
try:
|
||||
|
|
|
@ -109,7 +109,7 @@ class HonCommandLoader:
|
|||
categories: Optional[Dict[str, "HonCommand"]] = None,
|
||||
category_name: str = "",
|
||||
) -> Optional[HonCommand]:
|
||||
"""Try to crate HonCommand object"""
|
||||
"""Try to create HonCommand object"""
|
||||
if not isinstance(data, dict):
|
||||
self._additional_data[command_name] = data
|
||||
return None
|
||||
|
|
|
@ -24,12 +24,16 @@ class HonAPI:
|
|||
email: str = "",
|
||||
password: str = "",
|
||||
anonymous: bool = False,
|
||||
mobile_id: str = "",
|
||||
refresh_token: str = "",
|
||||
session: Optional[ClientSession] = None,
|
||||
) -> None:
|
||||
super().__init__()
|
||||
self._email: str = email
|
||||
self._password: str = password
|
||||
self._anonymous: bool = anonymous
|
||||
self._mobile_id: str = mobile_id
|
||||
self._refresh_token: str = refresh_token
|
||||
self._hon_handler: Optional[HonConnectionHandler] = None
|
||||
self._hon_anonymous_handler: Optional[HonAnonymousConnectionHandler] = None
|
||||
self._session: Optional[ClientSession] = session
|
||||
|
@ -69,7 +73,10 @@ class HonAPI:
|
|||
).create()
|
||||
if not self._anonymous:
|
||||
self._hon_handler = await HonConnectionHandler(
|
||||
self._email, self._password, self._session
|
||||
self._email,
|
||||
self._password,
|
||||
session=self._session,
|
||||
mobile_id=self._mobile_id,
|
||||
).create()
|
||||
return self
|
||||
|
||||
|
|
|
@ -199,7 +199,7 @@ class HonAuth:
|
|||
if access_token := re.findall("access_token=(.*?)&", text):
|
||||
self._auth.access_token = access_token[0]
|
||||
if refresh_token := re.findall("refresh_token=(.*?)&", text):
|
||||
self._auth.refresh_token = refresh_token[0]
|
||||
self._auth.refresh_token = parse.unquote(refresh_token[0])
|
||||
if id_token := re.findall("id_token=(.*?)&", text):
|
||||
self._auth.id_token = id_token[0]
|
||||
return bool(access_token and refresh_token and id_token)
|
||||
|
@ -264,7 +264,9 @@ class HonAuth:
|
|||
except exceptions.HonNoAuthenticationNeeded:
|
||||
return
|
||||
|
||||
async def refresh(self) -> bool:
|
||||
async def refresh(self, refresh_token: str = "") -> bool:
|
||||
if refresh_token:
|
||||
self._auth.refresh_token = refresh_token
|
||||
params = {
|
||||
"client_id": const.CLIENT_ID,
|
||||
"refresh_token": self._auth.refresh_token,
|
||||
|
|
|
@ -1,16 +1,15 @@
|
|||
import secrets
|
||||
from typing import Dict
|
||||
|
||||
from pyhon import const
|
||||
|
||||
|
||||
class HonDevice:
|
||||
def __init__(self) -> None:
|
||||
def __init__(self, mobile_id: str = "") -> None:
|
||||
self._app_version: str = const.APP_VERSION
|
||||
self._os_version: int = const.OS_VERSION
|
||||
self._os: str = const.OS
|
||||
self._device_model: str = const.DEVICE_MODEL
|
||||
self._mobile_id: str = secrets.token_hex(8)
|
||||
self._mobile_id: str = mobile_id or const.MOBILE_ID
|
||||
|
||||
@property
|
||||
def app_version(self) -> str:
|
||||
|
|
|
@ -19,12 +19,18 @@ _LOGGER = logging.getLogger(__name__)
|
|||
|
||||
class HonConnectionHandler(ConnectionHandler):
|
||||
def __init__(
|
||||
self, email: str, password: str, session: Optional[aiohttp.ClientSession] = None
|
||||
self,
|
||||
email: str,
|
||||
password: str,
|
||||
session: Optional[aiohttp.ClientSession] = None,
|
||||
mobile_id: str = "",
|
||||
refresh_token: str = "",
|
||||
) -> None:
|
||||
super().__init__(session=session)
|
||||
self._device: HonDevice = HonDevice()
|
||||
self._device: HonDevice = HonDevice(mobile_id)
|
||||
self._email: str = email
|
||||
self._password: str = password
|
||||
self._refresh_token: str = refresh_token
|
||||
if not self._email:
|
||||
raise HonAuthenticationError("An email address must be specified")
|
||||
if not self._password:
|
||||
|
@ -43,10 +49,17 @@ class HonConnectionHandler(ConnectionHandler):
|
|||
|
||||
async def create(self) -> Self:
|
||||
await super().create()
|
||||
self._auth = HonAuth(self.session, self._email, self._password, self._device)
|
||||
self._auth = HonAuth(
|
||||
self.session,
|
||||
self._email,
|
||||
self._password,
|
||||
self._device,
|
||||
)
|
||||
return self
|
||||
|
||||
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):
|
||||
await self.auth.authenticate()
|
||||
headers["cognito-token"] = self.auth.cognito_token
|
||||
|
|
|
@ -6,8 +6,9 @@ CLIENT_ID = (
|
|||
"3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9."
|
||||
"HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6"
|
||||
)
|
||||
APP_VERSION = "2.4.7"
|
||||
OS_VERSION = 31
|
||||
APP_VERSION = "2.6.5"
|
||||
OS_VERSION = 999
|
||||
OS = "android"
|
||||
DEVICE_MODEL = "exynos9820"
|
||||
USER_AGENT = "Chrome/110.0.5481.153"
|
||||
DEVICE_MODEL = "pyhOn"
|
||||
USER_AGENT = "Chrome/999.999.999.999"
|
||||
MOBILE_ID = "pyhOn"
|
||||
|
|
18
pyhon/hon.py
18
pyhon/hon.py
|
@ -21,6 +21,8 @@ class Hon:
|
|||
email: Optional[str] = "",
|
||||
password: Optional[str] = "",
|
||||
session: Optional[ClientSession] = None,
|
||||
mobile_id: str = "",
|
||||
refresh_token: str = "",
|
||||
test_data_path: Optional[Path] = None,
|
||||
):
|
||||
self._email: Optional[str] = email
|
||||
|
@ -29,6 +31,8 @@ class Hon:
|
|||
self._appliances: List[HonAppliance] = []
|
||||
self._api: Optional[HonAPI] = None
|
||||
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:
|
||||
return await self.create()
|
||||
|
@ -61,7 +65,11 @@ class Hon:
|
|||
|
||||
async def create(self) -> Self:
|
||||
self._api = await HonAPI(
|
||||
self.email, self.password, session=self._session
|
||||
self.email,
|
||||
self.password,
|
||||
session=self._session,
|
||||
mobile_id=self._mobile_id,
|
||||
refresh_token=self._refresh_token,
|
||||
).create()
|
||||
await self.setup()
|
||||
return self
|
||||
|
@ -103,8 +111,12 @@ class Hon:
|
|||
)
|
||||
await self._create_appliance(appliance, self.api)
|
||||
if (
|
||||
test_data := self._test_data_path / "hon-test-data" / "test_data"
|
||||
).exists() or (test_data := test_data / "test_data").exists():
|
||||
self._test_data_path
|
||||
and (
|
||||
test_data := self._test_data_path / "hon-test-data" / "test_data"
|
||||
).exists()
|
||||
or (test_data := test_data / "..").exists()
|
||||
):
|
||||
api = TestAPI(test_data)
|
||||
for appliance in await api.load_appliances():
|
||||
await self._create_appliance(appliance, api)
|
||||
|
|
|
@ -69,4 +69,9 @@ class HonParameterRange(HonParameter):
|
|||
|
||||
@property
|
||||
def values(self) -> List[str]:
|
||||
return [str(i) for i in range(int(self.min), int(self.max) + 1, int(self.step))]
|
||||
result = []
|
||||
i = self.min
|
||||
while i < self.max:
|
||||
i += self.step
|
||||
result.append(str(i))
|
||||
return result
|
||||
|
|
|
@ -14,8 +14,7 @@ if TYPE_CHECKING:
|
|||
class Callback(Protocol): # pylint: disable=too-few-public-methods
|
||||
def __call__(
|
||||
self, url: str | URL, *args: Any, **kwargs: Any
|
||||
) -> aiohttp.client._RequestContextManager:
|
||||
...
|
||||
) -> aiohttp.client._RequestContextManager: ...
|
||||
|
||||
|
||||
Parameter = Union[
|
||||
|
|
|
@ -1,3 +1,3 @@
|
|||
aiohttp>=3.8
|
||||
aiohttp>=3.8.6
|
||||
yarl>=1.8
|
||||
typing-extensions>=4.8
|
||||
|
|
4
setup.py
4
setup.py
|
@ -7,7 +7,7 @@ with open("README.md", "r", encoding="utf-8") as f:
|
|||
|
||||
setup(
|
||||
name="pyhOn",
|
||||
version="0.15.15",
|
||||
version="0.16.0",
|
||||
author="Andre Basche",
|
||||
description="Control hOn devices with python",
|
||||
long_description=long_description,
|
||||
|
@ -21,7 +21,7 @@ setup(
|
|||
packages=find_packages(),
|
||||
include_package_data=True,
|
||||
python_requires=">=3.10",
|
||||
install_requires=["aiohttp>=3.8", "typing-extensions>=4.8", "yarl>=1.8"],
|
||||
install_requires=["aiohttp>=3.8.6", "typing-extensions>=4.8", "yarl>=1.8"],
|
||||
classifiers=[
|
||||
"Development Status :: 4 - Beta",
|
||||
"Environment :: Console",
|
||||
|
|
Loading…
Reference in New Issue