Christian Mueller
65314f2338
FX=0 muss nach SS= kommen, damit nur das gewaehlte Segment zurueckgesetzt wird und nicht beide. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
import logging
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TIMEOUT = 5
|
|
|
|
|
|
def build_wled_url(hostname, r, g, b, segment=None, effect=""):
|
|
"""Build WLED HTTP API URL."""
|
|
base = f"http://{hostname}/win&A=255&R={r}&G={g}&B={b}"
|
|
if effect:
|
|
base = f"http://{hostname}{effect}&R={r}&G={g}&B={b}"
|
|
if segment is not None:
|
|
base = f"{base}&SS={segment}"
|
|
# FX=0 nach Segment-Auswahl, damit nur das gewaehlte Segment zurueckgesetzt wird
|
|
if not effect:
|
|
base = f"{base}&FX=0"
|
|
return base
|
|
|
|
|
|
def call_wled(url):
|
|
"""Send HTTP request to WLED device. Returns (success, status_text)."""
|
|
try:
|
|
r = requests.post(url, timeout=TIMEOUT)
|
|
return True, f"{r.status_code} {r.reason}"
|
|
except requests.exceptions.RequestException as e:
|
|
logger.warning("WLED nicht erreichbar: %s - %s", url, e)
|
|
return False, str(e)
|
|
|
|
|
|
def get_wled_info(hostname):
|
|
"""Query WLED JSON API for device info. Returns dict or None."""
|
|
try:
|
|
r = requests.get(f"http://{hostname}/json/info", timeout=TIMEOUT)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
return {
|
|
"online": True,
|
|
"name": data.get("name", ""),
|
|
"version": data.get("ver", ""),
|
|
"brand": data.get("brand", "WLED"),
|
|
"mac": data.get("mac", ""),
|
|
"ip": data.get("ip", ""),
|
|
"leds": data.get("leds", {}).get("count", 0),
|
|
"power_on": data.get("state", {}).get("on", False) if "state" not in data else None,
|
|
}
|
|
return {"online": False}
|
|
except requests.exceptions.RequestException:
|
|
return {"online": False}
|
|
|
|
|
|
def get_wled_config(hostname):
|
|
"""Download full WLED config as JSON."""
|
|
try:
|
|
r = requests.get(f"http://{hostname}/cfg.json", timeout=TIMEOUT)
|
|
if r.status_code == 200:
|
|
return r.json()
|
|
return None
|
|
except requests.exceptions.RequestException:
|
|
return None
|
|
|
|
|
|
def restore_wled_config(hostname, config_json):
|
|
"""Push config JSON to WLED device."""
|
|
try:
|
|
r = requests.post(
|
|
f"http://{hostname}/json/cfg",
|
|
json=config_json,
|
|
timeout=TIMEOUT,
|
|
)
|
|
return r.status_code == 200, f"{r.status_code} {r.reason}"
|
|
except requests.exceptions.RequestException as e:
|
|
return False, str(e)
|
|
|
|
|
|
def ota_update_wled(hostname, firmware_path):
|
|
"""Push firmware binary to WLED device via OTA. Returns (success, message)."""
|
|
try:
|
|
with open(firmware_path, "rb") as f:
|
|
r = requests.post(
|
|
f"http://{hostname}/update",
|
|
files={"update": ("firmware.bin", f, "application/octet-stream")},
|
|
timeout=120,
|
|
)
|
|
if r.status_code == 200:
|
|
return True, "Update erfolgreich, Geraet startet neu"
|
|
return False, f"{r.status_code} {r.reason}"
|
|
except requests.exceptions.RequestException as e:
|
|
return False, str(e)
|