Some unfinished hue stuff

src/bridges/hue/__init__.py

@@ -1 +1,2 @@
 from .hue_bridge import HueBridge
+from .hue_light import HueLight

src/bridges/hue/hue_light.py

@@ -1,8 +1,39 @@
 from core import Entity
+from bridges.hue import HueBridge
 
 
 class HueLight(Entity):
+
     def __init__(
-        self, *, id: str, name: str, room: str, groups: list[str] = ...
+        self,
+        *,
+        bridge: HueBridge,
+        id: str,
+        name: str,
+        room: str,
+        groups: list[str] = ...
     ) -> None:
         super().__init__(id=id, name=name, room=room, groups=groups)
+        self._bridge: HueBridge = bridge
+
+    def poll_state(self):
+        # TODO
+        pass
+
+    def set_brightness(self, brightness: float):
+        """Does not turn an entity on if brightness > 0 and entity turned off."""
+        raise NotImplementedError("Entity does not support 'set_brightness' operation.")
+
+    def set_hue(self, hue: float):
+        raise NotImplementedError("Entity does not support 'set_hue' operation.")
+
+    def set_saturation(self, saturation: float):
+        raise NotImplementedError("Entity does not support 'set_saturation' operation.")
+
+    def set_color(self, color: Color):
+        raise NotImplementedError("Entity does not support 'set_color' operation.")
+
+    def set_transition_duration(self, seconds: float):
+        raise NotImplementedError(
+            "Entity does not support 'set_transition_duration' operation."
+        )

src/core/__init__.py

@@ -2,3 +2,4 @@ from .bridge import Bridge, BridgeException
 from .entity import Entity
 from .group import Group
 from .room import Room
+from .color import Color

src/core/color.py

@@ -0,0 +1,5 @@
+class Color:
+    def __init__(self, *, hue: float, saturation: float, brightness: float):
+        self.hue = hue
+        self.saturation = saturation
+        self.brightness = brightness

src/core/entity.py

@@ -1,3 +1,11 @@
+from core import Color
+
+
+class EntityOpNotSupportedError(Exception):
+    def __init__(self, operation: str, *args):
+        super().__init__(f"Entity does not support '{operation}' operation.", *args)
+
+
 class Entity:
 
     def __init__(
@@ -40,17 +48,37 @@ class Entity:
     def __str__(self) -> str:
         return f"{self.name} [{self.id}, type {self.device_type}, room {self.room}, in {len(self.groups)} groups]"
 
-    def toggle_state(self):
+    async def poll_state(self):
+        """Implements an entity specific poll operation to get the latest state."""
+        raise EntityOpNotSupportedError("poll_state")
+
+    async def toggle_state(self):
         """Turns entity on, if off, and vice versa, if supported."""
         if self.__is_on__ == False:  # Neither True nor None
-            self.turn_on()
+            await self.turn_on()
         else:
-            self.turn_off()
+            await self.turn_off()
 
-    def turn_on(self):
+    async def turn_on(self):
         """Turns entity on, if action supported."""
         self.__is_on__ = True  # TODO: What if action unsuccessful?
 
-    def turn_off(self):
+    async def turn_off(self):
         """Turns entity on, if action supported."""
         self.__is_on__ = False  # TODO: What if action unsuccessful?
+
+    async def set_brightness(self, brightness: float):
+        """Does not turn an entity on if brightness > 0 and entity turned off."""
+        raise EntityOpNotSupportedError("set_brightness")
+
+    async def set_hue(self, hue: float):
+        raise EntityOpNotSupportedError("set_hue")
+
+    async def set_saturation(self, saturation: float):
+        raise EntityOpNotSupportedError("set_saturation")
+
+    async def set_color(self, color: Color):
+        raise EntityOpNotSupportedError("set_color")
+
+    async def set_transition_duration(self, seconds: float):
+        raise EntityOpNotSupportedError("set_transition_duration")

src/endpoints/handlers/hue.py

@@ -1,86 +0,0 @@
-from time import sleep
-from phue import Bridge
-from pathlib import Path
-
-
-class HueAdapter:
-    """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(HueAdapter.registered_ips_file).is_file():
-            return []
-
-        with open(HueAdapter.registered_ips_file, "r") as f:
-            return [ad.strip() for ad in f.readlines()]
-
-    def register_bridge(self, bridge_ip: str):
-        """Register a bridge IP."""
-        with open(HueAdapter.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

@@ -1,11 +1,31 @@
+import asyncio
 from fastapi import FastAPI, APIRouter
 
-from .handlers.hue import HueAdapter
+from bridges.hue import HueBridge, HueLight
 from fastapi import APIRouter
 from fastapi.responses import HTMLResponse
 
 router = APIRouter(tags=["hue"])
-hue = HueAdapter("192.168.178.85")
+hue = HueBridge("192.168.178.85")
+lights: dict[int, HueLight] = {}
+
+poll_delay_sec = 5
+
+
+async def hue_service():
+    global lights
+
+    while True:
+        try:
+            for light in lights:
+                light.poll_state()
+
+            # TODO: Get all new lights
+
+            await asyncio.sleep(poll_delay_sec)
+        except:
+            pass
+
 
 @router.get("/scenes", tags=["scene"])
 async def get_scenes():

src/main.py

@@ -2,7 +2,7 @@ import asyncio
 from contextlib import asynccontextmanager
 from fastapi import FastAPI
 import uvicorn
-from endpoints.hue import router as hue_router
+from endpoints.hue import hue_service, router as hue_router
 from endpoints.bedscale import bedscale_service, router as bettwaage_router
 from endpoints.fritzbox import track_network_devices, router as fritzbox_router
 
@@ -15,9 +15,10 @@ async def lifespan(app: FastAPI):
     """Start background services."""
     fritz_task = asyncio.create_task(track_network_devices(), name="Fritz!Box Tracker")
     bedscale_task = asyncio.create_task(bedscale_service(), name="Polling bed-scale")
+    hue_task = asyncio.create_task(hue_service(), name="Polling Hue Bridge")
 
     # Store references to the tasks
-    background_tasks.extend([fritz_task, bedscale_task])
+    background_tasks.extend([fritz_task, bedscale_task, hue_task])
 
     yield