Implemented new fritzbox

src/bridges/fritzbox/fritzbox_bridge.py

@@ -1,5 +1,4 @@
 import logging
-from typing import Optional
 from core.bridge import Bridge
 from fritzconnection import FritzConnection
 
@@ -51,13 +50,13 @@ class FritzBoxBridge(Bridge):
         *,
         id: str,
         ip: str,
-        port: Optional[int] = None,
+        port: int | None = None,
     ) -> None:
         """
         Args:
             id (str): Id of fritzbox bridge.
             ip (str): IP Address of fritzbox bridge in network to connect to.
-            port (Optional[int], optional): Port of fritzbox bridge in network to connect to. Defaults to None.
+            port (int, optional): Port of fritzbox bridge in network to connect to. Defaults to None.
         """
         self._ip = ip
         self._port = port

src/endpoints/fritzbox.py

@@ -0,0 +1,54 @@
+import asyncio
+import logging
+from datetime import datetime
+
+from fastapi import APIRouter
+from fastapi.responses import HTMLResponse
+from bridges.fritzbox import FritzBoxBridge
+
+router = APIRouter()
+
+fritzbox = FritzBoxBridge(id="fritzbox", ip="192.168.178.1")
+refresh_every_seconds = 10
+
+macaddresses_to_track = ["B2:06:77:EE:A9:0F"]  # Max' iPhone
+
+devices_last_online: dict[str, datetime] = {}
+
+
+async def track_network_devices():
+    global devices_last_online
+
+    # Initial values to avoid None
+    for macaddress in macaddresses_to_track:
+        devices_last_online[macaddress] = datetime(1970, 1, 1, 0, 0, 0)
+
+    while True:
+        try:
+            for macaddress in macaddresses_to_track:
+                device = fritzbox.get_device_state(macaddress)
+                if device.active:
+                    devices_last_online[macaddress] = datetime.now()
+
+        except Exception as ex:
+            logging.exception(ex)
+        finally:
+            await asyncio.sleep(refresh_every_seconds)
+
+
+@router.get("/{mac_address}/state")
+async def get_latest(mac_address: str):
+    if mac_address not in devices_last_online.keys():
+        return HTMLResponse(status_code=200, content="Mac Address not being tracked.")
+
+    last_online_delta = datetime.now() - last_online_delta[mac_address]
+
+    return {
+        "active": last_online_delta < refresh_every_seconds * 2,
+        "last_active": last_online_delta[mac_address],
+    }
+
+
+@router.get("/tracked")
+async def get_latest():
+    return list(devices_last_online.keys())

src/main.py

@@ -3,8 +3,8 @@ from contextlib import asynccontextmanager
 from fastapi import FastAPI
 import uvicorn
 from endpoints.hue import router as hue_router
-from endpoints.bettwaage import bedscale_service, router as bettwaage_router
-from endpoints.handlers.fritz import track_network_devices
+from endpoints.bedscale import bedscale_service, router as bettwaage_router
+from endpoints.fritzbox import track_network_devices, router as fritzbox_router
 
 # Background task references
 background_tasks = []
@@ -34,7 +34,8 @@ app = FastAPI(lifespan=lifespan)
 
 # API Routes
 app.include_router(hue_router, prefix="/hue", tags=["hue"])
-app.include_router(bettwaage_router, prefix="/bettwaage", tags=["bett"])
+app.include_router(bettwaage_router, prefix="/bettwaage", tags=["bed"])
+app.include_router(fritzbox_router, prefix="/fritzbox", tags=["fritzbox"])
 
 
 if __name__ == "__main__":