Dateien nach "app" hochladen

app/main.py

@@ -0,0 +1,161 @@
+#!/usr/bin/env python3
+"""
+Shelly RGBW2 Animator Service
+-----------------------------
+Listens on MQTT for animation commands and drives the Shelly directly
+via its HTTP API.  Designed to run as a lightweight Docker container
+or systemd service on a Raspberry Pi.
+
+MQTT topics (configurable):
+  <prefix>/set      – write: animation name or "stop" / "off"
+  <prefix>/state    – read:  current animation name or "off"
+
+Example:
+  mosquitto_pub -t shelly-animator/set -m "party"
+  mosquitto_pub -t shelly-animator/set -m "stop"
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import signal
+import sys
+import threading
+import time
+
+import paho.mqtt.client as mqtt
+
+from animations import REGISTRY, get_animation_class
+from shelly import ShellyRGBW2
+
+# ---------------------------------------------------------------------------
+# Config from environment
+# ---------------------------------------------------------------------------
+SHELLY_HOST = os.getenv("SHELLY_HOST", "192.168.178.151")
+SHELLY_USER = os.getenv("SHELLY_USER", "shelly@till.immers.ch")
+SHELLY_PASS = os.getenv("SHELLY_PASS", "sh3llyTi33*")
+
+MQTT_HOST = os.getenv("MQTT_HOST", "localhost")
+MQTT_PORT = int(os.getenv("MQTT_PORT", "1884"))
+MQTT_USER = os.getenv("MQTT_USER", "")
+MQTT_PASS = os.getenv("MQTT_PASS", "")
+MQTT_PREFIX = os.getenv("MQTT_PREFIX", "shelly-animator")
+
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
+
+logging.basicConfig(
+    level=getattr(logging, LOG_LEVEL, logging.INFO),
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    datefmt="%H:%M:%S",
+)
+log = logging.getLogger("main")
+
+# ---------------------------------------------------------------------------
+# Runtime state
+# ---------------------------------------------------------------------------
+shelly = ShellyRGBW2(SHELLY_HOST, SHELLY_USER, SHELLY_PASS)
+current_anim = None  # type: ignore
+lock = threading.Lock()
+
+
+def stop_current() -> None:
+    global current_anim
+    with lock:
+        if current_anim and current_anim.running:
+            current_anim.stop()
+        current_anim = None
+
+
+def start_animation(name: str) -> bool:
+    global current_anim
+    cls = get_animation_class(name)
+    if cls is None:
+        log.warning("Unknown animation: %s  (available: %s)", name, list(REGISTRY))
+        return False
+
+    with lock:
+        if current_anim and current_anim.running:
+            if current_anim.name == cls.name:
+                log.info("Animation '%s' already running", cls.name)
+                return True
+            current_anim.stop()
+
+        anim = cls(shelly)
+        anim.start()
+        current_anim = anim
+    return True
+
+
+def publish_state(client: mqtt.Client) -> None:
+    state = "off"
+    with lock:
+        if current_anim and current_anim.running:
+            state = current_anim.name
+    client.publish(f"{MQTT_PREFIX}/state", state, retain=True)
+
+
+# ---------------------------------------------------------------------------
+# MQTT callbacks
+# ---------------------------------------------------------------------------
+def on_connect(client: mqtt.Client, userdata, flags, reason_code, properties=None):
+    log.info("MQTT connected (rc=%s)", reason_code)
+    client.subscribe(f"{MQTT_PREFIX}/set")
+    publish_state(client)
+
+
+def on_message(client: mqtt.Client, userdata, msg: mqtt.MQTTMessage):
+    payload = msg.payload.decode("utf-8", errors="ignore").strip()
+    log.info("MQTT command: %s", payload)
+
+    if payload.lower() in ("stop", "off", "none", ""):
+        stop_current()
+        publish_state(client)
+        return
+
+    if start_animation(payload):
+        publish_state(client)
+    else:
+        log.warning("Ignored unknown command: %s", payload)
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+def main() -> None:
+    log.info("Shelly Animator starting")
+    log.info("  Shelly : %s", SHELLY_HOST)
+    log.info("  MQTT   : %s:%s  prefix=%s", MQTT_HOST, MQTT_PORT, MQTT_PREFIX)
+    log.info("  Anims  : %s", ", ".join(sorted(set(c.name for c in REGISTRY.values()))))
+
+    client = mqtt.Client(
+        mqtt.CallbackAPIVersion.VERSION2,
+        client_id="shelly-animator",
+    )
+    if MQTT_USER:
+        client.username_pw_set(MQTT_USER, MQTT_PASS)
+
+    client.on_connect = on_connect
+    client.on_message = on_message
+
+    # Graceful shutdown
+    def shutdown(signum, frame):
+        log.info("Shutting down...")
+        stop_current()
+        client.disconnect()
+        sys.exit(0)
+
+    signal.signal(signal.SIGTERM, shutdown)
+    signal.signal(signal.SIGINT, shutdown)
+
+    while True:
+        try:
+            client.connect(MQTT_HOST, MQTT_PORT, keepalive=60)
+            client.loop_forever()
+        except Exception as exc:
+            log.error("MQTT connection error: %s – retry in 5s", exc)
+            time.sleep(5)
+
+
+if __name__ == "__main__":
+    main()

app/shelly.py

@@ -0,0 +1,87 @@
+"""Minimal Shelly RGBW2 (Gen1) HTTP client – color mode only."""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+import requests
+from requests.auth import HTTPDigestAuth
+
+log = logging.getLogger("shelly")
+
+
+class ShellyRGBW2:
+    def __init__(
+        self,
+        host: str,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        timeout: float = 2.0,
+    ):
+        self.base = f"http://{host}"
+        self.timeout = timeout
+        self.session = requests.Session()
+        if username and password:
+            # Shelly Gen1 uses Digest auth when credentials are set
+            self.session.auth = HTTPDigestAuth(username, password)
+
+    def set_color(
+        self,
+        r: int,
+        g: int,
+        b: int,
+        *,
+        white: int = 0,
+        gain: int = 100,
+        turn_on: bool = True,
+        effect: int = 0,
+    ) -> bool:
+        """Send RGB (+ optional white/gain) to channel 0."""
+        params = {
+            "red": max(0, min(255, int(r))),
+            "green": max(0, min(255, int(g))),
+            "blue": max(0, min(255, int(b))),
+            "white": max(0, min(255, int(white))),
+            "gain": max(0, min(100, int(gain))),
+            "effect": int(effect),
+        }
+        if turn_on:
+            params["turn"] = "on"
+        try:
+            resp = self.session.get(
+                f"{self.base}/color/0",
+                params=params,
+                timeout=self.timeout,
+            )
+            resp.raise_for_status()
+            return True
+        except Exception as exc:
+            log.warning("Shelly set_color failed: %s", exc)
+            return False
+
+    def turn_off(self) -> bool:
+        try:
+            resp = self.session.get(
+                f"{self.base}/color/0",
+                params={"turn": "off"},
+                timeout=self.timeout,
+            )
+            resp.raise_for_status()
+            return True
+        except Exception as exc:
+            log.warning("Shelly turn_off failed: %s", exc)
+            return False
+
+    def turn_on(self) -> bool:
+        try:
+            resp = self.session.get(
+                f"{self.base}/color/0",
+                params={"turn": "on"},
+                timeout=self.timeout,
+            )
+            resp.raise_for_status()
+            return True
+        except Exception as exc:
+            log.warning("Shelly turn_on failed: %s", exc)
+            return False