Implemented basic philips hue endpoints to server

.gitignore

@@ -159,3 +159,4 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 .vscode/settings.json
+.vscode/launch.json

requirements.txt

@@ -1,2 +1,6 @@
 # For Philips Hue Counter
 phue
+
+# API
+fastapi
+uvicorn[standard]

src/endpoints/handlers/hue.py

@@ -0,0 +1,86 @@
+from time import sleep
+from phue import Bridge
+from pathlib import Path
+
+
+class HueHandler:
+    """Handler for Hue API calls."""
+
+    registered_ips_file = "hue_bridge_registered.txt"
+
+    def __init__(self, bridge_ip: str):
+        """Initialize the HueHandler."""
+        self.bridge = None
+        self.connect(bridge_ip)
+
+    def connect(self, bridge_ip: str):
+        if bridge_ip in self.get_registered_ips():
+            self.bridge = Bridge(bridge_ip)
+            self.bridge.connect()
+            return
+
+        # Connect loop
+        while True:
+            try:
+                self.bridge = Bridge(bridge_ip)
+                self.bridge.connect()
+                break
+            except Exception as e:
+                print(f"Failed to connect to bridge: {bridge_ip}")
+                print(e)
+                print("Trying again in 5 seconds..")
+                sleep(5)
+
+        self.register_bridge(bridge_ip)
+
+    def get_registered_ips(self) -> list:
+        """Get a list of registered bridge IPs."""
+        if not Path(HueHandler.registered_ips_file).is_file():
+            return []
+
+        with open(HueHandler.registered_ips_file, "r") as f:
+            return f.readlines()
+
+    def register_bridge(self, bridge_ip: str):
+        """Register a bridge IP."""
+        with open(HueHandler.registered_ips_file, "a") as f:
+            f.write(bridge_ip + "\n")
+
+    def list_scenes(self) -> dict:
+        return self.bridge.get_scene()
+
+    def get_scene_by_name(self, name):
+        for key, scene in self.list_scenes().items():
+            if scene["name"] == name:
+                scene["id"] = key
+                return scene
+        return None
+
+    def in_room_activate_scene(self, room_name: str, scene_name: str):
+        """Activate a scene in a room.
+
+        Args:
+            scene (str): The name of the scene to activate.
+            room (str): The name of the room to activate the scene in.
+        """
+        scene_id = self.get_scene_by_name(scene_name)["id"]
+        if scene_id is None:
+            raise "Scene not found."
+
+        self.bridge.set_group(room_name, {"scene": scene_id})
+
+    def in_room_deactivate_lights(self, room_name: str):
+        """Deactivate all lights in a room.
+
+        Args:
+            room_name (str): The name of the room to deactivate the lights in.
+        """
+        self.bridge.set_group(room_name, {"on": False})
+
+    def in_room_activate_lights(self, room_name: str):
+        """Activate all lights in a room.
+
+        Args:
+            room_name (str): The name of the room to activate the lights in.
+        """
+        self.bridge.set_group(room_name, {"on": True})

src/endpoints/hue.py

@@ -0,0 +1,44 @@
+from fastapi import APIRouter
+from fastapi.responses import HTMLResponse
+from .handlers.hue import HueHandler
+
+router = APIRouter()
+hue = HueHandler("192.168.178.85")
+
+
+@router.get("/scenes", tags=["scene"])
+async def get_scenes():
+    return hue.list_scenes()
+
+
+@router.post(
+    "/room/{room_name}/scene/{scene_name}",
+    tags=["room", "scene"],
+)
+async def activate_scene(room_name: str, scene_name: str):
+    try:
+        hue.in_room_activate_scene(room_name, scene_name)
+    except Exception as e:
+        return HTMLResponse(status_code=400, content=str(e))
+
+
+@router.post(
+    "/room/{room_name}/off",
+    tags=["room"],
+)
+async def deactivate_room(room_name: str):
+    try:
+        hue.in_room_deactivate_lights(room_name)
+    except Exception as e:
+        return HTMLResponse(status_code=400, content=str(e))
+
+
+@router.post(
+    "/room/{room_name}/on",
+    tags=["room"],
+)
+async def activate_room(room_name: str):
+    try:
+        hue.in_room_activate_lights(room_name)
+    except Exception as e:
+        return HTMLResponse(status_code=400, content=str(e))

src/hue_bridge_registered.txt

@@ -0,0 +1,15 @@
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85
+192.168.178.85

src/main.py

@@ -0,0 +1,9 @@
+from fastapi import FastAPI
+from endpoints.hue import router as hue_router
+
+app = FastAPI()
+
+app.include_router(hue_router, prefix="/hue", tags=["hue"])
+
+if __name__ == "__main__":
+    app.run()

src/old_philips_hue_examples.py

@@ -5,18 +5,18 @@ import logging
 import socket
 
 
-class PhilipsHue ():
+class PhilipsHue:
     def __init__(self, config):
         self.config = config
         self.connect()
 
     def connect(self):
-        registered = Path(self.config['registered_file']).is_file()
+        registered = Path(self.config["registered_file"]).is_file()
         success = False
         while success == False:
             try:
                 logging.info("Connecting to hue bridge")
-                self.bridge = Bridge(self.config['bridge_ip'])
+                self.bridge = Bridge(self.config["bridge_ip"])
                 self.bridge.connect()
                 success = True
             except Exception as e:
@@ -32,7 +32,7 @@ class PhilipsHue ():
         if registered == False:
             # register
             logging.info("Saving registration")
-            Path(self.config['registered_file']).touch()
+            Path(self.config["registered_file"]).touch()
 
     def get_state(self):
         return self.__execute__(lambda: self.bridge.get_api())
@@ -42,8 +42,8 @@ class PhilipsHue ():
 
     def get_scene_by_name(self, name):
         for key, scene in self.get_scenes().items():
-            if scene['name'] == name:
-                scene['id'] = key
+            if scene["name"] == name:
+                scene["id"] = key
                 return scene
         return None
 
@@ -60,12 +60,14 @@ class PhilipsHue ():
         return self.__execute__(lambda: self.bridge.get_group(id, command))
 
     def set_group_scene(self, group_name, scene_name):
-        scene_id = self.get_scene_by_name(scene_name)['id']
-        return self.__execute__(lambda: self.set_group(group_name, self.create_conf({'scene': scene_id})))
+        scene_id = self.get_scene_by_name(scene_name)["id"]
+        return self.__execute__(
+            lambda: self.set_group(group_name, self.create_conf({"scene": scene_id}))
+        )
 
     def create_conf(self, conf):
-        if 'transitiontime' not in conf.keys():
-            conf['transitiontime'] = self.config['transition_time']
+        if "transitiontime" not in conf.keys():
+            conf["transitiontime"] = self.config["transition_time"]
         return conf
 
     def __execute__(self, function):
@@ -74,13 +76,13 @@ class PhilipsHue ():
         except socket.timeout as e:
             # Try to reconnect
             logging.exception(
-                "Could not execute function. Trying to reconnect to bridge")
+                "Could not execute function. Trying to reconnect to bridge"
+            )
             logging.exception(str(e))
             try:
                 self.connect()
             except Exception as e:
-                logging.exception(
-                    "Reconnect did not succeed, skipping execution")
+                logging.exception("Reconnect did not succeed, skipping execution")
                 logging.exception(str(e))
                 return
             # Now try again