Automatischer Push: 11 Datei(en) geändert
- .vault-tools/trading-plattform-pruefung/LIESMICH.md - .vault-tools/trading-plattform-pruefung/abendkontrolle.py - .vault-tools/trading-plattform-pruefung/abendkontrolle_angriffe.py - .vault-tools/trading-plattform-pruefung/abendkontrolle_mutanten.py - .vault-tools/trading-plattform-pruefung/abendkontrolle_wsl.py - .vault-tools/trading-plattform-pruefung/browserlauf.py - .vault-tools/trading-plattform-pruefung/browserlauf_zugang.py - .vault-tools/trading-plattform-pruefung/ha_totmann.py - .vault-tools/trading-plattform-pruefung/ha_totmann_wsl.py - .vault-tools/trading-plattform-pruefung/scan_push.sh - AI/Regelwerk/codeworkflow.md Erzeugt vom PostToolUse-Hook .vault-tools/auto-push/auto_push.py. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Pascal MichelandClaude Opus 5 committed
1 parent
ea278dc24f
commit
c115a13190
11 files changed
+1848
-37
No files matched your search
@@ -0,0 +1,929 @@
|
||||
"""Richtet in Home Assistant den Totmann der Trading Plattform ein (Paket Z2, Anforderung Nr. 7).
|
||||
|
||||
Koordinator-Werkzeug, nicht Teil des Repos. Zwei Teile:
|
||||
- dieses Skript laeuft unter WINDOWS mit dem Python des HA-Projekts (dort liegen websockets,
|
||||
requests, src/ha_verbindung.py, src/config_datei.py und src/vorlage_auswerten.py; Adresse
|
||||
der Instanz und Name der Token-Variable stehen dort, nicht hier) und schreibt nach Home
|
||||
Assistant;
|
||||
- ha_totmann_wsl.py laeuft in WSL mit dem Python der Trading Plattform, liest dort die
|
||||
Geheimnisdatei und den Kalender von Alpaca und gibt Thema, Token und Kalender als eine
|
||||
JSON-Zeile zurueck. Dieses Skript liest sie nur als Pipe (aus_wsl) und gibt davon nichts
|
||||
aus, nur Laengen. Die Geheimnisdatei verlaesst WSL nie.
|
||||
Nichts in diesen Dateien ist ein Geheimnis. Thema und Token landen nur in secrets.yaml von
|
||||
Home Assistant.
|
||||
|
||||
Was der Totmann tut: Eine Automation prueft alle fuenf Minuten, ob das Lebenszeichen der
|
||||
Plattform (input_datetime.trading_gesendet; der Hausmelder schickt es spaetestens alle fuenf
|
||||
Minuten) aelter als 15 Minuten ist -- nur an Handelstagen, von 15 Minuten nach der Oeffnung
|
||||
bis zum Schluss. Dann meldet sie einmal je Stille und Handelstag ueber die HA-App (der Dienst
|
||||
ist das Argument --app-dienst, kein Literal hier) und ueber ntfy (rest_command, Adresse und
|
||||
Zugriff aus der secrets.yaml von Home Assistant).
|
||||
|
||||
Die Bedingung besteht aus vier Eingaben und einer Entscheidung (ENTSCHEIDUNG). Die Automation
|
||||
setzt die Eingaben live (EINGABEN_LIVE); trocken rechnet die Entscheidung mit gestellten
|
||||
Lagen (faelle) lokal, pruefen laesst Home Assistant dieselben Lagen per render_template
|
||||
rechnen. So ist belegt, dass Home Assistant die Entscheidung so versteht wie gemeint.
|
||||
|
||||
Handelstage und Fenster kommen beim Einrichten aus dem Kalender von Alpaca (ueber den
|
||||
WSL-Teil) fuer die naechsten --tage Tage und stehen als Epoch-Sekunden in der Bedingung. Nach
|
||||
dem letzten Datum schweigt der Totmann; der Alias nennt es, "einrichten" verlaengert. Die New
|
||||
Yorker Zeit rechnet das Werkzeug selbst (_ny_versatz_h), weil Python unter Windows ohne das
|
||||
Paket tzdata keine Zeitzonen kennt.
|
||||
|
||||
Schreibwege nach /config: secrets.yaml und configuration.yaml ueber die API des
|
||||
File-editor-Add-ons (src/config_datei.py des HA-Projekts; dessen Pfadregel erlaubt nur
|
||||
themes/, dieses Werkzeug ersetzt sie zur Laufzeit fuer genau diese zwei Dateien). Jeder
|
||||
eigene Block steht zwischen MARKE_ANFANG und MARKE_ENDE; "entfernen" nimmt genau diese
|
||||
Bloecke wieder heraus. Die Automation geht ueber /api/config/automation/config/<id>, wie in
|
||||
ha_einrichten.py.
|
||||
|
||||
Aktionen:
|
||||
pruefen lesend: Dienste, Bloecke (nur ja oder nein), Automation, Alter des
|
||||
Lebenszeichens, die Eingaben heute/jetzt/gesendet live, die juengste Spur der
|
||||
Automation (ob die Bedingung mit this ohne Fehler rechnet) und die Rechenprobe
|
||||
der Bedingung in Home Assistant
|
||||
trocken ohne Netz: baut Automation und Bloecke aus erfundenen Werten, rechnet die
|
||||
gestellten Lagen lokal und prueft die Regeln des WSL-Teils
|
||||
einrichten holt Thema, Token und Kalender aus WSL, schreibt die Bloecke (wenn noetig),
|
||||
prueft die Konfiguration, laedt rest_command neu oder verlangt einen Neustart
|
||||
von Home Assistant (Exit 4) und legt die Automation an oder ersetzt sie
|
||||
probe ruft rest_command und den App-Dienst je einmal -- zwei echte Meldungen
|
||||
entfernen loescht die Automation und nimmt beide Bloecke heraus
|
||||
|
||||
Aufruf: <ha-projekt>\\.venv\\Scripts\\python.exe ha_totmann.py <aktion> [<ha-projektordner>
|
||||
<trading-projektordner>] [--app-dienst notify.<name>] [--tage 400]
|
||||
Exit: 0 fertig, 2 Abbruch oder BEFUND, 3 Voraussetzung fehlt, 4 Neustart von HA noetig.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import tomllib
|
||||
from datetime import date
|
||||
from datetime import datetime
|
||||
from datetime import timedelta
|
||||
from datetime import timezone
|
||||
from pathlib import Path
|
||||
from pathlib import PureWindowsPath
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
import ha_totmann_wsl as wsl_teil # noqa: E402
|
||||
|
||||
AUTOMATION_ID = "trading_plattform_totmann"
|
||||
ALIAS = "Trading Plattform: Totmann"
|
||||
LEBENSZEICHEN = "input_datetime.trading_gesendet"
|
||||
REST_DIENST = "trading_totmann_ntfy"
|
||||
SECRET_URL = "trading_totmann_ntfy_url"
|
||||
SECRET_AUTH = "trading_totmann_ntfy_auth"
|
||||
MARKE_ANFANG = "# >>> trading-totmann (ha_totmann.py, Paket Z2)"
|
||||
MARKE_ENDE = "# <<< trading-totmann"
|
||||
ERLAUBT = ("secrets.yaml", "configuration.yaml")
|
||||
STILLE_S = 900
|
||||
# Wie weit die Uhr von Home Assistant von der dieses Rechners abweichen darf, bevor
|
||||
# eingaben_befunde es als Befund nennt.
|
||||
UHR_TOLERANZ_S = 120
|
||||
PRUEFTAKT = "/5"
|
||||
APP_DIENST_MUSTER = re.compile(r"^notify\.[a-z0-9_]+$")
|
||||
FREMDES_REST_COMMAND = re.compile(r"(?m)^rest_command:")
|
||||
WSL = r"C:\Windows\System32\wsl.exe"
|
||||
WSL_PY = "/home/claude/venvs/trading/bin/python"
|
||||
WSL_TEIL = Path(__file__).resolve().with_name("ha_totmann_wsl.py")
|
||||
WSL_TIMEOUT_S = 120
|
||||
TEXT = (
|
||||
"Die Trading Plattform schweigt im Handelsfenster seit mehr als 15 Minuten. Letztes "
|
||||
"Lebenszeichen: {{ states('" + LEBENSZEICHEN + "') }}. Nachsehen: Rechner, WSL, Dashboard."
|
||||
)
|
||||
PROBE_TEXT = "Probe des Totmanns (ha_totmann.py probe): Diese Meldung kam an. Nichts zu tun."
|
||||
# Erfundene Werte fuer trocken. Kein echtes Thema, kein echter Dienst, kein echter Token.
|
||||
TROCKEN_THEMA = "probe-thema-nicht-echt"
|
||||
TROCKEN_TOKEN = "tk_probe_nicht_echt"
|
||||
TROCKEN_DIENST = "notify.probe_dienst"
|
||||
|
||||
# Die vier Eingaben der Bedingung, wie die Automation sie liest. this ist die Automation
|
||||
# selbst; ihr last_triggered ist die letzte Ausloesung.
|
||||
EINGABEN_LIVE = {
|
||||
"heute": "now().strftime('%Y-%m-%d')",
|
||||
"jetzt": "now().timestamp()",
|
||||
"gesendet": "state_attr('" + LEBENSZEICHEN + "', 'timestamp') | float(0)",
|
||||
"zuletzt": "as_timestamp(this.attributes.last_triggered, 0)",
|
||||
}
|
||||
# Die Entscheidung, gemeinsam fuer Automation, trocken und pruefen. Ausloesen, wenn heute
|
||||
# ein Handelstag ist, jetzt im Fenster liegt, das Lebenszeichen aelter als STILLE_S ist und
|
||||
# die letzte Ausloesung vor dem Lebenszeichen lag -- oder vor dem heutigen Fenster: Haelt
|
||||
# eine Stille ueber Nacht an, meldet der neue Handelstag sie einmal.
|
||||
ENTSCHEIDUNG = (
|
||||
"f is not none and f[0] <= jetzt <= f[1] and jetzt - gesendet > "
|
||||
+ str(STILLE_S)
|
||||
+ " and (zuletzt <= gesendet or zuletzt < f[0])"
|
||||
)
|
||||
# Ein erfundener Kalender fuer trocken und die Rechenprobe in pruefen: zwei gewoehnliche Tage
|
||||
# und der verkuerzte Tag nach Thanksgiving 2026.
|
||||
PROBE_KALENDER = [
|
||||
{"date": "2026-09-28", "open": "09:30", "close": "16:00"},
|
||||
{"date": "2026-09-29", "open": "09:30", "close": "16:00"},
|
||||
{"date": "2026-11-27", "open": "09:30", "close": "13:00"},
|
||||
]
|
||||
|
||||
|
||||
class Voraussetzung(Exception):
|
||||
"""Etwas fehlt fuer den Lauf. Der Text traegt keinen Wert aus der Geheimnisdatei."""
|
||||
|
||||
|
||||
def _ny_versatz_h(tag: date) -> int:
|
||||
"""UTC-Versatz der New Yorker Boersenzeit an einem Tag: -4 in der Sommerzeit, sonst -5.
|
||||
|
||||
US-Regel seit 2007: vom zweiten Sonntag im Maerz bis zum ersten Sonntag im November.
|
||||
Umgestellt wird sonntags um 02:00; an einem Handelstag gilt der Versatz des ganzen Tages.
|
||||
"""
|
||||
maerz = date(tag.year, 3, 1)
|
||||
zweiter_sonntag_maerz = maerz + timedelta(days=(6 - maerz.weekday()) % 7 + 7)
|
||||
november = date(tag.year, 11, 1)
|
||||
erster_sonntag_november = november + timedelta(days=(6 - november.weekday()) % 7)
|
||||
return -4 if zweiter_sonntag_maerz <= tag < erster_sonntag_november else -5
|
||||
|
||||
|
||||
def fenster_je_tag(kalender: list[dict]) -> dict[str, list[int]]:
|
||||
"""Je Handelstag [Oeffnung plus STILLE_S, Schluss] als Epoch-Sekunden."""
|
||||
fenster: dict[str, list[int]] = {}
|
||||
for eintrag in kalender:
|
||||
tag = date.fromisoformat(eintrag["date"])
|
||||
zone = timezone(timedelta(hours=_ny_versatz_h(tag)))
|
||||
grenzen = []
|
||||
for feld in ("open", "close"):
|
||||
stunde, minute = (int(teil) for teil in eintrag[feld].split(":"))
|
||||
grenzen.append(
|
||||
int(datetime(tag.year, tag.month, tag.day, stunde, minute, tzinfo=zone).timestamp())
|
||||
)
|
||||
fenster[tag.isoformat()] = [grenzen[0] + STILLE_S, grenzen[1]]
|
||||
return fenster
|
||||
|
||||
|
||||
def gestellt(heute: str, jetzt: int, gesendet: int, zuletzt: int) -> dict[str, str]:
|
||||
"""Gestellte Eingaben als Vorlagen-Literale, fuer die Rechenprobe in pruefen."""
|
||||
return {
|
||||
"heute": "'" + heute + "'",
|
||||
"jetzt": str(jetzt),
|
||||
"gesendet": str(gesendet),
|
||||
"zuletzt": str(zuletzt),
|
||||
}
|
||||
|
||||
|
||||
def bedingung(fenster: dict[str, list[int]], eingaben: dict[str, str] | None = None) -> str:
|
||||
"""Die Vorlage der Bedingung; ohne eingaben die der Automation (EINGABEN_LIVE)."""
|
||||
werte = EINGABEN_LIVE if eingaben is None else eingaben
|
||||
return (
|
||||
"{% set fenster = " + json.dumps(fenster, separators=(",", ":")) + " %}"
|
||||
+ "".join(
|
||||
"{% set " + name + " = " + werte[name] + " %}"
|
||||
for name in ("heute", "jetzt", "gesendet", "zuletzt")
|
||||
)
|
||||
+ "{% set f = fenster.get(heute) %}"
|
||||
+ "{{ " + ENTSCHEIDUNG + " }}"
|
||||
)
|
||||
|
||||
|
||||
def _ohne_eingaben(vorlage: str) -> str:
|
||||
return re.sub(r"\{% set (heute|jetzt|gesendet|zuletzt) = .*? %\}", "", vorlage)
|
||||
|
||||
|
||||
def _lage(heute: str, jetzt: int, gesendet: int, zuletzt: int) -> dict:
|
||||
return {"heute": heute, "jetzt": jetzt, "gesendet": gesendet, "zuletzt": zuletzt}
|
||||
|
||||
|
||||
def faelle(fenster: dict[str, list[int]]) -> list[tuple[str, dict, bool]]:
|
||||
"""Gestellte Lagen: Name, Eingaben, ob die Bedingung ausloesen soll.
|
||||
|
||||
Montag 28.09. und Dienstag 29.09.2026 aus PROBE_KALENDER, alle Zeiten erfunden. mitte
|
||||
liegt eine Stunde nach dem Beginn des Montagsfensters.
|
||||
"""
|
||||
montag, dienstag = fenster["2026-09-28"], fenster["2026-09-29"]
|
||||
mitte = montag[0] + 3600
|
||||
return [
|
||||
("stille_im_fenster", _lage("2026-09-28", mitte, mitte - 1800, 0), True),
|
||||
("lebenszeichen_frisch", _lage("2026-09-28", mitte, mitte - 300, 0), False),
|
||||
(
|
||||
"schon_gemeldet_nach_dem_letzten_lebenszeichen",
|
||||
_lage("2026-09-28", mitte, mitte - 1800, mitte - 600),
|
||||
False,
|
||||
),
|
||||
("vor_dem_fenster", _lage("2026-09-28", montag[0] - 60, montag[0] - 7200, 0), False),
|
||||
("nach_dem_schluss", _lage("2026-09-28", montag[1] + 600, mitte, 0), False),
|
||||
("kein_handelstag", _lage("2026-09-27", mitte - 86400, 0, 0), False),
|
||||
(
|
||||
"stille_ueber_die_tagesgrenze",
|
||||
_lage("2026-09-29", dienstag[0] + 600, mitte - 1800, mitte - 600),
|
||||
True,
|
||||
),
|
||||
(
|
||||
"am_neuen_tag_schon_gemeldet",
|
||||
_lage("2026-09-29", dienstag[0] + 1200, mitte - 1800, dienstag[0] + 600),
|
||||
False,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def _entscheide_lokal(fenster: dict[str, list[int]], lage: dict) -> bool:
|
||||
# ENTSCHEIDUNG ist eine Konstante dieser Datei, kein fremder Text; eval ohne Builtins,
|
||||
# none steht fuer das None der Vorlagensprache.
|
||||
return bool(
|
||||
eval( # noqa: S307
|
||||
ENTSCHEIDUNG,
|
||||
{"__builtins__": {}},
|
||||
{
|
||||
"none": None,
|
||||
"f": fenster.get(lage["heute"]),
|
||||
"jetzt": lage["jetzt"],
|
||||
"gesendet": lage["gesendet"],
|
||||
"zuletzt": lage["zuletzt"],
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _wahrheit(wert: object) -> bool | None:
|
||||
if isinstance(wert, bool):
|
||||
return wert
|
||||
if isinstance(wert, str) and wert.strip().lower() in ("true", "false"):
|
||||
return wert.strip().lower() == "true"
|
||||
return None
|
||||
|
||||
|
||||
def automation(fenster: dict[str, list[int]], app_dienst: str) -> dict:
|
||||
letzter = max(fenster) if fenster else "keiner"
|
||||
return {
|
||||
"alias": f"{ALIAS} (Handelstage bis {letzter})",
|
||||
"description": (
|
||||
"Meldet ueber die HA-App und ntfy, wenn das Lebenszeichen der Trading Plattform "
|
||||
f"({LEBENSZEICHEN}) im Handelsfenster aelter als 15 Minuten ist; einmal je "
|
||||
f"Stille und Handelstag. Handelstage aus dem Kalender von Alpaca bis {letzter}. "
|
||||
"Angelegt ueber ha_totmann.py (Paket Z2); Rueckbau mit der Aktion entfernen."
|
||||
),
|
||||
"trigger": [{"platform": "time_pattern", "minutes": PRUEFTAKT}],
|
||||
"condition": [{"condition": "template", "value_template": bedingung(fenster)}],
|
||||
"action": [
|
||||
{
|
||||
"service": f"rest_command.{REST_DIENST}",
|
||||
"data": {"nachricht": TEXT},
|
||||
"continue_on_error": True,
|
||||
},
|
||||
{
|
||||
"service": app_dienst,
|
||||
"data": {"title": "Trading Totmann", "message": TEXT},
|
||||
"continue_on_error": True,
|
||||
},
|
||||
],
|
||||
"mode": "single",
|
||||
"max_exceeded": "silent",
|
||||
}
|
||||
|
||||
|
||||
def config_block(mit_token: bool) -> str:
|
||||
"""Der Block fuer configuration.yaml. Kein Thema: Adresse und Zugriff kommen per !secret."""
|
||||
zeilen = [
|
||||
MARKE_ANFANG,
|
||||
"# Totmann der Trading Plattform: Meldung ueber ntfy. Adresse und Zugriff stehen in",
|
||||
"# secrets.yaml; hier steht kein Thema.",
|
||||
"rest_command:",
|
||||
f" {REST_DIENST}:",
|
||||
f" url: !secret {SECRET_URL}",
|
||||
" method: POST",
|
||||
' content_type: "text/plain; charset=utf-8"',
|
||||
" timeout: 10",
|
||||
' payload: "{{ nachricht }}"',
|
||||
" headers:",
|
||||
' Title: "Trading Totmann"',
|
||||
' Priority: "high"',
|
||||
' Tags: "rotating_light"',
|
||||
]
|
||||
if mit_token:
|
||||
zeilen.append(f" Authorization: !secret {SECRET_AUTH}")
|
||||
zeilen.append(MARKE_ENDE)
|
||||
return "\n".join(zeilen) + "\n"
|
||||
|
||||
|
||||
def secrets_block(url: str, token: str) -> str:
|
||||
"""Der Block fuer secrets.yaml: die ganze Adresse mit Thema, dazu der Zugriff."""
|
||||
zeilen = [MARKE_ANFANG, f'{SECRET_URL}: "{url}"']
|
||||
if token:
|
||||
zeilen.append(f'{SECRET_AUTH}: "Bearer {token}"')
|
||||
zeilen.append(MARKE_ENDE)
|
||||
return "\n".join(zeilen) + "\n"
|
||||
|
||||
|
||||
def ohne_block(text: str) -> str:
|
||||
"""Der Text ohne den eigenen Block zwischen den Marken, mit genau einem Umbruch am Ende."""
|
||||
muster = re.compile(
|
||||
r"\n*" + re.escape(MARKE_ANFANG) + r".*?" + re.escape(MARKE_ENDE) + r"\n?", re.DOTALL
|
||||
)
|
||||
rest = muster.sub("\n", text).rstrip("\n")
|
||||
return rest + "\n" if rest else ""
|
||||
|
||||
|
||||
def anhaengen(text: str, block: str) -> str:
|
||||
rest = text.rstrip("\n")
|
||||
return (rest + "\n\n" if rest else "") + block
|
||||
|
||||
|
||||
def _ohne(text: str, *geheim: str) -> str:
|
||||
for wert in geheim:
|
||||
if wert:
|
||||
text = text.replace(wert, "***")
|
||||
return text
|
||||
|
||||
|
||||
def _wsl_pfad(pfad: str) -> str:
|
||||
"""Ein Windows-Pfad mit Laufwerksbuchstaben als Pfad unter /mnt in WSL."""
|
||||
windows = PureWindowsPath(pfad)
|
||||
if len(windows.drive) != 2 or not windows.drive.endswith(":"):
|
||||
raise Voraussetzung(f"kein Laufwerkspfad: {pfad}")
|
||||
return "/mnt/" + windows.drive[0].lower() + "/" + "/".join(windows.parts[1:])
|
||||
|
||||
|
||||
def pruefe_wsl_ausgabe(daten: object) -> dict:
|
||||
"""Die Zeile des WSL-Teils, nach denselben Regeln geprueft, die er selbst anlegt."""
|
||||
if not isinstance(daten, dict) or set(daten) != {"thema", "token", "kalender"}:
|
||||
raise Voraussetzung("WSL-Teil: Ausgabe in unerwarteter Form")
|
||||
thema, token = wsl_teil.zugang({"ntfy": {"thema": daten["thema"], "token": daten["token"]}})
|
||||
return {"thema": thema, "token": token, "kalender": wsl_teil.pruefe_kalender(daten["kalender"])}
|
||||
|
||||
|
||||
def aus_wsl(tage: int) -> dict:
|
||||
"""Startet den WSL-Teil und liest seine eine Zeile als Pipe. stdout wird nie ausgegeben."""
|
||||
lauf = subprocess.run(
|
||||
[
|
||||
WSL, "-d", "Debian", "-u", "claude", "--exec",
|
||||
WSL_PY, _wsl_pfad(str(WSL_TEIL)), "--tage", str(tage),
|
||||
],
|
||||
capture_output=True,
|
||||
timeout=WSL_TIMEOUT_S,
|
||||
)
|
||||
if lauf.returncode != 0:
|
||||
# stderr des WSL-Teils traegt nur Befunde ohne Werte.
|
||||
fehler = lauf.stderr.decode("utf-8", errors="replace").replace("\x00", "").strip()
|
||||
raise Voraussetzung(
|
||||
f"WSL-Teil endete mit Exit {lauf.returncode}: {fehler[:300] or 'ohne Meldung'}"
|
||||
)
|
||||
zeilen = [
|
||||
zeile
|
||||
for zeile in lauf.stdout.decode("utf-8", errors="replace").split("\n")
|
||||
if zeile.startswith("{")
|
||||
]
|
||||
if len(zeilen) != 1:
|
||||
raise Voraussetzung(f"WSL-Teil: {len(zeilen)} statt genau einer JSON-Zeile")
|
||||
return pruefe_wsl_ausgabe(json.loads(zeilen[0]))
|
||||
|
||||
|
||||
def trocken() -> int:
|
||||
"""Prueft die eigenen Bausteine ohne Netz, mit erfundenen Werten."""
|
||||
befunde = []
|
||||
for tag, erwartet in (
|
||||
(date(2026, 3, 6), -5),
|
||||
(date(2026, 3, 9), -4),
|
||||
(date(2026, 10, 30), -4),
|
||||
(date(2026, 11, 2), -5),
|
||||
(date(2027, 3, 12), -5),
|
||||
(date(2027, 3, 15), -4),
|
||||
):
|
||||
if _ny_versatz_h(tag) != erwartet:
|
||||
befunde.append(f"Versatz am {tag}: {_ny_versatz_h(tag)} statt {erwartet}")
|
||||
fenster = fenster_je_tag(PROBE_KALENDER)
|
||||
# 09:45 EDT ist 13:45 UTC, 16:00 EDT 20:00 UTC; 09:45 EST ist 14:45 UTC, 13:00 EST 18:00 UTC.
|
||||
for tag, soll in (("2026-09-28", ["13:45", "20:00"]), ("2026-11-27", ["14:45", "18:00"])):
|
||||
ist = [datetime.fromtimestamp(s, timezone.utc).strftime("%H:%M") for s in fenster[tag]]
|
||||
if ist != soll:
|
||||
befunde.append(f"Fenster am {tag}: {ist} statt {soll}")
|
||||
# Die Bedingung: jede gestellte Lage lokal, mit derselben ENTSCHEIDUNG wie in HA.
|
||||
lagen = faelle(fenster)
|
||||
for name, lage, soll in lagen:
|
||||
if _entscheide_lokal(fenster, lage) is not soll:
|
||||
befunde.append(f"Bedingung {name}: soll {'ausloesen' if soll else 'schweigen'}")
|
||||
live = bedingung(fenster)
|
||||
for name, ausdruck in EINGABEN_LIVE.items():
|
||||
if "{% set " + name + " = " + ausdruck + " %}" not in live:
|
||||
befunde.append(f"die Automation liest {name} nicht wie vorgesehen")
|
||||
if _ohne_eingaben(live) != _ohne_eingaben(bedingung(fenster, gestellt(**lagen[0][1]))):
|
||||
befunde.append("die Rechenprobe entschiede mit einem anderen Ausdruck als die Automation")
|
||||
roh = json.dumps(automation(fenster, TROCKEN_DIENST))
|
||||
anders = json.dumps(automation(fenster, "notify.anderer_dienst"))
|
||||
konf = config_block(True)
|
||||
geheim = secrets_block("https://ntfy.example/" + TROCKEN_THEMA, TROCKEN_TOKEN)
|
||||
if TROCKEN_THEMA in roh or TROCKEN_THEMA in konf:
|
||||
befunde.append("das Thema stuende in der Automation oder in configuration.yaml")
|
||||
if TROCKEN_TOKEN in roh or TROCKEN_TOKEN in konf:
|
||||
befunde.append("der Token stuende in der Automation oder in configuration.yaml")
|
||||
if TROCKEN_THEMA not in geheim or f"Bearer {TROCKEN_TOKEN}" not in geheim:
|
||||
befunde.append("Thema oder Token fehlen im Block fuer secrets.yaml")
|
||||
if roh.count(TROCKEN_DIENST) != 1 or "notify.anderer_dienst" not in anders:
|
||||
befunde.append("der App-Dienst kommt nicht aus dem Argument")
|
||||
if TROCKEN_DIENST in anders:
|
||||
befunde.append("der App-Dienst steht fest in der Automation")
|
||||
if f"!secret {SECRET_URL}" not in konf or f"!secret {SECRET_AUTH}" not in konf:
|
||||
befunde.append("configuration.yaml nennt die Geheimnisse nicht per !secret")
|
||||
if f"!secret {SECRET_AUTH}" in config_block(False):
|
||||
befunde.append("ohne Token verwiese configuration.yaml auf ein fehlendes Geheimnis")
|
||||
if ohne_block(anhaengen("a: 1\n", konf)) != "a: 1\n":
|
||||
befunde.append("entfernen liesse vom Block etwas stehen")
|
||||
if ohne_block(konf) != "":
|
||||
befunde.append("eine Datei nur mit dem Block bliebe nicht leer")
|
||||
if not FREMDES_REST_COMMAND.search(ohne_block("x: 1\nrest_command:\n y: 2\n")):
|
||||
befunde.append("ein fremdes rest_command wuerde uebersehen")
|
||||
if FREMDES_REST_COMMAND.search(ohne_block(anhaengen("x: 1\n", konf))):
|
||||
befunde.append("der eigene Block gaelte als fremdes rest_command")
|
||||
# Der WSL-Teil, ohne ihn zu starten: seine Regeln und der Weg ueber die Pipe.
|
||||
if _wsl_pfad(r"D:\Claude\.vault-tools\x\y.py") != "/mnt/d/Claude/.vault-tools/x/y.py":
|
||||
befunde.append("der Pfad fuer WSL ist falsch umgesetzt")
|
||||
try:
|
||||
wsl_teil.zugang({"ntfy": {"thema": "kein thema!", "token": ""}})
|
||||
befunde.append("ein Thema ausserhalb der Regel ginge durch")
|
||||
except wsl_teil.Befund as befund:
|
||||
if "kein thema" in str(befund):
|
||||
befunde.append("der Befund des WSL-Teils nennt den Wert")
|
||||
try:
|
||||
wsl_teil.pruefe_kalender([{"date": "2026-09-28", "open": "09:30", "close": "4pm"}])
|
||||
befunde.append("ein Kalender mit falscher Uhrzeit ginge durch")
|
||||
except wsl_teil.Befund:
|
||||
pass
|
||||
try:
|
||||
zurueck = pruefe_wsl_ausgabe(
|
||||
json.loads(wsl_teil.ausgabe(TROCKEN_THEMA, TROCKEN_TOKEN, PROBE_KALENDER))
|
||||
)
|
||||
if zurueck != {"thema": TROCKEN_THEMA, "token": TROCKEN_TOKEN, "kalender": PROBE_KALENDER}:
|
||||
befunde.append("die Zeile des WSL-Teils kommt nicht unveraendert an")
|
||||
except (Voraussetzung, wsl_teil.Befund) as fehler:
|
||||
befunde.append(f"die Zeile des WSL-Teils wird abgelehnt: {fehler}")
|
||||
try:
|
||||
pruefe_wsl_ausgabe({"thema": TROCKEN_THEMA})
|
||||
befunde.append("eine unvollstaendige Zeile des WSL-Teils ginge durch")
|
||||
except Voraussetzung:
|
||||
pass
|
||||
# Die Plausibilitaet der Live-Eingaben, mit gestellten Antworten: Mittag in Berlin.
|
||||
jetzt_probe = datetime(2026, 9, 28, 10, 0, tzinfo=timezone.utc).timestamp()
|
||||
gut = {
|
||||
"heute": datetime.fromtimestamp(jetzt_probe).strftime("%Y-%m-%d"),
|
||||
"jetzt": jetzt_probe,
|
||||
"gesendet": jetzt_probe - 100,
|
||||
}
|
||||
if eingaben_befunde(gut, jetzt_probe - 100, jetzt_probe):
|
||||
befunde.append("plausible Live-Eingaben gaelten als Befund")
|
||||
for name, schlecht, zeichen in (
|
||||
("gesendet_null", dict(gut, gesendet=0.0), None),
|
||||
("uhr_daneben", dict(gut, jetzt=jetzt_probe + 600), jetzt_probe - 100),
|
||||
("falscher_tag", dict(gut, heute="2026-09-27"), jetzt_probe - 100),
|
||||
("anderer_stempel", gut, jetzt_probe - 5000),
|
||||
("jetzt_kein_wert", dict(gut, jetzt=None), jetzt_probe - 100),
|
||||
):
|
||||
if not eingaben_befunde(schlecht, zeichen, jetzt_probe):
|
||||
befunde.append(f"Live-Eingaben {name} gingen durch")
|
||||
# Die Spur, mit gestellten Antworten von trace/list und trace/get.
|
||||
in_ordnung = {"timestamp": {"start": "x"}, "trace": {"condition/0": [{"result": {"result": False}}]}}
|
||||
for name, liste, lauf, soll in (
|
||||
("ohne_lauf", [], None, False),
|
||||
("in_ordnung", [{"run_id": "1"}], in_ordnung, False),
|
||||
("renderfehler", [{"run_id": "1"}], {"trace": {"condition/0": [{"error": "UndefinedError"}]}}, True),
|
||||
("ohne_bedingung", [{"run_id": "1"}], {"trace": {}}, True),
|
||||
("ohne_wahrheitswert", [{"run_id": "1"}], {"trace": {"condition/0": [{"result": {}}]}}, True),
|
||||
):
|
||||
if spur_befund(liste, lauf)[1] is not soll:
|
||||
befunde.append(f"Spur {name}: soll {'Befund' if soll else 'kein Befund'}")
|
||||
for befund in befunde:
|
||||
print(f"BEFUND {befund}")
|
||||
print(
|
||||
f"trocken: {len(fenster)} Tage, {len(lagen)} Lagen, Automation {len(roh)} Zeichen, "
|
||||
f"{len(befunde)} Befund(e)"
|
||||
)
|
||||
return 2 if befunde else 0
|
||||
|
||||
|
||||
def _nur_zwei_dateien(pfad: object, *, datei: bool) -> str:
|
||||
"""Ersetzt config_datei.pruefe_pfad fuer dieses Werkzeug: genau zwei Dateien in /config."""
|
||||
from config_datei import PfadVerboten
|
||||
|
||||
if pfad not in ERLAUBT:
|
||||
raise PfadVerboten(f"Pfad {pfad!r} ist keine der Dateien {ERLAUBT}")
|
||||
return str(pfad)
|
||||
|
||||
|
||||
def _dienste(ha) -> dict:
|
||||
return ha.befehl_ok("get_services") or {}
|
||||
|
||||
|
||||
def bedingung_pruefen(ha) -> int:
|
||||
"""Die Rechenprobe: Home Assistant rechnet jede gestellte Lage per render_template.
|
||||
|
||||
Lesend (render_template steht in der Leseliste des HA-Projekts). Gerechnet wird mit dem
|
||||
erfundenen PROBE_KALENDER und derselben ENTSCHEIDUNG wie in der Automation; nur die vier
|
||||
Eingaben sind gestellt. Zurueck kommt die Zahl der Befunde.
|
||||
"""
|
||||
from vorlage_auswerten import auswerten
|
||||
|
||||
fenster = fenster_je_tag(PROBE_KALENDER)
|
||||
befunde = 0
|
||||
for name, lage, soll in faelle(fenster):
|
||||
nachrichten = ha.roh_sammeln(
|
||||
"render_template",
|
||||
5.0,
|
||||
template=bedingung(fenster, gestellt(**lage)),
|
||||
strict=True,
|
||||
report_errors=True,
|
||||
)
|
||||
art, wert = auswerten(nachrichten)
|
||||
ist = _wahrheit(wert) if art == "ERGEBNIS" else None
|
||||
zeile = (
|
||||
f"Bedingung {name}: soll {'ausloesen' if soll else 'schweigen'}, "
|
||||
f"Home Assistant: {art} {wert!r}"
|
||||
)
|
||||
if ist is soll:
|
||||
print(zeile)
|
||||
else:
|
||||
befunde += 1
|
||||
print("BEFUND " + zeile)
|
||||
return befunde
|
||||
|
||||
|
||||
def eingaben_befunde(werte: dict, zeichen_ha: object, jetzt_hier: float) -> list[str]:
|
||||
"""Plausibilitaet der drei live gerechneten Eingaben ohne this (Nachkritik 2026-09-24).
|
||||
|
||||
werte: heute, jetzt und gesendet, wie render_template sie liefert. zeichen_ha: das
|
||||
Attribut timestamp des Lebenszeichens aus get_states. jetzt_hier: time.time() dieses
|
||||
Rechners; er steht in derselben Zeitzone wie Home Assistant (Berlin).
|
||||
"""
|
||||
befunde = []
|
||||
try:
|
||||
jetzt = float(werte["jetzt"])
|
||||
except (TypeError, ValueError, KeyError):
|
||||
return [f"jetzt ist keine Zahl: {werte.get('jetzt')!r}"]
|
||||
if abs(jetzt - jetzt_hier) > UHR_TOLERANZ_S:
|
||||
befunde.append(f"jetzt weicht {jetzt - jetzt_hier:+.0f} s von der Uhr dieses Rechners ab")
|
||||
heute_hier = datetime.fromtimestamp(jetzt).strftime("%Y-%m-%d")
|
||||
if str(werte.get("heute")) != heute_hier:
|
||||
befunde.append(f"heute ist {werte.get('heute')!r}, nach jetzt in Ortszeit {heute_hier}")
|
||||
try:
|
||||
gesendet = float(werte["gesendet"])
|
||||
except (TypeError, ValueError, KeyError):
|
||||
return befunde + [f"gesendet ist keine Zahl: {werte.get('gesendet')!r}"]
|
||||
if gesendet <= 0.0:
|
||||
befunde.append("gesendet ist 0: das Attribut timestamp des Lebenszeichens fehlt oder ist leer")
|
||||
elif not isinstance(zeichen_ha, (int, float)) or abs(gesendet - float(zeichen_ha)) > 1.0:
|
||||
befunde.append(f"gesendet ist nicht der Zeitstempel des Lebenszeichens ({zeichen_ha!r})")
|
||||
elif gesendet > jetzt + UHR_TOLERANZ_S:
|
||||
befunde.append("gesendet liegt in der Zukunft")
|
||||
return befunde
|
||||
|
||||
|
||||
def eingaben_pruefen(ha, zeichen_ha: object) -> int:
|
||||
"""Rechnet heute, jetzt und gesendet live per render_template, wie die Automation sie liest.
|
||||
|
||||
zuletzt liest this und ist nur im Lauf der Automation definiert; das belegt spur_pruefen.
|
||||
Zurueck kommt die Zahl der Befunde.
|
||||
"""
|
||||
from vorlage_auswerten import auswerten
|
||||
|
||||
werte: dict[str, object] = {}
|
||||
for name in ("heute", "jetzt", "gesendet"):
|
||||
nachrichten = ha.roh_sammeln(
|
||||
"render_template",
|
||||
5.0,
|
||||
template="{{ " + EINGABEN_LIVE[name] + " }}",
|
||||
strict=True,
|
||||
report_errors=True,
|
||||
)
|
||||
art, wert = auswerten(nachrichten)
|
||||
if art != "ERGEBNIS":
|
||||
print(f"BEFUND Eingabe {name}: Home Assistant: {art} {wert!r}")
|
||||
return 1
|
||||
werte[name] = wert
|
||||
befunde = eingaben_befunde(werte, zeichen_ha, time.time())
|
||||
for befund in befunde:
|
||||
print(f"BEFUND Eingabe {befund}")
|
||||
if not befunde:
|
||||
abstand = float(werte["jetzt"]) - float(werte["gesendet"])
|
||||
print(f"Eingaben live: heute {werte['heute']}, jetzt und gesendet plausibel (Abstand {abstand:.0f} s)")
|
||||
return len(befunde)
|
||||
|
||||
|
||||
def spur_befund(liste: object, lauf: object) -> tuple[str, bool]:
|
||||
"""Die juengste Spur der Automation: (Zeile, ob Befund).
|
||||
|
||||
liste ist die Antwort von trace/list, lauf die von trace/get fuer deren juengsten Lauf.
|
||||
Belegt, was render_template nicht kann: dass die Bedingung im Lauf der Automation mit
|
||||
this rechnet (Home Assistant 2026.9.2: automation/__init__.py:707-711 setzt this,
|
||||
helpers/trace.py:113-118 schreibt error oder result in den Schritt).
|
||||
"""
|
||||
if not isinstance(liste, list) or not liste:
|
||||
return "Spur: noch kein Lauf (die Automation prueft alle fuenf Minuten)", False
|
||||
if not isinstance(lauf, dict) or not isinstance(lauf.get("trace"), dict):
|
||||
return f"Spur: Form unbekannt ({type(lauf).__name__})", True
|
||||
schritte = lauf["trace"].get("condition/0")
|
||||
if not isinstance(schritte, list) or not schritte or not isinstance(schritte[0], dict):
|
||||
return "Spur: kein Schritt condition/0 im juengsten Lauf", True
|
||||
schritt = schritte[0]
|
||||
if "error" in schritt:
|
||||
return f"Spur: Bedingung mit Fehler: {str(schritt['error'])[:200]}", True
|
||||
ergebnis = schritt.get("result")
|
||||
wert = ergebnis.get("result") if isinstance(ergebnis, dict) else None
|
||||
if not isinstance(wert, bool):
|
||||
return f"Spur: Bedingung ohne Wahrheitswert ({str(ergebnis)[:200]})", True
|
||||
stempel = lauf.get("timestamp")
|
||||
start = stempel.get("start", "?") if isinstance(stempel, dict) else "?"
|
||||
return f"Spur: juengster Lauf {start}, Bedingung {wert}, kein Fehler", False
|
||||
|
||||
|
||||
def spur_pruefen(ha) -> int:
|
||||
"""Liest die juengste Spur der Automation (trace/list, trace/get). 1 bei Befund."""
|
||||
liste = ha.befehl_ok("trace/list", domain="automation", item_id=AUTOMATION_ID)
|
||||
lauf = None
|
||||
laeufe = [eintrag for eintrag in liste or [] if isinstance(eintrag, dict)]
|
||||
if laeufe:
|
||||
juengster = max(
|
||||
laeufe, key=lambda eintrag: str((eintrag.get("timestamp") or {}).get("start", ""))
|
||||
)
|
||||
lauf = ha.befehl_ok(
|
||||
"trace/get", domain="automation", item_id=AUTOMATION_ID, run_id=juengster.get("run_id")
|
||||
)
|
||||
zeile, befund = spur_befund(laeufe, lauf)
|
||||
print(("BEFUND " if befund else "") + zeile)
|
||||
return 1 if befund else 0
|
||||
|
||||
|
||||
def pruefen(ha, datei) -> int:
|
||||
dienste = _dienste(ha)
|
||||
rest = dienste.get("rest_command", {})
|
||||
print(f"rest_command.{REST_DIENST}: {'vorhanden' if REST_DIENST in rest else 'fehlt'}")
|
||||
print(f"rest_command.reload: {'vorhanden' if 'reload' in rest else 'fehlt'}")
|
||||
app = sorted(name for name in dienste.get("notify", {}) if name.startswith("mobile_app_"))
|
||||
print("notify-Dienste der App: " + (", ".join(f"notify.{n}" for n in app) or "keiner"))
|
||||
for name in ERLAUBT:
|
||||
inhalt = datei.lies(name)
|
||||
text = "" if inhalt is None else inhalt.decode("utf-8")
|
||||
zeile = f"{name}: {'fehlt' if inhalt is None else 'vorhanden'}, Block des Totmanns "
|
||||
zeile += "da" if MARKE_ANFANG in text else "nicht da"
|
||||
if name == "configuration.yaml":
|
||||
fremd = FREMDES_REST_COMMAND.search(ohne_block(text))
|
||||
zeile += f", fremdes rest_command {'ja' if fremd else 'nein'}"
|
||||
print(zeile)
|
||||
zustaende = ha.befehl_ok("get_states") or []
|
||||
auto = [s for s in zustaende if s.get("attributes", {}).get("id") == AUTOMATION_ID]
|
||||
if auto:
|
||||
print(
|
||||
f"Automation: {auto[0]['entity_id']} = {auto[0]['state']}, "
|
||||
f"{auto[0]['attributes'].get('friendly_name')}"
|
||||
)
|
||||
else:
|
||||
print("Automation: fehlt")
|
||||
zeichen = [s for s in zustaende if s.get("entity_id") == LEBENSZEICHEN]
|
||||
stempel = zeichen[0].get("attributes", {}).get("timestamp") if zeichen else None
|
||||
if isinstance(stempel, (int, float)):
|
||||
print(f"Lebenszeichen: vor {time.time() - stempel:.0f} s")
|
||||
else:
|
||||
print("Lebenszeichen: nicht lesbar")
|
||||
live = eingaben_pruefen(ha, stempel)
|
||||
if auto:
|
||||
live += spur_pruefen(ha)
|
||||
befunde = bedingung_pruefen(ha)
|
||||
print(f"Rechenprobe der Bedingung: {len(faelle(fenster_je_tag(PROBE_KALENDER)))} Lagen, {befunde} Befund(e)")
|
||||
return 2 if befunde or live else 0
|
||||
|
||||
|
||||
def einrichten(ha, datei, trading: Path, app_dienst: str, zugang: dict) -> int:
|
||||
dienste = _dienste(ha)
|
||||
if app_dienst.split(".", 1)[1] not in dienste.get("notify", {}):
|
||||
print(f"ABBRUCH: {app_dienst} steht nicht in der Dienstliste von Home Assistant")
|
||||
return 3
|
||||
thema, zugriff = zugang["thema"], zugang["token"]
|
||||
paper = tomllib.loads((trading / "config" / "paper.toml").read_text(encoding="utf-8"))
|
||||
ntfy_url = str(paper["benachrichtigung"]["ntfy_url"]).rstrip("/")
|
||||
fenster = fenster_je_tag(zugang["kalender"])
|
||||
if not fenster:
|
||||
print("ABBRUCH: Der Kalender von Alpaca nennt keinen Handelstag")
|
||||
return 3
|
||||
print(f"Kalender: {len(fenster)} Handelstage, {min(fenster)} bis {max(fenster)}")
|
||||
|
||||
alt = {name: datei.lies(name) for name in ERLAUBT}
|
||||
if alt["configuration.yaml"] is None:
|
||||
print("ABBRUCH: configuration.yaml nicht lesbar")
|
||||
return 3
|
||||
konf_alt = alt["configuration.yaml"].decode("utf-8")
|
||||
geh_alt = (alt["secrets.yaml"] or b"").decode("utf-8")
|
||||
if FREMDES_REST_COMMAND.search(ohne_block(konf_alt)):
|
||||
print("ABBRUCH: configuration.yaml hat schon einen Abschnitt rest_command; von Hand zusammenfuehren")
|
||||
return 3
|
||||
neu = {
|
||||
"secrets.yaml": anhaengen(
|
||||
ohne_block(geh_alt), secrets_block(f"{ntfy_url}/{thema}", zugriff)
|
||||
),
|
||||
"configuration.yaml": anhaengen(ohne_block(konf_alt), config_block(bool(zugriff))),
|
||||
}
|
||||
vorher = {"secrets.yaml": geh_alt, "configuration.yaml": konf_alt}
|
||||
geschrieben = []
|
||||
for name in ERLAUBT:
|
||||
if neu[name] != vorher[name]:
|
||||
datei.schreibe_und_pruefe(name, neu[name].encode("utf-8"))
|
||||
geschrieben.append(name)
|
||||
status, antwort = ha.http_post("/api/config/core/check_config", {})
|
||||
ergebnis = antwort.get("result") if isinstance(antwort, dict) else None
|
||||
if status != 200 or ergebnis != "valid":
|
||||
fehler = antwort.get("errors") if isinstance(antwort, dict) else antwort
|
||||
print("BEFUND Konfiguration ungueltig: " + _ohne(str(fehler), thema, zugriff)[:300])
|
||||
for name in geschrieben:
|
||||
datei.schreibe_und_pruefe(name, alt[name] or b"")
|
||||
print(f"Zurueckgeschrieben: {geschrieben}")
|
||||
return 2
|
||||
print(f"Geschrieben: {geschrieben or 'nichts, die Bloecke standen schon da'}; Konfiguration gueltig")
|
||||
|
||||
rest = _dienste(ha).get("rest_command", {})
|
||||
if REST_DIENST not in rest and "reload" in rest:
|
||||
ha.befehl_ok("call_service", domain="rest_command", service="reload", service_data={})
|
||||
rest = _dienste(ha).get("rest_command", {})
|
||||
if REST_DIENST not in rest:
|
||||
print(
|
||||
f"rest_command.{REST_DIENST} ist noch nicht geladen: Home Assistant neu starten "
|
||||
"(Einstellungen, System, Neu starten), danach einrichten noch einmal -- die Bloecke "
|
||||
"stehen dann schon da."
|
||||
)
|
||||
return 4
|
||||
|
||||
status, antwort = ha.http_post(
|
||||
f"/api/config/automation/config/{AUTOMATION_ID}", automation(fenster, app_dienst)
|
||||
)
|
||||
print(f"Automation angelegt oder ersetzt: HTTP {status}")
|
||||
if status != 200:
|
||||
print("ABBRUCH: Die Automation wurde nicht angenommen: " + _ohne(str(antwort), thema, zugriff)[:200])
|
||||
return 2
|
||||
return 0
|
||||
|
||||
|
||||
def probe(ha, app_dienst: str) -> int:
|
||||
dienste = _dienste(ha)
|
||||
if REST_DIENST not in dienste.get("rest_command", {}):
|
||||
print(f"ABBRUCH: rest_command.{REST_DIENST} fehlt; erst einrichten")
|
||||
return 3
|
||||
name = app_dienst.split(".", 1)[1]
|
||||
if name not in dienste.get("notify", {}):
|
||||
print(f"ABBRUCH: {app_dienst} steht nicht in der Dienstliste von Home Assistant")
|
||||
return 3
|
||||
ha.befehl_ok(
|
||||
"call_service", domain="rest_command", service=REST_DIENST, service_data={"nachricht": PROBE_TEXT}
|
||||
)
|
||||
ha.befehl_ok(
|
||||
"call_service",
|
||||
domain="notify",
|
||||
service=name,
|
||||
service_data={"title": "Trading Totmann (Probe)", "message": PROBE_TEXT},
|
||||
)
|
||||
print("Zwei Probemeldungen abgeschickt (ntfy und HA-App). Ob sie ankamen, sieht nur Pascal.")
|
||||
return 0
|
||||
|
||||
|
||||
def entfernen(ha, datei) -> int:
|
||||
status = ha.http_delete(f"/api/config/automation/config/{AUTOMATION_ID}")
|
||||
print(f"Automation geloescht: HTTP {status}")
|
||||
for name in ERLAUBT:
|
||||
inhalt = datei.lies(name)
|
||||
if inhalt is None:
|
||||
print(f"{name}: fehlt")
|
||||
continue
|
||||
text = inhalt.decode("utf-8")
|
||||
if MARKE_ANFANG not in text:
|
||||
print(f"{name}: kein Block")
|
||||
continue
|
||||
datei.schreibe_und_pruefe(name, ohne_block(text).encode("utf-8"))
|
||||
print(f"{name}: Block entfernt")
|
||||
status, antwort = ha.http_post("/api/config/core/check_config", {})
|
||||
print("Konfiguration: " + str(antwort.get("result") if isinstance(antwort, dict) else f"HTTP {status}"))
|
||||
rest = _dienste(ha).get("rest_command", {})
|
||||
if "reload" in rest:
|
||||
ha.befehl_ok("call_service", domain="rest_command", service="reload", service_data={})
|
||||
print("rest_command neu geladen")
|
||||
else:
|
||||
print(
|
||||
f"rest_command.{REST_DIENST} bleibt bis zum naechsten Neustart von Home Assistant "
|
||||
"geladen; ohne Automation ruft ihn niemand."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
# Die Befehle, die pruefen an Home Assistant schickt. Alle lesen nur.
|
||||
LESEND = frozenset({"get_services", "get_states", "trace/list", "trace/get", "render_template"})
|
||||
|
||||
|
||||
class _NurLesend:
|
||||
"""Die Verbindung fuer pruefen: laesst nur LESEND und das Lesen von Dateien durch.
|
||||
|
||||
Seit der Pruefung Z2 (2026-09-25). Vorher lief pruefen mit nur_lesen=False, und ein
|
||||
Fehler in pruefen, der einen schreibenden Befehl schickte, waere nicht abgefangen
|
||||
worden. Ueber supervisor/api geht nur GET und das Anlegen der Ingress-Sitzung
|
||||
(ENDPUNKT_SITZUNG), das ConfigDatei auch lesend braucht. Jede andere Methode der
|
||||
Verbindung -- http_post, http_delete -- ist hier gesperrt.
|
||||
"""
|
||||
|
||||
def __init__(self, ha, sitzung_endpunkt: str) -> None:
|
||||
self._ha = ha
|
||||
self._sitzung = sitzung_endpunkt
|
||||
|
||||
def _pruefe(self, typ: str, felder: dict) -> None:
|
||||
if typ in LESEND:
|
||||
return
|
||||
if typ == "supervisor/api":
|
||||
methode = str(felder.get("method", "")).lower()
|
||||
if methode == "get":
|
||||
return
|
||||
if methode == "post" and felder.get("endpoint") == self._sitzung:
|
||||
return
|
||||
raise RuntimeError(f"pruefen ist lesend: {typ} {felder.get('method', '')} gesperrt".strip())
|
||||
|
||||
def befehl_ok(self, typ: str, **felder):
|
||||
self._pruefe(typ, felder)
|
||||
return self._ha.befehl_ok(typ, **felder)
|
||||
|
||||
def roh_sammeln(self, typ: str, dauer: float, **felder):
|
||||
self._pruefe(typ, felder)
|
||||
return self._ha.roh_sammeln(typ, dauer, **felder)
|
||||
|
||||
def __getattr__(self, name: str):
|
||||
raise RuntimeError(f"pruefen ist lesend: {name} gesperrt")
|
||||
|
||||
|
||||
def online(argumente) -> int:
|
||||
ha_projekt, trading = Path(argumente.ha_projekt), Path(argumente.trading)
|
||||
zugang = None
|
||||
if argumente.aktion == "einrichten":
|
||||
# Vor der Verbindung zu Home Assistant: Der WSL-Teil braucht bis zu WSL_TIMEOUT_S,
|
||||
# und so lange soll keine Verbindung offen warten.
|
||||
zugang = aus_wsl(argumente.tage)
|
||||
print(
|
||||
f"Aus WSL: Thema Laenge {len(zugang['thema'])}, Token Laenge "
|
||||
f"{len(zugang['token'])}, {len(zugang['kalender'])} Handelstage"
|
||||
)
|
||||
sys.path.insert(0, str(ha_projekt / "src"))
|
||||
import config_datei
|
||||
import ha_verbindung as hv
|
||||
|
||||
# Die Pfadregel des HA-Projekts erlaubt nur themes/. Dieses Werkzeug braucht genau
|
||||
# secrets.yaml und configuration.yaml und ersetzt die Regel dafuer zur Laufzeit; die
|
||||
# Methoden von ConfigDatei schlagen pruefe_pfad erst beim Aufruf nach.
|
||||
config_datei.pruefe_pfad = _nur_zwei_dateien
|
||||
token = hv.lies_token()
|
||||
print(f"Token gelesen (Laenge {len(token)}), Instanz aus dem HA-Projekt, Aktion {argumente.aktion}")
|
||||
# nur_lesen=False auch fuer pruefen: get_services, get_states, trace/list und trace/get
|
||||
# stehen nicht in der Leseliste des HA-Projekts (src/ha_verbindung.py, LESELISTE; dort
|
||||
# nur render_template). Fuer pruefen sperrt statt dessen _NurLesend alles ausser genau
|
||||
# diesen Befehlen (Pruefung Z2); die Dateien oeffnet es ueber ConfigDatei mit
|
||||
# nur_lesen=True.
|
||||
with hv.HaVerbindung(hv.BASIS_URL, token, nur_lesen=False) as verbindung:
|
||||
ha = (
|
||||
_NurLesend(verbindung, config_datei.ENDPUNKT_SITZUNG)
|
||||
if argumente.aktion == "pruefen"
|
||||
else verbindung
|
||||
)
|
||||
datei = config_datei.ConfigDatei(
|
||||
hv.BASIS_URL,
|
||||
lambda endpunkt, methode: ha.befehl_ok("supervisor/api", endpoint=endpunkt, method=methode),
|
||||
nur_lesen=argumente.aktion == "pruefen",
|
||||
)
|
||||
with datei:
|
||||
if argumente.aktion == "pruefen":
|
||||
return pruefen(ha, datei)
|
||||
if argumente.aktion == "einrichten":
|
||||
return einrichten(ha, datei, trading, argumente.app_dienst, zugang)
|
||||
if argumente.aktion == "probe":
|
||||
return probe(ha, argumente.app_dienst)
|
||||
return entfernen(ha, datei)
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
zerleger = argparse.ArgumentParser(
|
||||
description="Totmann der Trading Plattform in Home Assistant (Paket Z2).",
|
||||
allow_abbrev=False,
|
||||
)
|
||||
zerleger.add_argument("aktion", choices=["pruefen", "trocken", "einrichten", "probe", "entfernen"])
|
||||
zerleger.add_argument("ha_projekt", nargs="?")
|
||||
zerleger.add_argument("trading", nargs="?")
|
||||
zerleger.add_argument("--app-dienst", default=None)
|
||||
zerleger.add_argument("--tage", type=int, default=400)
|
||||
argumente = zerleger.parse_args(argv)
|
||||
if argumente.aktion == "trocken":
|
||||
return trocken()
|
||||
if not argumente.ha_projekt or not argumente.trading:
|
||||
zerleger.error("pruefen, einrichten, probe und entfernen brauchen beide Projektordner")
|
||||
if argumente.aktion in ("einrichten", "probe") and (
|
||||
argumente.app_dienst is None or not APP_DIENST_MUSTER.match(argumente.app_dienst)
|
||||
):
|
||||
zerleger.error("--app-dienst notify.<name> fehlt oder hat nicht diese Form")
|
||||
try:
|
||||
return online(argumente)
|
||||
except (Voraussetzung, wsl_teil.Befund) as fehler:
|
||||
# Diese Texte tragen keinen Wert aus der Geheimnisdatei, nur Feldnamen und Laengen.
|
||||
print(f"ABBRUCH: {fehler}")
|
||||
return 3
|
||||
except Exception as ausnahme:
|
||||
# Nur der Typ: Eine Ausnahme kann eine Adresse mit dem Thema oder einen Kopf mit
|
||||
# einem Schluessel tragen.
|
||||
print(f"ABBRUCH: {type(ausnahme).__name__}")
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user