claude-regelwerk/.vault-tools/trading-plattform-pruefung/ha_einrichten.py

"""Richtet in Home Assistant die Gegenstelle fuer die Kennzahlen der Trading Plattform ein.

Koordinator-Werkzeug, nicht Teil des Repos. Laeuft unter WINDOWS mit dem Python des HA-Projekts
(dort liegen `websockets` und `src/ha_verbindung.py`; Adresse der Instanz und der Name der
Token-Variable stehen in jenem Modul, nicht hier). Das HA-Token bleibt auf der Windows-Seite und
wird nie ausgegeben; auf den Trading-Rechner kommt nur die Webhook-Adresse.

Aktionen:
  pruefen     nur lesend: Version, vorhandene Helfer mit dem Praefix, vorhandene Automation
  lesen       nur lesend: wie pruefen, dazu die Zustaende der Helfer, wie sie gerade in Home Assistant stehen
  einrichten  legt acht Helfer und eine Automation mit Webhook-Ausloeser an (local_only, nur POST),
              erzeugt dafuer eine Webhook-ID (64 Hex-Zeichen) und traegt die Adresse in die
              config/secrets.toml des Trading-Projekts ein ([homeassistant], aktiv = false).
              Gibt die ID nie aus. Bricht ab, wenn es den Abschnitt oder die Automation schon gibt.
  probe       schickt eine Beispiel-Nutzlast an den Webhook (Adresse aus der Geheimnisdatei) und
              liest die Helfer ueber die HA-API zurueck
  entfernen   loescht Automation und Helfer wieder; der Abschnitt in der Geheimnisdatei bleibt stehen
              und wird genannt

Aufruf: <ha-projekt>\\.venv\\Scripts\\python.exe ha_einrichten.py <aktion> <ha-projektordner> <trading-projektordner>
"""

import json
import os
import secrets
import sys
import time
import tomllib
import urllib.request
from pathlib import Path

AUTOMATION_ID = "trading_plattform_kennzahlen"
ALIAS = "Trading Plattform: Kennzahlen uebernehmen"
PRAEFIX = "trading_"

TEXTE = ("Trading Phase", "Trading Modus", "Trading Killswitch")
ZAHLEN = (
    ("Trading Position", -100000, 100000, 1, "Stk"),
    ("Trading Tagesergebnis USD", -1000000, 1000000, 0.01, "USD"),
    ("Trading Herzschlag Alter s", -1, 100000000, 1, "s"),
)
SCHALTER = ("Trading Veraltet",)
ZEITEN = ("Trading Gesendet",)


def aktionen() -> list[dict]:
    def text(ziel: str, schluessel: str, vorgabe: str) -> dict:
        return {
            "service": "input_text.set_value",
            "target": {"entity_id": f"input_text.{ziel}"},
            "data": {"value": "{{ (trigger.json." + schluessel + " | default('" + vorgabe + "') | string)[:100] }}"},
            "continue_on_error": True,
        }

    def zahl(ziel: str, schluessel: str, filter_: str) -> dict:
        return {
            "service": "input_number.set_value",
            "target": {"entity_id": f"input_number.{ziel}"},
            "data": {"value": "{{ trigger.json." + schluessel + " | default(0) | " + filter_ + " }}"},
            "continue_on_error": True,
        }

    return [
        text("trading_phase", "phase", "unbekannt"),
        text("trading_modus", "modus", ""),
        text("trading_killswitch", "killswitch", "unbekannt"),
        zahl("trading_position", "position", "int(0)"),
        zahl("trading_tagesergebnis_usd", "tagesergebnis_usd", "float(0) | round(2)"),
        zahl("trading_herzschlag_alter_s", "herzschlag_alter_s", "int(-1)"),
        {
            "service": "input_boolean.turn_{{ 'on' if (trigger.json.veraltet | default(false)) else 'off' }}",
            "target": {"entity_id": "input_boolean.trading_veraltet"},
            "continue_on_error": True,
        },
        {
            "service": "input_datetime.set_datetime",
            "target": {"entity_id": "input_datetime.trading_gesendet"},
            "data": {"timestamp": "{{ now().timestamp() }}"},
            "continue_on_error": True,
        },
    ]


def lies_abschnitt(datei: Path) -> dict | None:
    return tomllib.loads(datei.read_text(encoding="utf-8")).get("homeassistant")


def trage_ein(datei: Path, adresse: str) -> None:
    roh = datei.read_text(encoding="utf-8")
    vorher = tomllib.loads(roh)
    block = (
        "\n[homeassistant]\n"
        "# Webhook in Home Assistant fuer die Kennzahlen des Paper-Runners. Die ID in der Adresse ist das\n"
        "# Geheimnis: Wer sie kennt, kann die Anzeige in HA faelschen. Eingeschaltet wird mit aktiv = true.\n"
        f'webhook_url = "{adresse}"\n'
        "aktiv = false\n"
    )
    neu = roh.rstrip("\n") + "\n" + block
    nachher = tomllib.loads(neu)
    if {k: v for k, v in nachher.items() if k != "homeassistant"} != vorher:
        raise SystemExit("ABBRUCH: bestehende Abschnitte haetten sich geaendert")
    nachbar = datei.with_name(datei.name + ".neu")
    with open(nachbar, "w", encoding="utf-8", newline="\n") as strom:
        strom.write(neu)
    os.replace(nachbar, datei)


def helfer_liste(ha) -> dict[str, list[dict]]:
    ergebnis = {}
    for art in ("input_text", "input_number", "input_boolean", "input_datetime"):
        eintraege = ha.befehl_ok(f"{art}/list") or []
        ergebnis[art] = [e for e in eintraege if str(e.get("id", "")).startswith(PRAEFIX)]
    return ergebnis


def main() -> int:
    aktion, ha_projekt, trading = sys.argv[1], Path(sys.argv[2]), Path(sys.argv[3])
    sys.path.insert(0, str(ha_projekt / "src"))
    import ha_verbindung as hv

    # Seit E2 (2026-09-23): der Aufloeser des Projekts entscheidet -- TRADING_GEHEIMNISSE, sonst
    # config/secrets.toml relativ zum uebergebenen Projektordner.
    sys.path.insert(0, str(trading))
    from trading import geheimnisse as aufloeser  # noqa: E402

    aufgeloest = aufloeser.pfad()
    geheimnisse = aufgeloest if aufgeloest.is_absolute() else trading / aufgeloest
    token = hv.lies_token()
    print(f"Token gelesen (Laenge {len(token)}), Instanz aus dem HA-Projekt, Aktion {aktion}")

    # nur_lesen=False auch fuers Pruefen: Die Leseliste jenes Moduls kennt die Listenbefehle der
    # Helfer und get_states nicht. 'pruefen' und 'probe' senden trotzdem nur lesende Befehle.
    with hv.HaVerbindung(hv.BASIS_URL, token, nur_lesen=False) as ha:
        vorhanden = helfer_liste(ha)
        zahl_helfer = sum(len(v) for v in vorhanden.values())
        automationen = [
            s for s in (ha.befehl_ok("get_states") or [])
            if s.get("entity_id", "").startswith("automation.") and s.get("attributes", {}).get("id") == AUTOMATION_ID
        ]
        abschnitt = lies_abschnitt(geheimnisse)
        print(f"Helfer mit Praefix {PRAEFIX}: {zahl_helfer} -- " + ", ".join(f"{a}: {[e['id'] for e in v]}" for a, v in vorhanden.items() if v))
        print(f"Automation {AUTOMATION_ID}: {'vorhanden' if automationen else 'fehlt'}")
        print("[homeassistant] in der Geheimnisdatei: " + ("fehlt" if abschnitt is None else str({k: (type(v).__name__, len(v) if isinstance(v, str) else v) for k, v in abschnitt.items()})))

        if aktion == "pruefen":
            return 0

        if aktion == "lesen":
            # Nur lesen: die Zustaende der Helfer, wie sie gerade in Home Assistant stehen.
            zustaende = {s["entity_id"]: s["state"] for s in (ha.befehl_ok("get_states") or []) if f".{PRAEFIX}" in s.get("entity_id", "")}
            for name in sorted(zustaende):
                print(f"  {name} = {zustaende[name]}")
            return 0

        if aktion == "einrichten":
            if abschnitt is not None or automationen or zahl_helfer:
                print("ABBRUCH: Es gibt schon Teile der Einrichtung; erst 'entfernen' oder von Hand klaeren.")
                return 3
            for name in TEXTE:
                ha.befehl_ok("input_text/create", name=name, min=0, max=100, mode="text", icon="mdi:chart-line")
            for name, kleinst, groesst, schritt, einheit in ZAHLEN:
                ha.befehl_ok("input_number/create", name=name, min=kleinst, max=groesst, step=schritt, mode="box", unit_of_measurement=einheit, icon="mdi:chart-line")
            for name in SCHALTER:
                ha.befehl_ok("input_boolean/create", name=name, icon="mdi:alert-outline")
            for name in ZEITEN:
                ha.befehl_ok("input_datetime/create", name=name, has_date=True, has_time=True, icon="mdi:clock-outline")
            webhook_id = secrets.token_hex(32)
            status, antwort = ha.http_post(
                f"/api/config/automation/config/{AUTOMATION_ID}",
                {
                    "alias": ALIAS,
                    "description": "Uebernimmt Phase, Killswitch, Position und Tagesergebnis des Paper-Runners aus dem Webhook der Trading Plattform. Angelegt am 2026-09-20 ueber ha_einrichten.py; Rueckbau mit der Aktion entfernen.",
                    "trigger": [{"platform": "webhook", "webhook_id": webhook_id, "allowed_methods": ["POST"], "local_only": True}],
                    "condition": [],
                    "action": aktionen(),
                    "mode": "queued",
                    "max": 5,
                },
            )
            print(f"Automation angelegt: HTTP {status}")
            if status != 200:
                print("ABBRUCH: Die Automation wurde nicht angenommen:", str(antwort)[:200].replace(webhook_id, "***"))
                return 2
            trage_ein(geheimnisse, f"{hv.BASIS_URL}/api/webhook/{webhook_id}")
            print("Adresse eingetragen ([homeassistant], aktiv = false); die ID wurde nicht ausgegeben.")
            return 0

        if aktion == "probe":
            if abschnitt is None:
                print("ABBRUCH: kein [homeassistant] in der Geheimnisdatei")
                return 3
            nutzlast = {
                "phase": "probe_der_einrichtung", "modus": "probe", "killswitch": "angehalten", "position": 7,
                "tagesergebnis_usd": -12.34, "herzschlag_alter_s": 5, "veraltet": False,
                "gesendet": time.strftime("%Y-%m-%dT%H:%M:%S"),
            }
            anfrage = urllib.request.Request(
                abschnitt["webhook_url"], data=json.dumps(nutzlast).encode("utf-8"),
                headers={"Content-Type": "application/json"}, method="POST",
            )
            try:
                with urllib.request.urlopen(anfrage, timeout=5) as antwort:
                    print(f"Webhook: HTTP {antwort.status}")
            except Exception as ausnahme:
                # Nur der Typ: Die Ausnahme traegt die Adresse und damit die ID.
                print(f"Webhook GESCHEITERT: {type(ausnahme).__name__}")
                return 2
            time.sleep(2.0)
            zustaende = {s["entity_id"]: s["state"] for s in (ha.befehl_ok("get_states") or []) if f".{PRAEFIX}" in s.get("entity_id", "")}
            for name in sorted(zustaende):
                print(f"  {name} = {zustaende[name]}")
            erwartet = {
                "input_text.trading_phase": "probe_der_einrichtung", "input_text.trading_killswitch": "angehalten",
                "input_number.trading_position": "7.0", "input_number.trading_tagesergebnis_usd": "-12.34",
                "input_boolean.trading_veraltet": "off",
            }
            falsch = {k: (v, zustaende.get(k)) for k, v in erwartet.items() if zustaende.get(k) != v}
            print("Rueckgelesen wie gesendet." if not falsch else f"BEFUND: {falsch}")
            return 0 if not falsch else 1

        if aktion == "entfernen":
            status = ha.http_delete(f"/api/config/automation/config/{AUTOMATION_ID}")
            print(f"Automation geloescht: HTTP {status}")
            for art, eintraege in vorhanden.items():
                for eintrag in eintraege:
                    ha.befehl_ok(f"{art}/delete", **{f"{art}_id": eintrag["id"]})
                    print(f"  geloescht: {art}.{eintrag['id']}")
            if abschnitt is not None:
                print("Hinweis: [homeassistant] steht weiter in der Geheimnisdatei des Trading-Projekts; von Hand entfernen.")
            return 0

    print("unbekannte Aktion")
    return 2


if __name__ == "__main__":
    sys.exit(main())