Merge branch 'bettwaage-sidequest'

.gitignore

@@ -161,5 +161,6 @@ cython_debug/
 .vscode/settings.json
 .vscode/launch.json
 
-
+# Custom
 hue_bridge_registered.txt
+history.json

bettwaage-plotter/bett.py

@@ -0,0 +1,27 @@
+import requests as r
+from time import sleep
+
+bett_ip = "http://192.168.178.110:80"
+mash_ip = "http://192.168.178.84:9587"
+
+while True:
+    try:
+        tl = r.get(f"{bett_ip}/sensor/tl/").json()["value"]
+        tr = r.get(f"{bett_ip}/sensor/tr/").json()["value"]
+        br = r.get(f"{bett_ip}/sensor/br/").json()["value"]
+        
+        print(f"tl = {tl}")
+        print(f"tr = {tr}")
+        # print(f"tl = {tl}")
+        print(f"br = {br}")
+        print("==========")
+        print(f"total = {tl + tr + br * 2}")
+        print("==========")
+        
+        s = r.post(f"{mash_ip}/bettwaage/add?tl={int(tl * 1000)}&tr={int(tr * 1000)}&bl={int(br * 1000)}&br={int(br * 1000)}")
+        
+        sleep(1)
+    except KeyboardInterrupt:
+        exit()
+    except:
+        pass

bettwaage-plotter/main.py

@@ -0,0 +1,137 @@
+import requests
+import matplotlib.pyplot as plt
+from datetime import datetime
+import json
+
+
+file_path = "history.json"
+history_url = "http://192.168.178.84:9587/bettwaage/history"
+
+convert_time_to_seconds = True
+
+# Script
+data = None
+
+if file_path is None:
+    data = requests.get(history_url)
+    data = data.json()
+else:
+    with open(file_path, "r") as fp:
+        data = json.load(fp)
+
+
+# Experiment: Solving for missing foot with known total weight
+bed_weight = 78290
+person_weight = 63000
+known_total_weight = bed_weight + person_weight
+bed_only_weight = {}
+for d in data:
+    if d["total"] == bed_weight:
+        bed_only_weight = {
+            "tl": d["tl"],
+            "tr": d["tr"],
+            "bl": bed_weight - (d["tl"] + d["tr"] + d["br"]),
+            "br": d["br"],
+        }
+        break
+
+in_bed_data = None
+threshhold = 100000
+min_length = 100
+for d in data:
+    t = d["total"]
+    if t >= threshhold:
+        if in_bed_data is None:
+            in_bed_data = []
+        in_bed_data.append(d)
+    elif in_bed_data is not None:
+        if len(in_bed_data) < min_length:
+            in_bed_data = []
+        else:
+            break
+
+
+# Calculate bottom left
+for d in data:
+    d["bl"] = known_total_weight - (d["br"] + d["tr"] + d["tl"])
+    # Set known total weight
+    d["total"] = known_total_weight
+
+
+data = in_bed_data
+
+
+# Array data
+x = [d["timestamp"] for d in data]
+x = [datetime.strptime(d, "%Y-%m-%d %H:%M:%S.%f") for d in x]
+
+if convert_time_to_seconds:
+    max_time = max(x)
+    x = [(d - max_time).total_seconds() for d in x]
+
+total = [d["total"] for d in data]
+tl = [d["tl"] for d in data]
+tr = [d["tr"] for d in data]
+bl = [d["bl"] for d in data]
+br = [d["br"] for d in data]
+top = [l + r for l, r in zip(tl, tr)]
+bottom = [l + r for l, r in zip(bl, br)]
+left = [t + b for t, b in zip(tl, bl)]
+right = [t + b for t, b in zip(tr, br)]
+
+
+# Experiment: Calculate position over time
+bed_size = (140, 200)
+left_bed_only = bed_only_weight["tl"] + bed_only_weight["bl"]
+top_bed_only = bed_only_weight["tr"] + bed_only_weight["tl"]
+position_over_time = []
+for t, l in zip(top, left):
+    position_over_time.append(
+        (
+            bed_size[0] * (l - left_bed_only) / person_weight,
+            bed_size[1] * (t - top_bed_only) / person_weight,
+        )
+    )
+
+
+# Plot data
+fig, (ax0, ax1) = plt.subplots(nrows=2)
+
+ax0.set_xlabel("Time (s)")
+ax0.set_ylabel("Weight (kg)")
+
+ax0.plot(x, total, color="tab:blue")
+ax0.plot(x, tl, color="tab:red")
+ax0.plot(x, tr, color="tab:green")
+ax0.plot(x, bl, color="tab:orange")
+ax0.plot(x, br, color="tab:purple")
+ax0.plot(x, top, color="tab:pink")
+ax0.plot(x, bottom, color="tab:grey")
+
+ax0.legend(
+    ["Total", "Top Left", "Top Right", "Bottom Left", "Bottom Right", "Top", "Bottom"]
+)
+
+
+# Experiment: Plot position
+import math
+import colorsys
+
+segments = 100
+seg_length = math.ceil(len(position_over_time) / segments)
+horizontal, vertical = zip(*position_over_time)
+for i in range(0, segments):
+    low = int(i * seg_length)
+    high = min(int((i + 1) * seg_length), len(position_over_time))
+    ax1.plot(
+        horizontal[low:high],
+        vertical[low:high],
+        color=colorsys.hsv_to_rgb(i / segments * 0.5, 1, 0.7),
+        linewidth=0.3,
+    )
+ax1.set_xlim((0, bed_size[0]))
+ax1.set_ylim((0, bed_size[1]))
+ax1.invert_yaxis()
+
+
+plt.show()

bettwaage-plotter/requirements.txt

@@ -0,0 +1,3 @@
+matplotlib
+requests
+numpy

src-new/endpoints/bettwaage.py

@@ -0,0 +1,117 @@
+from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi import APIRouter
+
+from datetime import datetime
+import os
+import csv
+
+
+router = APIRouter()
+
+file_path = "bettwaage.csv"
+header = "timestamp;tl;tr;bl;br;total;"
+
+latest_values = []
+zero_values = [0, 0, 0, 0]
+scale_values = [1, 1, 1, 1]
+
+
+def add_line_to_history(line: str) -> None:
+    with open(file_path, "a") as fp:
+        fp.write(line + "\n")
+
+
+def convert_to_weight(value: int, zero_value: int, scale: float) -> float:
+    return (value - zero_value) * scale
+
+
+@router.get("/file", tags=["file"])
+async def get_file():
+    with open(file_path, "r", encoding="UTF-8") as fp:
+        return HTMLResponse("\n".join(fp.readlines()))
+
+
+@router.get("/history")
+async def get_history(count: int = None) -> []:
+    points = []
+    with open(file_path, "r", encoding="UTF-8") as fp:
+        reader = csv.DictReader(fp, delimiter=";")
+        for row in reader:
+            if not row:
+                continue
+
+            points.append(
+                {
+                    "timestamp": row["timestamp"],
+                    "total": float(row["total"]),
+                    "tl": float(row["tl"]),
+                    "tr": float(row["tr"]),
+                    "bl": float(row["bl"]),
+                    "br": float(row["br"]),
+                }
+            )
+
+    if count:
+        return points[-count]
+    else:
+        return points
+
+
+@router.post("/add")
+async def add_weight(tl: int, tr: int, bl: int, br: int):
+    global latest_values
+    latest_values = [tl, tr, bl, br]
+
+    tl = convert_to_weight(tl, zero_values[0], scale_values[0])
+    tr = convert_to_weight(tr, zero_values[1], scale_values[1])
+    bl = convert_to_weight(bl, zero_values[2], scale_values[2])
+    br = convert_to_weight(br, zero_values[3], scale_values[3])
+
+    sum = tl + tr + bl + br
+    add_line_to_history(f"{str(datetime.now())};{tl};{tr};{bl};{br};{sum};")
+    return "Added data"
+
+
+@router.get("/latest")
+async def get_latest():
+    if not latest_values:
+        return HTMLResponse(status_code=200, content="No data given yet")
+    total = sum(latest_values)
+    return JSONResponse(
+        {
+            "tl": latest_values[0],
+            "tr": latest_values[1],
+            "bl": latest_values[2],
+            "br": latest_values[3],
+            "total": total,
+        }
+    )
+
+
+@router.delete("/delete", tags=["file"])
+async def delete_file():
+    os.remove(file_path)
+    add_line_to_history(header)
+    return "Deleted file and created new file with headers"
+
+
+@router.post("/zero", tags=["calibration"])
+async def set_zero():
+    if not latest_values:
+        return HTMLResponse(
+            status_code=400, content="Requiring data before setting zeros."
+        )
+    global zero_values
+    zero_values = latest_values
+    return "Set zeroes to: " + " | ".join(str(v) for v in zero_values)
+
+
+@router.post("/scales", tags=["calibration"])
+async def set_scales(tl: float, tr: float, bl: float, br: float):
+    global scale_values
+    scale_values = [tl, tr, bl, br]
+    return "Set scales to: " + " | ".join(str(v) for v in scale_values)
+
+
+if not os.path.exists(file_path):
+    add_line_to_history(header)

src-new/hue/hue_adapter.py

@@ -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})

src-new/hue/hue_feature.py

@@ -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))

src-new/main.py

@@ -7,5 +7,7 @@ mash: MaSH = MaSH("config.yaml")
 mash.add_module(HueModule())
 mash.add_module(MatrixClockModule())
 
+app.include_router(bettwaage_router, prefix="/bettwaage", tags=["bett"])
+
 if __name__ == "__main__":
     mash.run()