pyhOn/pyhon/api.py
2023-03-13 03:58:45 +01:00

151 lines
6 KiB
Python

import asyncio
import json
import logging
import secrets
from datetime import datetime
from typing import List
import aiohttp as aiohttp
from pyhon import const
from pyhon.auth import HonAuth
from pyhon.device import HonDevice
_LOGGER = logging.getLogger()
class HonConnection:
def __init__(self, email, password, session=None) -> None:
super().__init__()
self._email = email
self._password = password
self._request_headers = {"Content-Type": "application/json"}
self._session = session
self._devices = []
self._mobile_id = secrets.token_hex(8)
async def __aenter__(self):
self._session = aiohttp.ClientSession()
await self.setup()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self._session.close()
@property
def devices(self) -> List[HonDevice]:
return self._devices
@property
async def _headers(self):
if "cognito-token" not in self._request_headers or "id-token" not in self._request_headers:
auth = HonAuth()
if await auth.authorize(self._email, self._password, self._mobile_id):
self._request_headers["cognito-token"] = auth.cognito_token
self._request_headers["id-token"] = auth.id_token
else:
raise PermissionError("Can't Login")
return self._request_headers
async def setup(self):
async with aiohttp.ClientSession() as session:
async with session.get(f"{const.API_URL}/commands/v1/appliance",
headers=await self._headers) as resp:
try:
appliances = (await resp.json())["payload"]["appliances"]
for appliance in appliances:
device = HonDevice(self, appliance)
await asyncio.gather(*[
device.load_attributes(),
device.load_commands(),
device.load_statistics()])
self._devices.append(device)
except json.JSONDecodeError:
_LOGGER.error("No JSON Data after GET: %s", await resp.text())
return False
return True
async def load_commands(self, device: HonDevice):
params = {
"applianceType": device.appliance_type,
"code": device.appliance["code"],
"applianceModelId": device.appliance_model_id,
"firmwareId": device.appliance["eepromId"],
"macAddress": device.mac_address,
"fwVersion": device.appliance["fwVersion"],
"os": const.OS,
"appVersion": const.APP_VERSION,
"series": device.appliance["series"],
}
url = f"{const.API_URL}/commands/v1/retrieve"
async with self._session.get(url, params=params, headers=await self._headers) as response:
result = (await response.json()).get("payload", {})
if not result or result.pop("resultCode") != "0":
return {}
return result
async def command_history(self, device: HonDevice):
url = f"{const.API_URL}/commands/v1/appliance/{device.mac_address}/history"
async with self._session.get(url, headers=await self._headers) as response:
result = await response.json()
if not result or not result.get("payload"):
return {}
return result["payload"]["history"]
async def load_attributes(self, device: HonDevice, loop=False):
params = {
"macAddress": device.mac_address,
"applianceType": device.appliance_type,
"category": "CYCLE"
}
url = f"{const.API_URL}/commands/v1/context"
async with self._session.get(url, params=params, headers=await self._headers) as response:
if response.status == 403 and not loop:
_LOGGER.error("%s - Error %s - %s", url, response.status, await response.text())
self._request_headers.pop("cognito-token", None)
self._request_headers.pop("id-token", None)
return await self.load_attributes(device, loop=True)
return (await response.json()).get("payload", {})
async def load_statistics(self, device: HonDevice):
params = {
"macAddress": device.mac_address,
"applianceType": device.appliance_type
}
url = f"{const.API_URL}/commands/v1/statistics"
async with self._session.get(url, params=params, headers=await self._headers) as response:
return (await response.json()).get("payload", {})
async def send_command(self, device, command, parameters, ancillary_parameters):
now = datetime.utcnow().isoformat()
data = {
"macAddress": device.mac_address,
"timestamp": f"{now[:-3]}Z",
"commandName": command,
"transactionId": f"{device.mac_address}_{now[:-3]}Z",
"applianceOptions": device.commands_options,
"device": {
"mobileId": self._mobile_id,
"mobileOs": const.OS,
"osVersion": const.OS_VERSION,
"appVersion": const.APP_VERSION,
"deviceModel": const.DEVICE_MODEL
},
"attributes": {
"channel": "mobileApp",
"origin": "standardProgram",
"energyLabel": "0"
},
"ancillaryParameters": ancillary_parameters,
"parameters": parameters,
"applianceType": device.appliance_type
}
url = f"{const.API_URL}/commands/v1/send"
async with self._session.post(url, headers=await self._headers, json=data) as resp:
try:
json_data = await resp.json()
except json.JSONDecodeError:
return False
if json_data["payload"]["resultCode"] == "0":
return True
return False