Add nuki bridge backend

Add a backend to support nuki bridges

Signed-off-by: Thomas Schmid <tom@lfence.de>

pydoorlock/Doorlock.py

@@ -24,6 +24,7 @@ from threading import Thread
 from time import sleep
 from os.path import join
 from pydoorlock.AvrDoorlock import AvrDoorlockBackend
+from pydoorlock.NukiBridge import NukiBridge
 from pydoorlock.SimulationBackend import SimulationBackend
 
 from .Config import Config
@@ -134,6 +135,10 @@ class DoorHandler:
 
         if backend_type == "avr":
             self.backend = AvrDoorlockBackend(self)
+        elif backend_type == "nuki":
+            self.backend = NukiBridge(cfg.str("NUKI_ENDPOINT", "backend"),
+                                      cfg.str("NUKI_APITOKEN", "backend"),
+                                      cfg.str("NUKI_DEVICE", "backend"))
         elif backend_type == "simulation":
             self.backend = SimulationBackend(self)
         else:

pydoorlock/NukiBridge.py

@@ -0,0 +1,127 @@
+import json
+import requests
+from urllib.parse import urljoin
+from .Door import DoorState
+from .DoorlockBackend import DoorlockBackend
+import logging
+import threading
+import time
+from typing import Set
+
+log = logging.getLogger(__name__)
+
+class NukiBridgeDevice():
+    """
+    class to interface with a nuki smartlock
+    """
+    def __init__(self, endpoint: str, apitoken: str, device_name: str) -> None:
+        self.endpoint = endpoint
+        self.apitoken = apitoken
+
+        devices = self.list()
+
+        if devices is None:
+            log.error("Unable to get device id")
+            raise RuntimeError("unable to get nuki device id")
+        for device in devices:
+            if device["name"] == device_name:
+                self.device_id = device["nukiId"]
+    
+    def get_device_id(self):
+        return self.device_id
+
+    def make_request(self, endpoint, parameters: dict =  {}):
+        url = urljoin(self.endpoint, endpoint)
+        parameters["token"] = self.apitoken
+
+        try:
+            return requests.get(url, parameters)
+        except Exception as e:
+            log.error(f"{e}")
+            return None
+        
+    def info(self):
+        response = self.make_request("info")
+        return response
+
+    def list(self):
+        response = self.make_request("list")
+        
+        if response is None:
+            return None
+
+        return json.loads(response.content)
+
+    def get_device_state(self):
+        response = self.list()
+
+        if response is None:
+            return None
+
+        for info in response:
+                if info["nukiId"] == self.get_device_id():
+                    return info["lastKnownState"]
+
+    def lock(self, device_name: str = None): 
+        if device_name == None:
+            nukiId = self.device_id
+        else:
+            RuntimeError("")
+
+        response = self.make_request("lock", {"nukiid": nukiId})
+        
+        if response is None:
+            return False
+
+        if response.status_code == 200:
+            json_response = json.loads(response.content)
+            return json_response["success"]
+
+        log.error(f"Nuki responded with an error: {json_response}")
+        return False
+
+    def unlock(self, device_name: str = None):
+        if device_name == None:
+            nukiId = self.device_id
+        else:
+            RuntimeError("")
+
+        response = self.make_request("unlock", {"nukiid": nukiId})
+
+        if response is None:
+            return False
+
+        if response.status_code == 200:
+            json_response = json.loads(response.content)
+            return json_response["success"]
+        return False
+
+
+class NukiBridge(DoorlockBackend):
+    def __init__(self, endpoint, api_key, device_name):
+        #self.device = NukiBridgeDevice(endpoint,api_key, device_name)
+        self.device = NukiBridgeDevice(endpoint, api_key, device_name)
+        self.poll_thread = threading.Thread(target=self.poll_worker)
+        self.poll_thread.start()
+
+    def poll_worker(self):
+        while True:
+            state = self.device.get_device_state()
+
+            if state is None:
+                continue
+
+            self.current_state = state
+            print(state)
+            time.sleep(10)
+
+    def set_state(self, state):
+        if state == DoorState.Open:
+            log.info("open nuki")
+            return self.device.unlock()
+        elif state == DoorState.Closed:
+            log.info("close nuki")
+            return self.device.lock()
+
+    def get_state(self, state):
+        return self.current_state

pydoorlock/WebApp.py

@@ -202,4 +202,4 @@ def webapp_run(cfg, my_logic, status, version, template_folder, static_folder):
     webapp.template_folder = template_folder
     webapp.static_folder = static_folder
     webapp.debug = debug
-    webapp.run(host=host, port=8080)
+    webapp.run(host = host, port = 8080, use_reloader=False)