diff --git a/.gitignore b/.gitignore index 40def25..ef5bac5 100644 --- a/.gitignore +++ b/.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 diff --git a/requirements.txt b/requirements.txt index 420ffe5..3fd5bc7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,6 @@ # For Philips Hue Counter phue + +# API +fastapi +uvicorn[standard] \ No newline at end of file diff --git a/src/endpoints/handlers/hue.py b/src/endpoints/handlers/hue.py new file mode 100644 index 0000000..dd85153 --- /dev/null +++ b/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}) diff --git a/src/endpoints/hue.py b/src/endpoints/hue.py new file mode 100644 index 0000000..74fd6d2 --- /dev/null +++ b/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)) diff --git a/src/hue_bridge_registered.txt b/src/hue_bridge_registered.txt new file mode 100644 index 0000000..a73b77a --- /dev/null +++ b/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 diff --git a/src/main.py b/src/main.py index e69de29..f88263b 100644 --- a/src/main.py +++ b/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() diff --git a/src/philips_hue.py b/src/old_philips_hue_examples.py similarity index 76% rename from src/philips_hue.py rename to src/old_philips_hue_examples.py index 0d09c4d..1327ba2 100644 --- a/src/philips_hue.py +++ b/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