bettwaage-sidequest #2
18 changed files with 464 additions and 25 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
@ -162,4 +162,5 @@ cython_debug/
|
|||
.vscode/launch.json
|
||||
|
||||
# Custom
|
||||
hue_bridge_registered.txt
|
||||
history.json
|
||||
|
|
|
@ -1,4 +1,4 @@
|
|||
# Max's Smart Home - MaSH
|
||||
# Max' Smart Home - MaSH
|
||||
|
||||
Should be a very simple **server** implementation of what is required in Max's smart home. Trying not to overcomplicate things and thereby ruin motivation to work on this.
|
||||
|
||||
|
|
36
example.conf.yaml
Normal file
36
example.conf.yaml
Normal file
|
@ -0,0 +1,36 @@
|
|||
features:
|
||||
hue:
|
||||
hue-bridge:
|
||||
ip: 192.168.178.23
|
||||
matrixclock:
|
||||
clock:
|
||||
ip: 192.168.178.23
|
||||
|
||||
home:
|
||||
latitude: 52.51860
|
||||
longitude: 13.37565
|
||||
|
||||
beds:
|
||||
- id: max-bed
|
||||
name: Bettwaage
|
||||
room: *max
|
||||
|
||||
rooms:
|
||||
- id : &hw hallway
|
||||
name: Flur
|
||||
doors:
|
||||
- to: ~
|
||||
- to: *bath
|
||||
- to: *kit
|
||||
- to: *living
|
||||
- id : &bath bath
|
||||
name: Badezimmer
|
||||
- id : &kit kitchen
|
||||
name: Küche
|
||||
- id : &living living
|
||||
name: Wohnzimmer
|
||||
doors:
|
||||
- to: *max
|
||||
- id : &max max
|
||||
name: Max' Zimmer
|
||||
|
|
@ -7,4 +7,10 @@ fritzconnection
|
|||
# API
|
||||
requests
|
||||
fastapi
|
||||
uvicorn[standard]
|
||||
uvicorn[standard]
|
||||
|
||||
# Clients
|
||||
requests
|
||||
|
||||
# Config file
|
||||
pyyaml
|
||||
|
|
|
@ -19,6 +19,7 @@ async def get_file():
|
|||
|
||||
@router.get("/history")
|
||||
async def get_history(count: int = None) -> list[dict]:
|
||||
|
||||
points = []
|
||||
with open(file_path, "r", encoding="UTF-8") as fp:
|
||||
reader = csv.DictReader(fp, delimiter=";")
|
||||
|
@ -50,6 +51,7 @@ async def get_latest():
|
|||
return JSONResponse(local_history[-1])
|
||||
|
||||
|
||||
|
||||
@router.delete("/delete", tags=["file"])
|
||||
async def delete_file():
|
||||
os.remove(file_path)
|
||||
|
|
|
@ -3,7 +3,7 @@ from phue import Bridge
|
|||
from pathlib import Path
|
||||
|
||||
|
||||
class HueHandler:
|
||||
class HueAdapter:
|
||||
"""Handler for Hue API calls."""
|
||||
|
||||
registered_ips_file = "hue_bridge_registered.txt"
|
||||
|
@ -35,15 +35,15 @@ class HueHandler:
|
|||
|
||||
def get_registered_ips(self) -> list:
|
||||
"""Get a list of registered bridge IPs."""
|
||||
if not Path(HueHandler.registered_ips_file).is_file():
|
||||
if not Path(HueAdapter.registered_ips_file).is_file():
|
||||
return []
|
||||
|
||||
with open(HueHandler.registered_ips_file, "r") as f:
|
||||
return f.readlines()
|
||||
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(HueHandler.registered_ips_file, "a") as f:
|
||||
with open(HueAdapter.registered_ips_file, "a") as f:
|
||||
f.write(bridge_ip + "\n")
|
||||
|
||||
def list_scenes(self) -> dict:
|
||||
|
|
|
@ -1,9 +1,25 @@
|
|||
from fastapi import FastAPI, APIRouter
|
||||
|
||||
from hue.hue_adapter import HueAdapter
|
||||
from ..mash.feature import Feature
|
||||
from fastapi import APIRouter
|
||||
from fastapi.responses import HTMLResponse
|
||||
from .handlers.hue import HueHandler
|
||||
|
||||
router = APIRouter()
|
||||
hue = HueHandler("192.168.178.85")
|
||||
router = APIRouter(tags=["hue"])
|
||||
hue = HueAdapter("192.168.178.85")
|
||||
|
||||
########## Integration ##########
|
||||
|
||||
|
||||
class HueIntegration(Feature):
|
||||
def __init__(self) -> None:
|
||||
super().__init__("hue")
|
||||
|
||||
def add_routes(self, server: FastAPI) -> None:
|
||||
server.include_router(router, prefix="/hue")
|
||||
|
||||
|
||||
########## Routes ##########
|
||||
|
||||
|
||||
@router.get("/scenes", tags=["scene"])
|
||||
|
|
88
src/example.py
Normal file
88
src/example.py
Normal file
|
@ -0,0 +1,88 @@
|
|||
class Device:
|
||||
def __init__(self, device_id: str, home: "Home") -> None:
|
||||
self.device_id: str = device_id
|
||||
self.home: Home = home
|
||||
|
||||
def trigger_change(self):
|
||||
self.home.trigger_device_change(self.device_id)
|
||||
|
||||
|
||||
class Home:
|
||||
automations: list["Automation"]
|
||||
|
||||
def _get_device_behaviours_(self, device_id: str) -> list["Automation"]:
|
||||
return [b for a in self.automations]
|
||||
|
||||
def trigger_device_change(self, device_id: str):
|
||||
pass
|
||||
|
||||
|
||||
class Behaviour:
|
||||
def __init__(self) -> None:
|
||||
self.action: function = None
|
||||
self.rule: function = None
|
||||
self.devices: list[str] = []
|
||||
|
||||
|
||||
class Automation:
|
||||
def __init__(self) -> None:
|
||||
self.behaviours: list[Behaviour] = []
|
||||
|
||||
def device(self, device_id: str) -> dict:
|
||||
return {"contrast": 3}
|
||||
|
||||
def trigger(self):
|
||||
def decorator(func):
|
||||
|
||||
return func
|
||||
return decorator
|
||||
|
||||
|
||||
class PeopleCountEngineV1(Automation):
|
||||
@Automation.trigger(
|
||||
devices=["matrixclock"],
|
||||
rule=lambda h: h.device("matrixclock").contrast == 6
|
||||
)
|
||||
def turn_light_on_sometimes(self, home: Home):
|
||||
home.room("max").lights().on = True
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@Automation.trigger(
|
||||
people=["max"],
|
||||
rule=lambda h: h.person("max").athome()
|
||||
)
|
||||
def turn_light_on_sometimes(self, home: Home):
|
||||
home.room("max").lights().on = h.person("max").athome()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@Automation.state(h.room("Max").lights())
|
||||
def max_room_light():
|
||||
if max.ishome():
|
||||
return "off"
|
||||
|
||||
scene = "Daylight scene"
|
||||
|
||||
if nighttime:
|
||||
scene = "nighttime"
|
||||
|
||||
if max.working:
|
||||
scene.dim(0.5)
|
||||
|
||||
return scene
|
||||
|
||||
|
||||
|
||||
from mash.mash import MaSH
|
||||
|
||||
mash = MaSH()
|
||||
|
||||
mash.add_automation(PeopleCountEngineV1())
|
86
src/hue/hue_adapter.py
Normal file
86
src/hue/hue_adapter.py
Normal file
|
@ -0,0 +1,86 @@
|
|||
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})
|
60
src/hue/hue_feature.py
Normal file
60
src/hue/hue_feature.py
Normal file
|
@ -0,0 +1,60 @@
|
|||
from fastapi import FastAPI, APIRouter
|
||||
|
||||
from hue.hue_adapter import HueAdapter
|
||||
from ..mash.feature import Feature
|
||||
from fastapi import APIRouter
|
||||
from fastapi.responses import HTMLResponse
|
||||
|
||||
router = APIRouter(tags=["hue"])
|
||||
hue = HueAdapter("192.168.178.85")
|
||||
|
||||
########## Integration ##########
|
||||
|
||||
|
||||
class HueIntegration(Feature):
|
||||
def __init__(self) -> None:
|
||||
super().__init__("hue")
|
||||
|
||||
def add_routes(self, server: FastAPI) -> None:
|
||||
server.include_router(router, prefix="/hue")
|
||||
|
||||
|
||||
########## Routes ##########
|
||||
|
||||
|
||||
@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))
|
|
@ -1,15 +0,0 @@
|
|||
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
|
32
src/mash/entities/entity.py
Normal file
32
src/mash/entities/entity.py
Normal file
|
@ -0,0 +1,32 @@
|
|||
class Entity:
|
||||
def __init__(
|
||||
self, *, id: str, name: str, room: str, device_type: str, groups: list[str] = []
|
||||
) -> None:
|
||||
self._id = id
|
||||
self._name = name
|
||||
self._room = room
|
||||
self._device_type = device_type
|
||||
self._groups = set(groups)
|
||||
|
||||
@property
|
||||
def id(self) -> str:
|
||||
return self._id
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def room(self) -> str:
|
||||
return self._room
|
||||
|
||||
@property
|
||||
def device_type(self) -> str:
|
||||
return self._device_type
|
||||
|
||||
@property
|
||||
def groups(self) -> set[str]:
|
||||
return self._groups
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.name} [{self.id}, type {self.device_type}, room {self.room}, in {len(self.groups)} groups]"
|
66
src/mash/entities/group.py
Normal file
66
src/mash/entities/group.py
Normal file
|
@ -0,0 +1,66 @@
|
|||
from mash.entities.entity import Entity
|
||||
from fnmatch import fnmatch
|
||||
|
||||
|
||||
class Group:
|
||||
def __init__(self, *, entities: list[Entity] = []) -> None:
|
||||
self.entities: list[Entity] = entities
|
||||
|
||||
def __len__(self):
|
||||
return len(self.entities)
|
||||
|
||||
def __getitem__(self, id: str) -> "Group":
|
||||
if type(id) is int:
|
||||
raise "Numerical index not supported."
|
||||
|
||||
return self.id(id)
|
||||
|
||||
def __get_entities_with_specific_property__(
|
||||
entities: list[Entity],
|
||||
property_getter: callable[[Entity], str],
|
||||
target_pattern: str,
|
||||
) -> list[Entity]:
|
||||
"""Returns all entities for which the property getter matches the desired pattern.
|
||||
|
||||
Args:
|
||||
entities (list[Entity]): List of entities.
|
||||
property_getter (callable[[Entity], str]): Takes one entity and returns the value of the filtered property.
|
||||
target_pattern (str): Pattern that is matched against.
|
||||
|
||||
Returns:
|
||||
list[Entity]: A new group of entities or an empty group, if no entity property matches the target pattern.
|
||||
"""
|
||||
return Group(
|
||||
entities=[
|
||||
e for e in entities if fnmatch(property_getter(e), target_pattern)
|
||||
]
|
||||
)
|
||||
|
||||
def device_type(self, device_type: str) -> "Group":
|
||||
return Group.__get_entities_with_specific_property__(
|
||||
self.entities, lambda e: e.device_type, device_type
|
||||
)
|
||||
|
||||
def id(self, id: str) -> "Group":
|
||||
return Group.__get_entities_with_specific_property__(
|
||||
self.entities, lambda e: e.id, id
|
||||
)
|
||||
|
||||
def room(self, room: str) -> "Group":
|
||||
return Group.__get_entities_with_specific_property__(
|
||||
self.entities, lambda e: e.room, room
|
||||
)
|
||||
|
||||
def name(self, name: str) -> "Group":
|
||||
return Group.__get_entities_with_specific_property__(
|
||||
self.entities, lambda e: e.name, name
|
||||
)
|
||||
|
||||
def lights(self) -> "Group":
|
||||
return self.device_type("light")
|
||||
|
||||
def beds(self) -> "Group":
|
||||
return self.device_type("bed")
|
||||
|
||||
def max(self) -> "Group":
|
||||
return self.room("max")
|
9
src/mash/feature.py
Normal file
9
src/mash/feature.py
Normal file
|
@ -0,0 +1,9 @@
|
|||
from fastapi import FastAPI
|
||||
|
||||
|
||||
class Feature:
|
||||
def __init__(self, feature_id: str) -> None:
|
||||
self.integration_id = feature_id
|
||||
|
||||
def add_routes(self, server: FastAPI) -> None:
|
||||
pass
|
7
src/mash/home.py
Normal file
7
src/mash/home.py
Normal file
|
@ -0,0 +1,7 @@
|
|||
from mash.entities.entity import Entity
|
||||
from mash.entities.group import Group
|
||||
|
||||
|
||||
class Home(Group):
|
||||
def __init__(self, *, entities: list[Entity] = []) -> None:
|
||||
super().__init__(entities=entities)
|
26
src/mash/mash.py
Normal file
26
src/mash/mash.py
Normal file
|
@ -0,0 +1,26 @@
|
|||
import yaml
|
||||
from fastapi import FastAPI
|
||||
|
||||
from mash.feature import Feature
|
||||
|
||||
|
||||
class MaSH:
|
||||
def __init__(self, config_path: str) -> None:
|
||||
self.server: FastAPI = FastAPI()
|
||||
self.config: dict = None
|
||||
|
||||
self._load_config_(config_path)
|
||||
|
||||
def _load_config_(self, config_path: str) -> None:
|
||||
try:
|
||||
with open(config_path, "r", encoding="UTF-8") as fp:
|
||||
self.config = yaml.safe_load(fp)
|
||||
except FileNotFoundError:
|
||||
raise f"Config file for MaSH server could not be opened at [{config_path}]."
|
||||
|
||||
def add_integration(self, feature: Feature) -> None:
|
||||
feature.add_routes(self.server)
|
||||
self.server.include_router(feature.get_router())
|
||||
|
||||
def run(self):
|
||||
self.server.run()
|
13
src/matrix_clock/matrix_clock_adapter.py
Normal file
13
src/matrix_clock/matrix_clock_adapter.py
Normal file
|
@ -0,0 +1,13 @@
|
|||
import asyncio
|
||||
import requests as r
|
||||
|
||||
|
||||
class MatrixClockAdapter:
|
||||
def __init__(self, ip_address: str) -> None:
|
||||
self.ip_address = ip_address.strip("/")
|
||||
|
||||
if not self.ip_address.startswith("http"):
|
||||
self.ip_address = f"http://{self.ip_address}"
|
||||
|
||||
async def turn_off(self):
|
||||
await asyncio.run(r.post(f"{self.ip_address}/off"))
|
6
src/matrix_clock/matrix_clock_feature.py
Normal file
6
src/matrix_clock/matrix_clock_feature.py
Normal file
|
@ -0,0 +1,6 @@
|
|||
from mash.feature import Feature
|
||||
|
||||
|
||||
class MatrixClockIntegration(Feature):
|
||||
def __init__(self) -> None:
|
||||
super().__init__("matrixclock")
|
Loading…
Reference in a new issue