admin/Smart-Dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d557a1651fc1a662f85644101ab3b558b4527917
Smart-Dashboard/restricted/autoActions/transports.py
T
adminandClaude Opus 5 d557a1651f Neigungs-Umweg gilt fuer jedes Kommando, nicht nur setOrientation 
Die Kugelschreiber-Mechanik haengt nicht am Kommando, sondern an der
Neigung selbst - "Position+Neigung" braucht den Umweg ueber 0 % genauso wie
"Neigung". Erkannt wird er deshalb am Parameter und nicht am Kommandonamen:
beide Kommandos nennen ihren Neigungsparameter "Neigung", das
Geraetemodell liefert die Zuordnung also mit.

Geprueft im Probelauf in beiden Versendern: Position+Neigung mit 20 % geht
direkt raus, mit 90 % mit vorgeschaltetem Umweg.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-31 20:14:05 +02:00

571 lines
22 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Transporte fuer den AutoAction-Runner.
Ein Transport weiss, wie man mit einer Sorte Geraet redet - er liest deren
Messwerte und schickt deren Kommandos. Welcher zustaendig ist, entscheidet die
URL des Aktors in der Tabelle `actors`:
mqtt://... MQTT-Geraete (Home-Assistant-Discovery)
http://... Shelly und andere HTTP-Geraete
wled://... WLED-Lampen
io://, rts://, internal://, ogp:// Tahoma - das Schema haengt an der
Funkart des Geraets, deshalb wird dort nicht danach
entschieden, sondern an der Box-Kennung in der URL
Logic das gerechnete Geraet "Zeitpunkt" (Uhrzeit, Datum, Sonne)
Alle liegen in einer Datei statt in einem Paket wie bei deviceDiscovery: es
sind fuenf kurze Klassen, und wer eine sechste Geraeteart anschliesst, sieht
hier auf einen Blick, was dafuer zu tun ist.
Jeder Transport hat zwei Haelften:
zustaende_anmelden(states) einmalig beim Start
zustaende_lesen() liefert {state_id: wert} - nur was neu ist
senden(aktion) fuehrt ein Kommando aus
`states` ist eine Liste von Dicts mit actor_url, state_url, value_path und
id, `aktion` ein Dict mit actor_url, command_url und params (Liste aus
{url, name, wert}).
"""
import json
import logging
import re
import time
import urllib3
from datetime import datetime
from urllib.parse import quote
logger = logging.getLogger("autoaction.transport")
# Die Tahoma-Box hat ein selbst ausgestelltes Zertifikat auf einen Namen, den
# nur das Heimnetz kennt. Die Pruefung ist dort bewusst aus (wie in
# ajax/tahoma.php); ohne diese Zeile warnt urllib3 bei jeder einzelnen
# Abfrage und uebertoent das eigentliche Protokoll.
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
def wert_aus_pfad(daten, pfad):
"""
Einen Teilwert aus einer Nutzlast holen: "[3]", "ssid", "seg[0].col[0]".
Ein leerer Pfad heisst: die Nutzlast selbst.
Gebraucht wird das an zwei Stellen. MQTT-Geraete legen mehrere Messwerte
auf ein Topic - der go-eCharger schickt sechzehn Zahlen als JSON-Feld -,
und WLED liefert seinen gesamten Zustand als ein Dokument.
"""
if not pfad:
return daten
for teil in pfad.split("."):
treffer = re.match(r"^([^\[]*)((?:\[\d+\])*)$", teil)
if not treffer:
raise KeyError(pfad)
if treffer.group(1):
daten = daten[treffer.group(1)]
for index in re.findall(r"\[(\d+)\]", treffer.group(2)):
daten = daten[int(index)]
return daten
def uebersetze(wert, tabelle):
"""
Aus der gesendeten Zahl den Zustandsnamen machen: aus "2" wird "Charging".
Manche Geraete schicken einen Zahlencode und meinen einen Zustand. Welche
Zahl welchen Namen hat, steht in possible_values - derselben Spalte, aus
der auch der Editor seine Auswahlliste baut. Eine Bedingung vergleicht
damit genau den Klartext, den man dort ausgewaehlt hat.
Steht die Zahl nicht in der Tabelle, bleibt sie stehen: ein erfundener
Name waere schlimmer als ein roher Wert.
"""
if not tabelle:
return wert
if wert in tabelle:
return tabelle[wert]
try: # "2.0" und "2" meinen dieselbe Stufe
ganz = str(int(float(wert)))
except (TypeError, ValueError):
return wert
return tabelle.get(ganz, wert)
class Transport:
"""Gemeinsame Form. Wer nichts zu lesen hat, erbt die leeren Methoden."""
schema = ""
def passt(self, actor_url):
return actor_url.startswith(self.schema)
def zustaende_anmelden(self, states):
pass
def zustaende_lesen(self):
return {}
def senden(self, aktion):
raise NotImplementedError
# ---------------------------------------------------------------------------
# MQTT
# ---------------------------------------------------------------------------
class MQTTTransport(Transport):
"""
Die state_url ist hier das vollstaendige Topic, die parameter_url das
Kommando-Topic. Werte kommen von selbst herein und landen in einem
Zwischenspeicher, den der Runner im Takt abholt.
"""
schema = "mqtt://"
def __init__(self, client, dry_run=False):
self.client = client
self.dry_run = dry_run
self.topics = {} # topic -> [(state_id, value_path, wertetabelle), ...]
self.neu = {} # state_id -> wert
def zustaende_anmelden(self, states):
self.topics = {}
for s in states:
if not s["state_url"]:
continue
self.topics.setdefault(s["state_url"], []).append(
(s["id"], s.get("value_path"), s.get("wertetabelle") or {}))
for topic in self.topics:
self.client.subscribe(topic)
mehrfach = sum(1 for e in self.topics.values() if len(e) > 1)
logger.info("MQTT: %d Topics abonniert, %d davon mit mehreren Messwerten",
len(self.topics), mehrfach)
def nachricht(self, topic, payload):
"""Wird vom Runner aus dem on_message-Rueckruf gerufen."""
eintraege = self.topics.get(topic, [])
if not eintraege:
return
text = payload.decode("utf-8", "replace").strip() if isinstance(payload, bytes) else str(payload)
try:
daten = json.loads(text)
except ValueError:
daten = None # kein JSON - dann gilt der Rohtext
for state_id, pfad, tabelle in eintraege:
wert = self._wert(text, daten, pfad, tabelle)
if wert is not None:
self.neu[state_id] = wert
@staticmethod
def _wert(text, daten, pfad, tabelle=None):
"""
Aus der Nutzlast den Wert eines einzelnen Messwerts machen.
Mehrere Messwerte teilen sich oft ein Topic; welcher Teil gemeint ist,
steht in value_path - gelesen aus dem value_template der
Home-Assistant-Discovery. Ohne Pfad gilt die ganze Nutzlast, und
JSON-Skalare werden ausgepackt: manche Geraete schicken 21.4 mit
Anfuehrungszeichen, andere true statt ON.
Steht eine Werttabelle dabei, wird aus der gesendeten Zahl der
Zustandsname: aus 2 wird "Charging".
"""
if daten is None:
return uebersetze(text, tabelle)
try:
wert = wert_aus_pfad(daten, pfad)
except (KeyError, IndexError, TypeError):
logger.debug("Pfad %s nicht in der Nutzlast: %s", pfad, text[:80])
return None
if isinstance(wert, bool):
wert = "true" if wert else "false"
elif isinstance(wert, (int, float, str)):
wert = str(wert)
else:
return json.dumps(wert, ensure_ascii=False)
return uebersetze(wert, tabelle)
def zustaende_lesen(self):
werte, self.neu = self.neu, {}
return werte
def senden(self, aktion):
if not aktion["params"]:
# Kommando ohne Parameter: das Kommando selbst ist die Nutzlast.
self._publish(aktion["command_url"], "")
return
for p in aktion["params"]:
self._publish(p["url"] or aktion["command_url"], p["wert"])
def _publish(self, topic, nutzlast):
if not topic:
raise ValueError("Kommando ohne Topic")
if self.dry_run:
logger.info("[dry-run] MQTT %s <- %s", topic, nutzlast)
return
ergebnis = self.client.publish(topic, nutzlast, qos=1, retain=False)
ergebnis.wait_for_publish(timeout=5)
# ---------------------------------------------------------------------------
# HTTP (Shelly und Verwandte)
# ---------------------------------------------------------------------------
class HTTPTransport(Transport):
"""
Die actor_url ist der Endpunkt, die state_url ein Feldname in dessen
JSON-Antwort ("tC", "a_voltage"). Solche Geraete melden sich nicht von
selbst, sie werden im Poll-Takt gefragt.
Beim Senden wird die command_url als Abfrageargument an die actor_url
gehaengt ("turn=on") und die Parameter mit ihrem eigenen Namen dazu.
Frueher stand hier eine Uebersetzungstabelle, weil die Shelly-Kommandos
ohne URL in der Datenbank landeten - das ist im Discovery behoben, die
Zuordnung gehoert dorthin und nicht in den Runner.
"""
schema = "http"
def __init__(self, requests_modul, timeout=5, dry_run=False):
self.requests = requests_modul
self.timeout = timeout
self.dry_run = dry_run
self.states = []
def zustaende_anmelden(self, states):
self.states = [s for s in states if s["state_url"]]
logger.info("HTTP: %d Messwerte an %d Endpunkten",
len(self.states), len({s["actor_url"] for s in self.states}))
def zustaende_lesen(self):
werte = {}
# Je Endpunkt eine Anfrage, auch wenn mehrere Messwerte daran haengen.
nach_url = {}
for s in self.states:
nach_url.setdefault(s["actor_url"], []).append(s)
for url, states in nach_url.items():
try:
antwort = self.requests.get(url, timeout=self.timeout)
daten = antwort.json()
except Exception as fehler:
logger.debug("HTTP %s nicht erreichbar: %s", url, fehler)
continue
for s in states:
if isinstance(daten, dict) and s["state_url"] in daten:
werte[s["id"]] = str(daten[s["state_url"]])
return werte
def senden(self, aktion):
argumente = {}
for teil in (aktion["command_url"] or "").split("&"):
if "=" in teil:
schluessel, wert = teil.split("=", 1)
argumente[schluessel] = wert
for p in aktion["params"]:
if p["url"]:
argumente[p["url"]] = p["wert"]
if not argumente:
raise RuntimeError("Kommando ohne URL und ohne Parameter - im "
"Geraetemodell fehlt die Angabe, was zu schicken ist")
if self.dry_run:
logger.info("[dry-run] HTTP %s %s", aktion["actor_url"], argumente)
return
antwort = self.requests.get(aktion["actor_url"], params=argumente, timeout=self.timeout)
if antwort.status_code >= 400:
raise RuntimeError("HTTP %d von %s" % (antwort.status_code, aktion["actor_url"]))
# ---------------------------------------------------------------------------
# WLED
# ---------------------------------------------------------------------------
class WLEDTransport(Transport):
"""
WLED-Lampen sprechen ueber eine einzige JSON-Schnittstelle:
GET http://IP/json/state liefert den Zustand, POST dorthin setzt ihn.
Das Geraetemodell nutzt das elegant aus - die command_url ist eine
JSON-Vorlage mit Platzhaltern:
{"bri":%brightness%}
{"seg":[{"col":[[%red%,%green%,%blue%]]}]}
Gesendet wird also nicht Argument fuer Argument, sondern die ausgefuellte
Vorlage am Stueck. Deshalb haben die Parameter hier auch keine eigene URL:
ihr Name ist der Platzhalter.
Die state_url ist ein Pfad in die Antwort ("on", "bri",
"seg[0].col[0]") - dieselbe Schreibweise, die auch in current_value steht.
"""
schema = "wled://"
def __init__(self, requests_modul, timeout=5, dry_run=False):
self.requests = requests_modul
self.timeout = timeout
self.dry_run = dry_run
self.states = []
@staticmethod
def _adresse(actor_url):
return "http://" + actor_url[len("wled://"):].rstrip("/")
def zustaende_anmelden(self, states):
self.states = [s for s in states if s["state_url"]]
logger.info("WLED: %d Messwerte an %d Lampen",
len(self.states), len({s["actor_url"] for s in self.states}))
def zustaende_lesen(self):
werte = {}
nach_lampe = {}
for s in self.states:
nach_lampe.setdefault(s["actor_url"], []).append(s)
for actor_url, states in nach_lampe.items():
try:
antwort = self.requests.get(self._adresse(actor_url) + "/json/state",
timeout=self.timeout)
daten = antwort.json()
except Exception as fehler:
logger.debug("WLED %s nicht erreichbar: %s", actor_url, fehler)
continue
for s in states:
# Bei WLED ist die state_url selbst schon der Pfad. Ein
# gesetzter value_path hat trotzdem Vorrang, falls das
# Geraetemodell spaeter darauf umgestellt wird.
pfad = s.get("value_path") or s["state_url"]
try:
werte[s["id"]] = str(wert_aus_pfad(daten, pfad))
except (KeyError, IndexError, TypeError):
logger.debug("WLED %s: Pfad %s nicht gefunden", actor_url, pfad)
return werte
def senden(self, aktion):
vorlage = aktion["command_url"]
if not vorlage:
raise RuntimeError("WLED-Kommando ohne Vorlage")
for p in aktion["params"]:
vorlage = vorlage.replace("%" + p["name"] + "%", str(p["wert"]))
try:
rumpf = json.loads(vorlage)
except ValueError:
# Ein nicht ersetzter Platzhalter oder ein Textwert an einer
# Stelle, wo eine Zahl stehen muss. Lieber hier abbrechen als der
# Lampe etwas Unverstaendliches schicken.
raise RuntimeError("WLED-Vorlage ergibt kein gueltiges JSON: " + vorlage[:120])
if self.dry_run:
logger.info("[dry-run] WLED %s <- %s", aktion["actor_url"],
json.dumps(rumpf, ensure_ascii=False))
return
antwort = self.requests.post(self._adresse(aktion["actor_url"]) + "/json/state",
json=rumpf, timeout=self.timeout)
if antwort.status_code >= 400:
raise RuntimeError("WLED antwortete mit %d" % antwort.status_code)
# ---------------------------------------------------------------------------
# Tahoma
# ---------------------------------------------------------------------------
class TahomaTransport(Transport):
"""
Die actor_url ist die deviceURL, die state_url ein Statusname
("core:ClosureState"), die command_url ein Kommandoname ("setClosure").
Geschickt wird ueber exec/apply - genauso wie in ajax/tahoma.php, nur ohne
die dortige Sonderbehandlung fuer "faehrt gerade".
Zustaendig ist dieser Transport fuer alles, was die Kennung der eigenen
Box in der URL traegt. Am Schema laesst sich das nicht festmachen: es
beschreibt die Funkart, und dieselbe Box liefert io:// fuer die
Jalousien, rts:// fuer die Dachfenster und internal:// fuer die Alarm-
anlage. Ohne PIN in der config.ini ist niemand zustaendig - dann meldet
der Runner beim Ausloesen "kein Transport", statt still nichts zu tun.
"""
schema = "io://"
# Bis zu dieser Neigung fahren die Aussenjalousien direkt.
#
# Sie haben eine Kugelschreiber-Mechanik: beim Herunterfahren stehen die
# Lamellen bei etwa 30 %. In Richtung 0 % laesst sich von jeder Stellung
# aus direkt neigen; darueber hinaus muss die Mechanik erst einmal auf
# 0 % zurueck, sonst rastet sie nicht um - ein "Neigung 80 %" ohne
# diesen Umweg bleibt wirkungslos.
#
# Das gilt fuer jedes Kommando, das die Neigung setzt - "Neigung" ebenso
# wie "Position+Neigung". Erkannt wird es deshalb am Parameter und nicht
# am Kommandonamen: beide heissen ihren Neigungsparameter "Neigung".
#
# Dasselbe steht in restricted/commands.php: beide Versender brauchen es.
NEIGUNG_DIREKT_MAX = 30
# So heisst der Neigungsparameter im Geraetemodell.
NEIGUNG_PARAMETER = "Neigung"
# So lange wird hoechstens auf das Ende einer Neigungsfahrt gewartet.
# Auf core:MovingState ist dabei kein Verlass - gemessen hat die Box
# waehrend einer Neigungsfahrt kein einziges Mal "faehrt" gemeldet, der
# Zustandswert selbst ist die verlaessliche Auskunft.
# Gemessen braucht eine Jalousie von 100 % auf 0 % gut fuenfzehn Sekunden.
NEIGUNG_WARTE_SEKUNDEN = 25
def __init__(self, requests_modul, pin, token, timeout=10, dry_run=False):
self.requests = requests_modul
self.pin = pin
self.token = token
self.timeout = timeout
self.dry_run = dry_run
self.states = []
def passt(self, actor_url):
return bool(self.pin) and ("://" + self.pin + "/") in actor_url
@property
def basis(self):
return "https://gateway-%s:8443/enduser-mobile-web/1/enduserAPI" % self.pin
def _kopf(self):
return {"Content-Type": "application/json",
"Authorization": "Bearer " + self.token}
def zustaende_anmelden(self, states):
self.states = [s for s in states if s["state_url"]]
logger.info("Tahoma: %d Messwerte an %d Geraeten",
len(self.states), len({s["actor_url"] for s in self.states}))
def zustaende_lesen(self):
if not self.token:
return {}
werte = {}
nach_geraet = {}
for s in self.states:
nach_geraet.setdefault(s["actor_url"], []).append(s)
for geraet, states in nach_geraet.items():
try:
antwort = self.requests.get(
self.basis + "/setup/devices/" + quote(geraet, safe="") + "/states",
headers=self._kopf(), timeout=self.timeout, verify=False)
zustaende = {z["name"]: z.get("value") for z in antwort.json()}
except Exception as fehler:
logger.debug("Tahoma %s nicht erreichbar: %s", geraet, fehler)
continue
for s in states:
if s["state_url"] in zustaende:
werte[s["id"]] = str(zustaende[s["state_url"]])
return werte
def senden(self, aktion):
if not self.token:
raise RuntimeError("Kein Tahoma-Token in der config.ini")
# Die Reihenfolge der Parameter ist die aus command_parameters - bei
# setClosureAndOrientation also erst Position, dann Winkel.
parameter = [self._zahl(p["wert"]) for p in aktion["params"]]
neigungsziel = None
for p in aktion["params"]:
if p["name"] == self.NEIGUNG_PARAMETER:
wert = self._zahl(p["wert"])
if isinstance(wert, (int, float)):
neigungsziel = wert
# Kugelschreiber-Mechanik, siehe NEIGUNG_DIREKT_MAX. Der Umweg gehoert
# hierher und nicht in die Automatik - sonst muesste ihn jeder kennen,
# der eine Jalousie anspricht.
umweg = neigungsziel is not None and neigungsziel > self.NEIGUNG_DIREKT_MAX
if self.dry_run:
logger.info("[dry-run] Tahoma %s %s%s", aktion["command_url"], parameter,
" (mit Umweg ueber Neigung 0)" if umweg else "")
return
if umweg:
self._apply(aktion["actor_url"], "setOrientation", [0])
self._warteAufNeigung(aktion["actor_url"], 0)
self._apply(aktion["actor_url"], aktion["command_url"], parameter)
def _apply(self, actor_url, name, parameter):
"""Ein Kommando an die Box schicken."""
rumpf = {"label": "AutoAction",
"actions": [{"deviceURL": actor_url,
"commands": [{"name": name, "parameters": parameter}]}]}
antwort = self.requests.post(self.basis + "/exec/apply", headers=self._kopf(),
data=json.dumps(rumpf), timeout=self.timeout, verify=False)
if antwort.status_code >= 400:
raise RuntimeError("Tahoma antwortete mit %d: %s"
% (antwort.status_code, antwort.text[:120]))
def _warteAufNeigung(self, actor_url, ziel):
"""Wartet, bis die Neigung den Zielwert zeigt."""
for _ in range(self.NEIGUNG_WARTE_SEKUNDEN):
time.sleep(1)
try:
antwort = self.requests.get(
self.basis + "/setup/devices/" + quote(actor_url, safe="") + "/states",
headers=self._kopf(), timeout=self.timeout, verify=False)
for zustand in antwort.json():
if zustand.get("name") != "core:SlateOrientationState":
continue
wert = zustand.get("value")
# Ein fehlendes Feld darf nicht als 0 durchgehen - das
# waere ausgerechnet beim Ziel 0 ein falsches Erfolgs-
# signal.
if wert is not None and int(wert) == ziel:
return True
except Exception as fehler:
logger.debug("Neigung nicht lesbar: %s", fehler)
logger.warning("Neigung von %s hat %s%% nicht innerhalb von %d s erreicht",
actor_url, ziel, self.NEIGUNG_WARTE_SEKUNDEN)
return False
@staticmethod
def _zahl(wert):
"""Tahoma erwartet Zahlen als Zahlen, Text als Text."""
try:
return int(wert)
except (TypeError, ValueError):
pass
try:
return float(wert)
except (TypeError, ValueError):
return wert
# ---------------------------------------------------------------------------
# Logic - das gerechnete Geraet
# ---------------------------------------------------------------------------
class LogicTransport(Transport):
"""
Uhrzeit, Datum, Sonnenauf- und -untergang. Es gibt nichts zu abonnieren und
nichts zu schalten, die Werte entstehen im Takt. Sonnenzeiten kommen aus
solarLog.daylight, dieselbe Tabelle, aus der auch ajax/getSunrise.php liest.
"""
schema = "Logic"
def __init__(self, sonnenzeiten):
"""sonnenzeiten: Funktion() -> (sonnenaufgang, sonnenuntergang) als "HH:MM"."""
self.sonnenzeiten = sonnenzeiten
self.states = []
def passt(self, actor_url):
return actor_url == "Logic"
def zustaende_anmelden(self, states):
self.states = states
logger.info("Logic: %d Messwerte", len(states))
def zustaende_lesen(self):
jetzt = datetime.now()
auf, unter = self.sonnenzeiten()
tabelle = {
"time": jetzt.strftime("%H:%M"),
"date": jetzt.strftime("%d.%m.%Y"),
"sunrise": auf,
"sunset": unter,
}
return {s["id"]: tabelle[s["state_url"]]
for s in self.states if s["state_url"] in tabelle}
def senden(self, aktion):
raise RuntimeError("Das Geraet \"Zeitpunkt\" kann nichts schalten")
Reference in New Issue View Git Blame Copy Permalink