"""MEL API access.""" from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from aiohttp import ClientResponseError, ContentTypeError from aiohttp import ClientSession BASE_URL = "https://app.melcloud.com/Mitsubishi.Wifi.Client" def _headers(token: str) -> Dict[str, str]: return { "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:73.0) " "Gecko/20100101 Firefox/73.0", "Accept": "application/json, text/javascript, */*; q=0.01", "Accept-Language": "en-US,en;q=0.5", "Accept-Encoding": "gzip, deflate, br", "X-MitsContextKey": token, "X-Requested-With": "XMLHttpRequest", "Cookie": "policyaccepted=true", } async def _do_login(_session: ClientSession, email: str, password: str): body = { "Email": email, "Password": password, "Language": 0, "AppVersion": "1.19.1.1", "Persist": True, "CaptchaResponse": None, } async with _session.post( f"{BASE_URL}/Login/ClientLogin", json=body, raise_for_status=True ) as resp: return await resp.json() async def login( email: str, password: str, session: Optional[ClientSession] = None, *, user_update_interval: Optional[timedelta] = None, conf_update_interval: Optional[timedelta] = None, device_set_debounce: Optional[timedelta] = None, ): """Login using email and password.""" if session: response = await _do_login(session, email, password) else: async with ClientSession() as _session: response = await _do_login(_session, email, password) return Client( response.get("LoginData").get("ContextKey"), session, user_update_interval=user_update_interval, conf_update_interval=conf_update_interval, device_set_debounce=device_set_debounce, ) class Client: """MELCloud client. Please do not use this class directly. It is better to use the get_devices method exposed by the __init__.py. """ def __init__( self, token: str, session: Optional[ClientSession] = None, *, user_update_interval=timedelta(minutes=5), conf_update_interval=timedelta(seconds=59), device_set_debounce=timedelta(seconds=1), ): """Initialize MELCloud client.""" self._token = token if session: self._session = session self._managed_session = False else: self._session = ClientSession() self._managed_session = True self._user_update_interval = user_update_interval self._conf_update_interval = conf_update_interval self._device_set_debounce = device_set_debounce self._last_user_update = None self._last_conf_update = None self._device_confs: List[Dict[str, Any]] = [] self._account: Optional[Dict[str, Any]] = None @property def token(self) -> str: """Return currently used token.""" return self._token @property def device_confs(self) -> List[Dict[Any, Any]]: """Return device configurations.""" return self._device_confs @property def account(self) -> Optional[Dict[Any, Any]]: """Return account.""" return self._account async def _fetch_user_details(self): """Fetch user details.""" async with self._session.get( f"{BASE_URL}/User/GetUserDetails", headers=_headers(self._token), raise_for_status=True, ) as resp: self._account = await resp.json() async def _fetch_device_confs(self): """Fetch all configured devices.""" url = f"{BASE_URL}/User/ListDevices" async with self._session.get( url, headers=_headers(self._token), raise_for_status=True ) as resp: entries = await resp.json() new_devices = [] for entry in entries: new_devices = new_devices + entry["Structure"]["Devices"] for area in entry["Structure"]["Areas"]: new_devices = new_devices + area["Devices"] for floor in entry["Structure"]["Floors"]: new_devices = new_devices + floor["Devices"] for area in floor["Areas"]: new_devices = new_devices + area["Devices"] visited = set() self._device_confs = [ d for d in new_devices if d["DeviceID"] not in visited and not visited.add(d["DeviceID"]) ] async def update_confs(self): """Update device_confs and account. Calls are rate limited to allow Device instances to freely poll their own state while refreshing the device_confs list and account. """ now = datetime.now() if ( self._last_conf_update is None or now - self._last_conf_update > self._conf_update_interval ): await self._fetch_device_confs() self._last_conf_update = now if ( self._last_user_update is None or now - self._last_user_update > self._user_update_interval ): await self._fetch_user_details() self._last_user_update = now async def fetch_device_units(self, device) -> Optional[Dict[Any, Any]]: """Fetch unit information for a device. User provided info such as indoor/outdoor unit model names and serial numbers. If the request returns 403, then ignore it """ try: async with self._session.post( f"{BASE_URL}/Device/ListDeviceUnits", headers=_headers(self._token), json={"deviceId": device.device_id} ) as resp: resp.raise_for_status() try: data = await resp.json() except ContentTypeError: return None if isinstance(data, dict): return data return None except ClientResponseError as e: if e.status == 403: return None raise async def fetch_device_state(self, device) -> Optional[Dict[Any, Any]]: """Fetch state information of a device. This method should not be called more than once a minute. Rate limiting is left to the caller. """ device_id = device.device_id building_id = device.building_id async with self._session.get( f"{BASE_URL}/Device/Get?id={device_id}&buildingID={building_id}", headers=_headers(self._token), raise_for_status=True, ) as resp: return await resp.json() async def fetch_energy_report(self, device) -> Optional[Dict[Any, Any]]: """Fetch energy report containing today and 1-2 days from the past. If the request returns 403, then ignore it""" device_id = device.device_id from_str = (datetime.today() - timedelta(days=2)).strftime("%Y-%m-%d") to_str = (datetime.today() + timedelta(days=2)).strftime("%Y-%m-%d") try: async with self._session.post( f"{BASE_URL}/EnergyCost/Report", headers=_headers(self._token), json={ "DeviceId": device_id, "UseCurrency": False, "FromDate": f"{from_str}T00:00:00", "ToDate": f"{to_str}T00:00:00" } ) as resp: resp.raise_for_status() try: data = await resp.json() except ContentTypeError: return None if isinstance(data, dict): return data return None except ClientResponseError as e: if e.status == 403: return None raise async def set_device_state(self, device): """Update device state. This method is as dumb as it gets. Device is responsible for updating the state and managing EffectiveFlags. """ device_type = device.get("DeviceType") if device_type == 0: setter = "SetAta" elif device_type == 1: setter = "SetAtw" elif device_type == 3: setter = "SetErv" else: raise ValueError(f"Unsupported device type [{device_type}]") async with self._session.post( f"{BASE_URL}/Device/{setter}", headers=_headers(self._token), json=device, raise_for_status=True, ) as resp: return await resp.json()