feat: Brücke von evcc zum Energie-Cockpit

bruecke.py

@@ -0,0 +1,176 @@
+"""Brücke zwischen einer lokalen evcc-Installation und dem Energie-Cockpit.
+
+Liest in festem Takt die Ladevorgänge aus der REST-Schnittstelle von evcc
+(`GET /api/sessions`) und schickt sie an das Energie-Cockpit. Damit erscheinen
+die Ladungen dort mit ihrem Sonnenanteil, so wie es bei einer Fronius-Wallbox
+über deren Cloud-Freigabe der Fall ist.
+
+Die Brücke baut die Verbindung selbst nach außen auf. Am Router muss nichts
+geöffnet werden, und die evcc-Oberfläche bleibt im Heimnetz.
+
+Es wird nichts zwischengespeichert: Bei jedem Lauf geht der gesamte Rückblick
+erneut hinaus. Das Cockpit erkennt bereits bekannte Ladungen an ihrer Kennung
+und aktualisiert sie, statt sie doppelt anzulegen.
+"""
+
+import json
+import logging
+import os
+import sys
+import time
+from datetime import datetime, timedelta, timezone
+
+import requests
+
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s  %(levelname)-7s %(message)s',
+    datefmt='%Y-%m-%d %H:%M:%S',
+)
+logger = logging.getLogger('bruecke')
+
+
+class Konfigurationsfehler(Exception):
+    """Eine Angabe in der .env fehlt oder ist unbrauchbar."""
+
+
+def _pflicht(name):
+    wert = os.environ.get(name, '').strip()
+    if not wert:
+        raise Konfigurationsfehler(
+            f'{name} ist nicht gesetzt. Bitte in der Datei .env eintragen.'
+        )
+    return wert
+
+
+def _zahl(name, vorgabe):
+    roh = os.environ.get(name, '').strip()
+    if not roh:
+        return vorgabe
+    try:
+        return int(roh)
+    except ValueError:
+        raise Konfigurationsfehler(f'{name} muss eine ganze Zahl sein, gelesen wurde "{roh}".')
+
+
+def konfiguration_lesen():
+    """Sammelt alle Einstellungen aus der Umgebung und prüft sie."""
+    return {
+        'evcc_url': os.environ.get('EVCC_URL', 'http://evcc:7070').rstrip('/'),
+        'ziel_url': _pflicht('ZIEL_URL').rstrip('/'),
+        'token': _pflicht('TOKEN'),
+        'takt_minuten': _zahl('TAKT_MINUTEN', 60),
+        'rueckblick_tage': _zahl('RUECKBLICK_TAGE', 30),
+        'ladepunkt': os.environ.get('LADEPUNKT', '').strip(),
+    }
+
+
+def sessions_holen(evcc_url):
+    """Holt alle Ladevorgänge aus evcc. Wirft bei Fehlern eine Ausnahme."""
+    antwort = requests.get(f'{evcc_url}/api/sessions', timeout=30)
+    antwort.raise_for_status()
+    inhalt = antwort.json()
+
+    # evcc antwortet je nach Version entweder mit der blossen Liste
+    # oder mit einer Hülle {"result": [...]}.
+    if isinstance(inhalt, dict):
+        inhalt = inhalt.get('result', [])
+    if not isinstance(inhalt, list):
+        raise ValueError('Unerwartete Antwort von evcc, erwartet wurde eine Liste von Ladungen.')
+    return inhalt
+
+
+def _zeitpunkt(roh):
+    """Wandelt einen RFC3339-Zeitstempel von evcc in ein datetime um."""
+    if not roh:
+        return None
+    try:
+        return datetime.fromisoformat(str(roh).replace('Z', '+00:00'))
+    except ValueError:
+        return None
+
+
+def sessions_auswaehlen(sessions, rueckblick_tage, ladepunkt):
+    """Behält abgeschlossene Ladungen aus dem Rückblickfenster."""
+    grenze = datetime.now(timezone.utc) - timedelta(days=rueckblick_tage)
+    ausgewaehlt = []
+
+    for session in sessions:
+        ende = _zeitpunkt(session.get('finished'))
+        if ende is None:
+            continue  # läuft noch oder ohne Ende, kommt beim nächsten Lauf mit
+        if ende < grenze:
+            continue
+        if ladepunkt and session.get('loadpoint') != ladepunkt:
+            continue
+        ausgewaehlt.append(session)
+
+    return ausgewaehlt
+
+
+def senden(konfiguration, sessions):
+    """Schickt die Ladungen an das Energie-Cockpit."""
+    nutzlast = {
+        'quelle': 'evcc',
+        'gesendet': datetime.now(timezone.utc).isoformat(),
+        'sessions': sessions,
+    }
+    antwort = requests.post(
+        konfiguration['ziel_url'],
+        json=nutzlast,
+        headers={'X-API-Key': konfiguration['token']},
+        timeout=60,
+    )
+    antwort.raise_for_status()
+    try:
+        return antwort.json()
+    except json.JSONDecodeError:
+        return {}
+
+
+def durchlauf(konfiguration):
+    """Ein vollständiger Durchlauf: holen, auswählen, senden."""
+    sessions = sessions_holen(konfiguration['evcc_url'])
+    ausgewaehlt = sessions_auswaehlen(
+        sessions, konfiguration['rueckblick_tage'], konfiguration['ladepunkt']
+    )
+
+    if not ausgewaehlt:
+        logger.info('Keine abgeschlossenen Ladungen im Rückblick, nichts zu senden.')
+        return
+
+    ergebnis = senden(konfiguration, ausgewaehlt)
+    uebernommen = ergebnis.get('uebernommen')
+    if uebernommen is None:
+        logger.info('%d Ladungen gesendet.', len(ausgewaehlt))
+    else:
+        logger.info('%d Ladungen gesendet, %s davon übernommen.',
+                    len(ausgewaehlt), uebernommen)
+
+
+def main():
+    try:
+        konfiguration = konfiguration_lesen()
+    except Konfigurationsfehler as fehler:
+        logger.error('%s', fehler)
+        return 1
+
+    logger.info('Brücke gestartet. evcc: %s, Ziel: %s, Takt: %d Minuten.',
+                konfiguration['evcc_url'], konfiguration['ziel_url'],
+                konfiguration['takt_minuten'])
+
+    while True:
+        try:
+            durchlauf(konfiguration)
+        except requests.exceptions.RequestException as fehler:
+            logger.error('Verbindungsfehler, es wird im nächsten Takt erneut versucht: %s', fehler)
+        except ValueError as fehler:
+            logger.error('Antwort nicht verwertbar: %s', fehler)
+        except Exception:
+            logger.exception('Unerwarteter Fehler im Durchlauf.')
+
+        time.sleep(konfiguration['takt_minuten'] * 60)
+
+
+if __name__ == '__main__':
+    sys.exit(main())