evcc-bruecke/bruecke.py

"""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())