Add documentation

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)

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
 
 

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)

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: