diff --git a/homgarapi/api.py b/homgarapi/api.py index 06e320f..9ab9503 100644 --- a/homgarapi/api.py +++ b/homgarapi/api.py @@ -2,6 +2,7 @@ import hashlib import os from datetime import datetime, timedelta +from typing import Optional, List import requests @@ -25,10 +26,23 @@ def __str__(self): class HomgarApi: - def __init__(self, cache): - self.session = requests.Session() - self.cache = cache - self.base = "https://region3.homgarus.com" + def __init__( + self, + auth_cache: Optional[dict] = None, + api_base_url: str = "https://region3.homgarus.com", + requests_session: requests.Session = None + ): + """ + Create an object for interacting with the Homgar API + :param auth_cache: A dictionary in which authentication information will be stored. + Save this dict on exit and supply it again next time constructing this object to avoid logging in + if a valid token is still present. + :param api_base_url: The base URL for the Homgar API. Omit trailing slash. + :param requests_session: Optional requests lib session to use. New session is created if omitted. + """ + self.session = requests_session or requests.Session() + self.cache = auth_cache or {} + self.base = api_base_url def _request(self, method, url, with_auth=True, headers=None, **kwargs): logger.log(TRACE, "%s %s %s", method, url, kwargs) @@ -52,9 +66,15 @@ def _get_json(self, path, **kwargs): def _post_json(self, path, body, **kwargs): return self._request_json("POST", path, json=body, **kwargs) - def login(self, email, password): + def login(self, email: str, password: str, area_code="31") -> None: + """ + Perform a new login. + :param email: Account e-mail + :param password: Account password + :param area_code: Seems to need to be the phone country code associated with the account, e.g. "31" for NL + """ data = self._post_json("/auth/basic/app/login", { - "areaCode": "31", + "areaCode": area_code, "phoneOrEmail": email, "password": hashlib.md5(password.encode('utf-8')).hexdigest(), "deviceId": binascii.b2a_hex(os.urandom(16)).decode('utf-8') @@ -64,11 +84,23 @@ def login(self, email, password): self.cache['token_expires'] = datetime.utcnow().timestamp() + data.get('tokenExpired') self.cache['refresh_token'] = data.get('refreshToken') - def get_homes(self) -> [HomgarHome]: + def get_homes(self) -> List[HomgarHome]: + """ + Retrieves all HomgarHome objects associated with the logged in account. + Requires first logging in. + :return: List of HomgarHome objects + """ data = self._get_json("/app/member/appHome/list") return [HomgarHome(hid=h.get('hid'), name=h.get('homeName')) for h in data] - def get_devices_for_hid(self, hid: str): + def get_devices_for_hid(self, hid: str) -> List[HomgarHubDevice]: + """ + Retrieves a device tree associated with the home identified by the given hid (home ID). + This function returns a list of hubs associated with the home. Each hub contains associated + subdevices that use the hub as gateway. + :param hid: The home ID to retrieve hubs and associated subdevices for + :return: List of hubs with associated subdevicse + """ data = self._get_json("/app/device/getDeviceByHid", params={"hid": str(hid)}) hubs = [] @@ -114,7 +146,11 @@ def get_device_class(dev_data): return hubs - def get_device_status(self, hub: HomgarHubDevice): + def get_device_status(self, hub: HomgarHubDevice) -> None: + """ + Updates the device status of all subdevices associated with the given hub device. + :param hub: The hub to update + """ data = self._get_json("/app/device/getDeviceStatus", params={"mid": str(hub.mid)}) id_map = {status_id: device for device in [hub, *hub.subdevices] for status_id in device.get_device_status_ids()} @@ -123,9 +159,14 @@ def get_device_status(self, hub: HomgarHubDevice): if device is not None: device.set_device_status(subdevice_status) - def ensure_logged_in(self, email, password): + def ensure_logged_in(self, email: str, password: str, area_code: str = "31") -> None: + """ + Ensures this API object has valid credentials. + Attempts to verify the token stored in the auth cache. If invalid, attempts to login. + See login() for parameter info. + """ if ( self.cache.get('email') != email or datetime.fromtimestamp(self.cache.get('token_expires', 0)) - datetime.utcnow() < timedelta(minutes=60) ): - self.login(email, password) + self.login(email, password, area_code=area_code) diff --git a/homgarapi/devices.py b/homgarapi/devices.py index 26dacc3..14f9245 100644 --- a/homgarapi/devices.py +++ b/homgarapi/devices.py @@ -1,4 +1,5 @@ import re +from typing import List STATS_VALUE_REGEX = re.compile(r'^(\d+)\((\d+)/(\d+)/(\d+)\)') @@ -15,12 +16,21 @@ def _temp_to_mk(f): class HomgarHome: + """ + Represents a home in Homgar. + A home can have a number of hubs, each of which can contain sensors/controllers (subdevices). + """ def __init__(self, hid, name): self.hid = hid self.name = name class HomgarDevice: + """ + Base class for Homgar devices; both hubs and subdevices. + Each device has a model (name and code), name, some identifiers and may have alerts. + """ + FRIENDLY_DESC = "Unknown HomGar device" def __init__(self, model, model_code, name, did, mid, alerts, **kwargs): @@ -37,28 +47,64 @@ def __init__(self, model, model_code, name, did, mid, alerts, **kwargs): def __str__(self): return f"{self.FRIENDLY_DESC} \"{self.name}\" (DID {self.did})" - def get_device_status_ids(self): + def get_device_status_ids(self) -> List[str]: + """ + The response for /app/device/getDeviceStatus contains a subDeviceStatus for each of the subdevices. + This function returns which IDs in the subDeviceStatus apply to this device. + Usually this is just Dxx where xx is the device address, but the hub has some additional special keys. + set_device_status() will be called on this object for all subDeviceStatus entries matching any of the + return IDs. + :return: The subDeviceStatus this device should listen to. + """ return [] - def set_device_status(self, api_obj): + def set_device_status(self, api_obj: dict) -> None: + """ + Called after a call to /app/device/getDeviceStatus with an entry from $.data.subDeviceStatus + that matches one of the IDs returned by get_device_status_ids(). + Should update the device status with the contents of the given API response. + :param api_obj: The $.data.subDeviceStatus API response that should be used to update this device's status + """ if api_obj['id'] == f"D{self.address:02d}": self._parse_status_d_value(api_obj['value']) - def _parse_status_d_value(self, val): + def _parse_status_d_value(self, val: str) -> None: + """ + Parses a $.data.subDeviceStatus[x].value field for an entry with ID 'Dxx' where xx is the device address. + These fields consist of a common part and a device-specific part separated by a ';'. + This call should update the device status. + :param val: Value of the $.data.subDeviceStatus[x].value field to apply + """ general_str, specific_str = val.split(';') self._parse_general_status_d_value(general_str) self._parse_device_specific_status_d_value(specific_str) - def _parse_general_status_d_value(self, s): - # unknowns are all '1' in my case, possibly battery state + connected state + def _parse_general_status_d_value(self, s: str): + """ + Parses the part of a $.data.subDeviceStatus[x].value field before the ';' character, + which has the same format for all subdevices. It has three ','-separated fields. The first and last fields + are always '1' in my case, I presume it's to do with battery state / connection state. + The second field is the RSSI in dBm. + :param s: The value to parse and apply + """ unknown_1, rf_rssi, unknown_2 = s.split(',') self.rf_rssi = int(rf_rssi) - def _parse_device_specific_status_d_value(self, s): + def _parse_device_specific_status_d_value(self, s: str): + """ + Parses the part of a $.data.subDeviceStatus[x].value field after the ';' character, + which is in a device-specific format. + Should update the device state. + :param s: The value to parse and apply + """ raise NotImplementedError() class HomgarHubDevice(HomgarDevice): + """ + A hub acts as a gateway for sensors and actuators (subdevices). + A home contains an arbitrary number of hubs, each of which contains an arbitrary number of subdevices. + """ def __init__(self, subdevices, **kwargs): super().__init__(**kwargs) self.address = 1 @@ -72,6 +118,10 @@ def _parse_device_specific_status_d_value(self, s): class HomgarSubDevice(HomgarDevice): + """ + A subdevice is a device that is associated with a hub. + It can be a sensor or an actuator. + """ def __init__(self, address, port_number, **kwargs): super().__init__(**kwargs) self.address = address # device address within the sensor network @@ -124,8 +174,13 @@ def set_device_status(self, api_obj): super().set_device_status(api_obj) def _parse_device_specific_status_d_value(self, s): - # 781(781/723/1),52(64/50/1),P=10213(10222/10205/1), - # temp[.1F](day-max/day-min/trend?),humidity[%](day-max/day-min/trend?),P=pressure[Pa](day-max/day-min/trend?), + """ + Observed example value: + 781(781/723/1),52(64/50/1),P=10213(10222/10205/1), + + Deduced meaning: + temp[.1F](day-max/day-min/trend?),humidity[%](day-max/day-min/trend?),P=pressure[Pa](day-max/day-min/trend?), + """ temp_str, hum_str, press_str, *_ = s.split(',') self.temp_mk_current, self.temp_mk_daily_max, self.temp_mk_daily_min, self.temp_trend = [_temp_to_mk(v) for v in _parse_stats_value(temp_str)] self.hum_current, self.hum_daily_max, self.hum_daily_min, self.hum_trend = _parse_stats_value(hum_str) @@ -149,8 +204,13 @@ def __init__(self, **kwargs): self.light_lux_current = None def _parse_device_specific_status_d_value(self, s): - # 766,52,G=31351 - # temp[.1F],soil-moisture[%],G=light[.1lux] + """ + Observed example value: + 766,52,G=31351 + + Deduced meaning: + temp[.1F],soil-moisture[%],G=light[.1lux] + """ temp_str, moist_str, light_str = s.split(',') self.temp_mk_current = _temp_to_mk(temp_str) self.moist_percent_current = int(moist_str) @@ -175,8 +235,13 @@ def __init__(self, **kwargs): self.rainfall_mm_total = None def _parse_device_specific_status_d_value(self, s): - # R=270(0/0/270) - # R=total?[.1mm](hour?[.1mm]/24hours?[.1mm]/7days?[.1mm]) + """ + Observed example value: + R=270(0/0/270) + + Deduced meaning: + R=total?[.1mm](hour?[.1mm]/24hours?[.1mm]/7days?[.1mm]) + """ self.rainfall_mm_total, self.rainfall_mm_hour, self.rainfall_mm_daily, self.rainfall_mm_7days = [.1*v for v in _parse_stats_value(s[2:])] def __str__(self): @@ -202,8 +267,13 @@ def __init__(self, **kwargs): self.hum_trend = None def _parse_device_specific_status_d_value(self, s): - # 755(1020/588/1),54(91/24/1), - # temp[.1F](day-max/day-min/trend?),humidity[%](day-max/day-min/trend?) + """ + Observed example value: + 755(1020/588/1),54(91/24/1), + + Deduced meaning: + temp[.1F](day-max/day-min/trend?),humidity[%](day-max/day-min/trend?) + """ temp_str, hum_str, *_ = s.split(',') self.temp_mk_current, self.temp_mk_daily_max, self.temp_mk_daily_min, self.temp_trend = [_temp_to_mk(v) for v in _parse_stats_value(temp_str)] self.hum_current, self.hum_daily_max, self.hum_daily_min, self.hum_trend = _parse_stats_value(hum_str) @@ -220,9 +290,15 @@ class RainPoint2ZoneTimer(HomgarSubDevice): FRIENDLY_DESC = "2-Zone Water Timer" def _parse_device_specific_status_d_value(self, s): - # 0,9,0,0,0,0|0,1291,0,0,0,0 - # left|right, each: - # ?,last-usage[.1l],?,?,?,? + """ + TODO deduce meaning of these fields. + Observed example value: + 0,9,0,0,0,0|0,1291,0,0,0,0 + + What we know so far: + left/right zone separated by '|' character + fields for each zone: ?,last-usage[.1l],?,?,?,? + """ pass diff --git a/homgarapi/logutil.py b/homgarapi/logutil.py index d6bbe91..2746bc7 100644 --- a/homgarapi/logutil.py +++ b/homgarapi/logutil.py @@ -4,5 +4,5 @@ TRACE = logging.DEBUG - 1 -def get_logger(file: str): - return logging.getLogger(Path(__file__).stem) +def get_logger(file: str) -> logging.Logger: + return logging.getLogger(Path(file).stem) diff --git a/main.py b/main.py index 79562f3..379c70c 100644 --- a/main.py +++ b/main.py @@ -16,7 +16,7 @@ def demo(api: HomgarApi, config): for home in api.get_homes(): print(f"({home.hid}) {home.name}:") - for hub in api.get_devices_for_home(home.hid): + for hub in api.get_devices_for_hid(home.hid): print(f" - {hub}") api.get_device_status(hub) for subdevice in hub.subdevices: