Merge pending changes in order to fix release process

Author: ldotlopez

hnap/__init__.py

@@ -17,10 +17,10 @@
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
 # USA.
 
-
 from .devices import Camera, DeviceFactory, Motion, Router, Siren, Sound, Water
 from .soapclient import AuthenticationError, MethodCallError
 
+
 __all__ = [
     "AuthenticationError",
     "MethodCallError",

hnap/devices.py

@@ -20,14 +20,71 @@
 
 import datetime
 import enum
+import logging
 
 from .soapclient import MethodCallError, SoapClient
 
+_LOGGER = logging.getLogger(__name__)
 
-def DeviceFactory(hostname, password, username="Admin", port=80, **kwargs):
-    device = SoapClient(hostname, password, username, port)
-    device.authenticate()
-    info = device.device_info()
+
+class Sound(enum.Enum):
+    EMERGENCY = 1
+    FIRE = 2
+    AMBULANCE = 3
+    POLICE = 4
+    DOOR_CHIME = 5
+    BEEP = 6
+
+    @classmethod
+    def fromstring(cls, s):
+        s = s.upper()
+        for c in ["-", " ", "."]:
+            s = s.replace(c, "_")
+
+        return getattr(cls, s)
+
+
+class Device:
+    def __init__(
+        self,
+        *,
+        client=None,
+        hostname=None,
+        password=None,
+        username="Admin",
+        port=80,
+    ):
+        self.client = client or SoapClient(
+            hostname=hostname, password=password, username=username, port=port
+        )
+        self._info = None
+
+    def authenticate(self):
+        self.client.authenticate()
+        info = dict(self.client.call("GetDeviceSettings"))
+        for k in ["@xmlns", "SOAPActions", "GetDeviceSettingsResult"]:
+            info.pop(k, None)
+
+        try:
+            info["ModuleTypes"] = info["ModuleTypes"]["string"]
+        except KeyError:
+            pass
+
+        self._info = info
+
+    @property
+    def info(self):
+        return self._info
+
+
+def DeviceFactory(
+    *, client=None, hostname=None, password=None, username="Admin", port=80
+):
+    client = client or SoapClient(
+        hostname=hostname, password=password, username=username, port=port
+    )
+    client.authenticate()
+    info = client.device_info()
 
     module_types = info["ModuleTypes"]
     if not isinstance(module_types, list):
@@ -43,47 +100,40 @@ def DeviceFactory(hostname, password, username="Admin", port=80, **kwargs):
     else:
         raise TypeError(module_types)
 
-    return cls(hostname, password, username=username, port=port, **kwargs)
-
-
-class _Device(SoapClient):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self._info = None
-
-    def authenticate(self):
-        super().authenticate()
-        self._info = self.device_info()
-
-    @property
-    def info(self):
-        return self._info
+    return cls(client=client)
 
 
-class Camera(_Device):
+class Camera(Device):
     pass
 
 
-class Motion(_Device):
-    def __init__(self, *args, delta=30, **kwargs):
+class Motion(Device):
+    def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.delta = delta
+        self._delta = None
 
-    def login(self):
-        super().login()
+    @property
+    def delta(self):
+        return self._delta
 
-        # Auto-adjust delta
-        res = self.call(
-            "SetMotionDetectorSettings", ModuleID=1, Backoff=self.delta
+    @delta.setter
+    def delta(self, seconds):
+        self.client.call(
+            "SetMotionDetectorSettings", ModuleID=1, Backoff=self._delta
         )
-        res = self.call("GetMotionDetectorSettings", ModuleID=1)
+        _LOGGER.warning("set delta property has no effect")
+
+    def authenticate(self):
+        super().authenticate()
+
+        res = self.client.call("GetMotionDetectorSettings", ModuleID=1)
         try:
-            self.delta = int(res["Backoff"])
-        except (ValueError, TypeError):
-            self.delta = 30
+            self._delta = int(res["Backoff"])
+        except (ValueError, TypeError, KeyError):
+            _LOGGER.warning("Unable to get delta from device")
 
     def get_latest_detection(self):
-        res = self.call("GetLatestDetection", ModuleID=1)
+        res = self.client.call("GetLatestDetection", ModuleID=1)
         return datetime.datetime.fromtimestamp(float(res["LatestDetectTime"]))
 
     def is_active(self):
@@ -93,7 +143,7 @@ def is_active(self):
         return delta <= self.delta
 
 
-class Router(SoapClient):
+class Router(Device):
     # NOT tested
     # See https://github.com/waffelheld/dlink-device-tracker/blob/master/custom_components/dlink_device_tracker/dlink_hnap.py#L95
 
@@ -116,30 +166,15 @@ def get_clients(self):
         return ret
 
 
-class Sound(enum.Enum):
-    EMERGENCY = 1
-    FIRE = 2
-    AMBULANCE = 3
-    POLICE = 4
-    DOOR_CHIME = 5
-    BEEP = 6
-
-    @classmethod
-    def fromstring(cls, s):
-        s = s.upper()
-        for c in ["-", " ", "."]:
-            s = s.replace(c, "_")
-
-        return getattr(cls, s)
-
-
-class Siren(_Device):
+class Siren(Device):
     def is_playing(self):
-        res = self.call("GetSirenAlarmSettings", ModuleID=1, Controller=1)
+        res = self.client.call(
+            "GetSirenAlarmSettings", ModuleID=1, Controller=1
+        )
         return res["IsSounding"] == "true"
 
     def play(self, sound=Sound.EMERGENCY, volume=100, duration=60):
-        ret = self.call(
+        ret = self.client.call(
             "SetSoundPlay",
             ModuleID=1,
             Controller=1,
@@ -151,16 +186,18 @@ def play(self, sound=Sound.EMERGENCY, volume=100, duration=60):
             raise MethodCallError(f"Unable to play. Response: {ret}")
 
     def beep(self, volume=100, duration=1):
-        return self.play(sound=Sound.BEEP, duration=duration, volume=volume)
+        return self.client.play(
+            sound=Sound.BEEP, duration=duration, volume=volume
+        )
 
     def stop(self):
-        ret = self.call("SetAlarmDismissed", ModuleID=1, Controller=1)
+        ret = self.client.call("SetAlarmDismissed", ModuleID=1, Controller=1)
 
         if ret["SetAlarmDismissedResult"] != "OK":
             raise MethodCallError(f"Unable to stop. Response: {ret}")
 
 
-class Water(_Device):
+class Water(Device):
     def is_active(self):
         ret = self.call("GetWaterDetectorState", ModuleID=1)
         return ret.get("IsWater") == "true"

hnap/soapclient.py

@@ -18,6 +18,7 @@
 # USA.
 
 
+import functools
 import hashlib
 import hmac
 import logging
@@ -48,19 +49,24 @@ class SoapClient:
         "cookie": "",
         "private_key": "",
         "public_key": "",
-        "pin": None,
+        "password": None,
         "result": "",
         "url": "http://{hostname}:{port}/HNAP1",
         "username": None,
     }
 
-    def __init__(self, hostname, pin, username="admin", port=80):
+    def __init__(self, hostname, password, username="admin", port=80):
         self.HNAP_AUTH = self.HNAP_AUTH.copy()
         self.HNAP_AUTH["url"] = self.HNAP_AUTH["url"].format(
             hostname=hostname, port=port
         )
         self.HNAP_AUTH["username"] = username
-        self.HNAP_AUTH["pin"] = pin
+        self.HNAP_AUTH["password"] = password
+        self._authenticated = False
+
+    @property
+    def authenticated(self):
+        return self._authenticated
 
     def _build_method_envelope(self, method, **parameters):
         parameters_xml = "\n".join(
@@ -95,7 +101,7 @@ def _save_login_result(self, body):
             self.HNAP_AUTH[key] = elements[0].firstChild.nodeValue
 
         self.HNAP_AUTH["private_key"] = hex_hmac_md5(
-            self.HNAP_AUTH["public_key"] + self.HNAP_AUTH["pin"],
+            self.HNAP_AUTH["public_key"] + self.HNAP_AUTH["password"],
             self.HNAP_AUTH["challenge"],
         ).upper()
 
@@ -162,9 +168,11 @@ def authenticate(self):
             + self.HNAP_LOGIN_METHOD
             + '"',
         }
+
         resp = requests.request(
             method=method, url=url, data=data, headers=headers
         )
+
         if resp.status_code != 200:
             raise AuthenticationError(
                 f"Invalid response while login-in: {resp.status_code} "
@@ -189,6 +197,8 @@ def authenticate(self):
         if res["LoginResult"] != "success":
             raise AuthenticationError(res["LoginResult"])
 
+        self._authenticated = True
+
     def device_info(self):
         info = dict(self.call("GetDeviceSettings"))
         for k in ["@xmlns", "SOAPActions", "GetDeviceSettingsResult"]:
@@ -228,3 +238,13 @@ class AuthenticationError(ClientError):
 
 class MethodCallError(ClientError):
     pass
+
+
+def auth_required(fn):
+    @functools.wraps(fn)
+    def _wrap(self, *args, **kwargs):
+        if not self._authenticated:
+            self.authenticated()
+        return fn(self, *args, **kwargs)
+
+    return _wrap

setup.cfg

@@ -2,7 +2,7 @@
 
 [metadata]
 name = hnap
-version = 0.0.5
+version = 0.0.7
 author = Luis López
 author_email = luis@cuarentaydos.com
 description = Python clients for HNAP devices