From 8f8d206166eb9117250fafa37ccfbe0067530d7e Mon Sep 17 00:00:00 2001 From: Till Immer Date: Thu, 8 Oct 2026 14:44:53 +0000 Subject: [PATCH] Dateien nach "app" hochladen --- app/main.py | 161 ++++++++++++++++++++++++++++++++++++++++++++++++++ app/shelly.py | 87 +++++++++++++++++++++++++++ 2 files changed, 248 insertions(+) create mode 100644 app/main.py create mode 100644 app/shelly.py diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..ba5fc6a --- /dev/null +++ b/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): + /set – write: animation name or "stop" / "off" + /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() diff --git a/app/shelly.py b/app/shelly.py new file mode 100644 index 0000000..b6e6998 --- /dev/null +++ b/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