diff --git a/packages/modules/vehicles/porsche/README.md b/packages/modules/vehicles/porsche/README.md new file mode 100644 index 0000000000..4ade89a85b --- /dev/null +++ b/packages/modules/vehicles/porsche/README.md @@ -0,0 +1,50 @@ +# Porsche Connect – SoC-Modul für openWB + +Liest den Ladestand (SoC), die Reichweite und den Kilometerstand eines Porsche +mit **Porsche Connect** aus (z. B. Macan EV ab 2024, Taycan, Cayenne E3, +Panamera G2, 911 ab 992, 718). Die Werte stehen openWB u. a. fürs +PV-Überschussladen bis zu einem Ziel-SoC zur Verfügung. + +## Voraussetzungen + +- Aktives **Porsche-Connect-Abo** für das Fahrzeug. +- **My-Porsche-Zugangsdaten** (Porsche ID = E-Mail + Passwort). + +## Einrichtung in openWB + +1. Software-Integration → Fahrzeuge → beim Fahrzeug als **SoC-Modul** „Porsche + Connect" wählen. +2. Eintragen: + - **E-Mail** und **Passwort** der Porsche ID. + - **VIN** (optional). Leer lassen, wenn nur **ein** Fahrzeug im Konto ist; + bei mehreren Fahrzeugen ist die VIN Pflicht. +3. Speichern. Der SoC wird beim nächsten Abfragezyklus aktualisiert. + +## Wichtige Hinweise / Grenzen + +- **Inoffiziell:** Diese Schnittstelle ist nicht von Porsche freigegeben + (`official=False`). Porsche kann Login oder API jederzeit ändern. +- **Captcha:** Verlangt Auth0 ein Captcha, kann sich das Modul nicht anmelden. + Dann einmal in der My-Porsche-App/-Website (idealerweise aus demselben Netz) + anmelden und es danach erneut versuchen. +- **Schonender Abruf:** Es wird der zuletzt vom Fahrzeug übertragene Stand + (`get_stored_overview`-Äquivalent) gelesen; das Auto wird dafür **nicht + geweckt**. Der SoC kann daher etwas nachlaufen. +- **Token:** Access-/Refresh-Token werden unter + `data/modules/porsche/token_.json` zwischengespeichert, damit nicht + bei jedem Zyklus ein voller Login (mit Captcha-Risiko) nötig ist. + +## Herkunft + +Login-Flow (Auth0 „Identifier First") und Endpunkte sind portiert aus der +Community-Bibliothek [pyporscheconnectapi](https://github.com/CJNE/pyporscheconnectapi) +(Apache-2.0), hier als schlanke synchrone `requests`-Implementierung ohne +zusätzliche Abhängigkeiten. + +## Test + +Unit-Tests (gemocktes HTTP, keine echten Zugangsdaten nötig): + +```bash +PYTHONPATH=packages python -m pytest packages/modules/vehicles/porsche/porsche_test.py --noconftest +``` diff --git a/packages/modules/vehicles/porsche/__init__.py b/packages/modules/vehicles/porsche/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/packages/modules/vehicles/porsche/api.py b/packages/modules/vehicles/porsche/api.py new file mode 100644 index 0000000000..133951a9f9 --- /dev/null +++ b/packages/modules/vehicles/porsche/api.py @@ -0,0 +1,361 @@ +"""Synchroner Client fuer die (inoffizielle) Porsche-Connect-API. + +Der Login-Flow (Auth0 "Identifier First": E-Mail -> Passwort -> Resume -> Token) +und die Endpunkte sind portiert aus der Community-Bibliothek pyporscheconnectapi +(https://github.com/CJNE/pyporscheconnectapi, Apache-2.0). Dort ist alles async +(httpx); openWB ruft SoC-Module synchron auf, daher hier eine schlanke +requests-Implementierung ohne zusaetzliche Abhaengigkeiten. + +WICHTIG: Diese Schnittstelle ist nicht offiziell von Porsche unterstuetzt und kann +sich jederzeit aendern. Ein aktives Porsche-Connect-Abo ist Voraussetzung. +""" +import json +import logging +import time +from pathlib import Path +from typing import Dict, List, Optional, Tuple +from urllib.parse import parse_qs, urlparse + +import requests + +log = logging.getLogger(__name__) + +# --- Endpunkte / Client-Konstanten (aus pyporscheconnectapi/const.py) -------- +AUTHORIZATION_SERVER = "identity.porsche.com" +AUTHORIZATION_URL = f"https://{AUTHORIZATION_SERVER}/authorize" +TOKEN_URL = f"https://{AUTHORIZATION_SERVER}/oauth/token" +REDIRECT_URI = "my-porsche-app://auth0/callback" +AUDIENCE = "https://api.porsche.com" +CLIENT_ID = "XhygisuebbrqQ80byOuU5VncxLIm8E6H" +X_CLIENT_ID = "41843fb4-691d-4970-85c7-2673e8ecef40" +API_BASE_URL = "https://api.ppa.porsche.com/app" +USER_AGENT = "openWB-porsche/1.0" +TIMEOUT = 30 + +SCOPE = ("openid profile email offline_access mbb ssodb badge vin dealers cars " + "charging manageCharging plugAndCharge climatisation manageClimatisation " + "pid:user_profile.porscheid:read pid:user_profile.name:read " + "pid:user_profile.vehicles:read pid:user_profile.emails:read " + "pid:user_profile.locale:read") + +# Nur die Messgroessen, die openWB braucht (schlanke Antwort, weckt das Auto nicht). +MEASUREMENTS = ["BATTERY_LEVEL", "E_RANGE", "RANGE", "MILEAGE", "BATTERY_CHARGING_STATE"] + +# Token-Persistenz: repo/data/modules/porsche/ (parents[4] == Repo-Wurzel) +_DATA_PATH = Path(__file__).resolve().parents[4] / "data" / "modules" / "porsche" + + +class PorscheApiError(Exception): + """Allgemeiner API-Fehler.""" + + +class PorscheWrongCredentials(PorscheApiError): + """E-Mail/Passwort wurde von Porsche abgelehnt.""" + + +class PorscheCaptchaRequired(PorscheApiError): + """Porsche verlangt ein Captcha - kann headless nicht geloest werden. + + Abhilfe: einmal in der 'My Porsche'-App/Website vom selben Netz aus anmelden, + danach greift der Login meist wieder ohne Captcha. + """ + + +class PorscheConnectApi: + def __init__(self, email: str, password: str, vehicle_id: int) -> None: + if not email or not password: + raise PorscheApiError("E-Mail und Passwort muessen konfiguriert sein.") + self.email = email + self.password = password + self.vehicle_id = vehicle_id + self.session = requests.Session() + self.session.headers.update({"User-Agent": USER_AGENT, "X-Client-ID": X_CLIENT_ID}) + self._token: Dict = self._load_token() + + # --- Token-Persistenz ---------------------------------------------------- + @property + def _token_file(self) -> Path: + return _DATA_PATH / f"token_{self.vehicle_id}.json" + + def _load_token(self) -> Dict: + try: + with open(self._token_file, "r") as f: + return json.load(f) + except (FileNotFoundError, ValueError): + return {} + + def _save_token(self) -> None: + try: + _DATA_PATH.mkdir(parents=True, exist_ok=True) + with open(self._token_file, "w") as f: + json.dump(self._token, f) + except OSError: + log.exception("Porsche-Token konnte nicht gespeichert werden (nicht kritisch).") + + def _token_expired(self, leeway: int = 60) -> bool: + expires_at = self._token.get("expires_at") + if not expires_at: + return True + return (expires_at - leeway) < time.time() + + # --- OAuth2 (Auth0 Identifier-First) ------------------------------------- + def _location_params(self, resp: requests.Response) -> Dict[str, List[str]]: + if resp.status_code != 302 or "Location" not in resp.headers: + raise PorscheApiError( + f"Erwartete 302-Weiterleitung, erhielt {resp.status_code}.") + return parse_qs(urlparse(resp.headers["Location"]).query) + + def _fetch_authorization_code(self) -> str: + # 1. /authorize - liefert bei bestehender Session direkt den code, + # sonst eine Weiterleitung auf die Login-Seite mit state-Parameter. + resp = self.session.get( + AUTHORIZATION_URL, + params={ + "response_type": "code", + "client_id": CLIENT_ID, + "redirect_uri": REDIRECT_URI, + "audience": AUDIENCE, + "scope": SCOPE, + "state": "openwb", + }, + allow_redirects=False, + timeout=TIMEOUT, + ) + params = self._location_params(resp) + code = params.get("code", [None])[0] + if code is not None: + return code + + # 2. Keine Session -> Identifier-First-Login durchlaufen. + state = params.get("state", [None])[0] + if state is None: + raise PorscheApiError("Kein 'state' in der Auth0-Weiterleitung gefunden.") + resume_path = self._login_with_identifier(state) + + # 3. Auth-Code-Anfrage fortsetzen. + resp = self.session.get( + f"https://{AUTHORIZATION_SERVER}{resume_path}", + allow_redirects=False, + timeout=TIMEOUT, + ) + params = self._location_params(resp) + code = params.get("code", [None])[0] + if code is None: + raise PorscheApiError("Kein Authorization-Code nach dem Login erhalten.") + return code + + def _login_with_identifier(self, state: str) -> str: + # 2a. E-Mail + resp = self.session.post( + f"https://{AUTHORIZATION_SERVER}/u/login/identifier", + params={"state": state}, + data={ + "state": state, + "username": self.email, + "js-available": True, + "webauthn-available": False, + "is-brave": False, + "webauthn-platform-available": False, + "action": "default", + }, + allow_redirects=False, + timeout=TIMEOUT, + ) + if resp.status_code == 401: + raise PorscheWrongCredentials("E-Mail wurde abgelehnt.") + if resp.status_code == 400: + raise PorscheCaptchaRequired( + "Porsche verlangt ein Captcha. Bitte einmal in der My-Porsche-App " + "anmelden und es danach erneut versuchen.") + + # 2b. Passwort + resp = self.session.post( + f"https://{AUTHORIZATION_SERVER}/u/login/password", + params={"state": state}, + data={ + "state": state, + "username": self.email, + "password": self.password, + "action": "default", + }, + allow_redirects=False, + timeout=TIMEOUT, + ) + if resp.status_code == 400: + raise PorscheWrongCredentials("Passwort wurde abgelehnt.") + if "Location" not in resp.headers: + raise PorscheApiError( + f"Login-Schritt Passwort: unerwarteter Status {resp.status_code}.") + # Auth0 braucht einen kurzen Moment, bis der Resume-Pfad gueltig ist. + time.sleep(2.5) + return resp.headers["Location"] + + def _exchange_code(self, code: str) -> None: + resp = self.session.post( + TOKEN_URL, + data={ + "client_id": CLIENT_ID, + "grant_type": "authorization_code", + "code": code, + "redirect_uri": REDIRECT_URI, + }, + timeout=TIMEOUT, + ) + resp.raise_for_status() + self._store_token_response(resp.json()) + + def _refresh(self) -> bool: + refresh_token = self._token.get("refresh_token") + if not refresh_token: + return False + resp = self.session.post( + TOKEN_URL, + data={ + "client_id": CLIENT_ID, + "grant_type": "refresh_token", + "refresh_token": refresh_token, + }, + timeout=TIMEOUT, + ) + if resp.status_code == 403: + log.debug("Porsche-Refresh-Token ungueltig, voller Login noetig.") + return False + resp.raise_for_status() + self._store_token_response(resp.json()) + return True + + def _store_token_response(self, data: Dict) -> None: + data["expires_at"] = int(time.time()) + int(data.get("expires_in", 0)) + # refresh_token bleibt erhalten, falls die Antwort keinen neuen liefert + if not data.get("refresh_token") and self._token.get("refresh_token"): + data["refresh_token"] = self._token["refresh_token"] + self._token = data + self._save_token() + + def _ensure_token(self) -> str: + if not self._token_expired(): + return self._token["access_token"] + if self._token.get("refresh_token") and self._refresh(): + return self._token["access_token"] + code = self._fetch_authorization_code() + self._exchange_code(code) + return self._token["access_token"] + + # --- Daten-Endpunkte ----------------------------------------------------- + def _api_get(self, path: str) -> Dict: + resp = self.session.get( + f"{API_BASE_URL}{path}", + headers={"Authorization": f"Bearer {self._ensure_token()}"}, + timeout=TIMEOUT, + ) + resp.raise_for_status() + return resp.json() + + def list_vehicles(self) -> List[Dict]: + data = self._api_get("/connect/v1/vehicles") + return data if isinstance(data, list) else data.get("vehicles", []) + + def resolve_vin(self, configured_vin: Optional[str]) -> str: + if configured_vin: + return configured_vin.strip().upper() + vehicles = self.list_vehicles() + if not vehicles: + raise PorscheApiError("Keine Fahrzeuge im Porsche-Konto gefunden.") + if len(vehicles) > 1: + vins = ", ".join(v.get("vin", "?") for v in vehicles) + raise PorscheApiError( + f"Mehrere Fahrzeuge im Konto ({vins}). Bitte VIN in der Konfiguration angeben.") + return vehicles[0]["vin"] + + def fetch_soc(self, configured_vin: Optional[str]) -> Tuple[float, Optional[float], + Optional[float], Optional[float]]: + """Liest den gespeicherten Fahrzeugstatus (weckt das Auto nicht). + + Returns: (soc [%], range [km] | None, soc_timestamp [s] | None, odometer [km] | None) + """ + vin = self.resolve_vin(configured_vin) + query = "&".join(f"mf={m}" for m in MEASUREMENTS) + status = self._api_get(f"/connect/v1/vehicles/{vin}?{query}") + + measurements = {m["key"]: m for m in status.get("measurements", []) + if m.get("status", {}).get("isEnabled", True)} + + battery = measurements.get("BATTERY_LEVEL", {}).get("value", {}) + soc = battery.get("percent") + if soc is None: + raise PorscheApiError( + f"Kein BATTERY_LEVEL fuer VIN {vin} erhalten (Fahrzeug ohne Porsche Connect?).") + + range_km = None + range_meas = measurements.get("E_RANGE", measurements.get("RANGE", {})).get("value", {}) + if isinstance(range_meas, dict): + range_km = range_meas.get("kilometers") or range_meas.get("value") + + odometer = None + mileage = measurements.get("MILEAGE", {}).get("value", {}) + if isinstance(mileage, dict): + odometer = mileage.get("kilometers") or mileage.get("value") + + soc_ts = self._extract_timestamp(measurements.get("BATTERY_LEVEL", {})) + + return float(soc), range_km, soc_ts, odometer + + # --- Kommandos (optional, Komfort) --------------------------------------- + # HINWEIS: openWB regelt die Ladung ueber die Wallbox. Direct Charging steuert + # das Auto direkt und ist NICHT in openWBs automatische Lade-Logik eingebunden - + # es ist ein manuell ausloesbares Zusatzfeature (z. B. via cli.py). + def _api_post(self, path: str, payload: Dict) -> Dict: + resp = self.session.post( + f"{API_BASE_URL}{path}", + headers={"Authorization": f"Bearer {self._ensure_token()}"}, + json=payload, + timeout=TIMEOUT, + ) + resp.raise_for_status() + return resp.json() if resp.content else {} + + def send_command(self, configured_vin: Optional[str], key: str, + payload: Optional[Dict] = None, poll_timeout: int = 60) -> str: + """Sendet ein Fahrzeug-Kommando und wartet auf das Ergebnis. + + Returns den finalen Ergebnis-Code (z. B. "SUCCESS"). Wirft bei Fehler. + """ + vin = self.resolve_vin(configured_vin) + body = {"key": key, "payload": payload or {"spin": None}} + response = self._api_post(f"/connect/v1/vehicles/{vin}/commands", body) + status = response.get("status", {}) + status_id = status.get("id") + result = status.get("result") + if not (status_id and result == "ACCEPTED"): + return result or "UNKNOWN" + + deadline = time.time() + poll_timeout + while time.time() < deadline: + time.sleep(5) + poll = self._api_get(f"/connect/v1/vehicles/{vin}/commands/{status_id}") + result = poll.get("status", {}).get("result") + if result in ("ERROR", "FAIL", "FAILED"): + raise PorscheApiError(f"Kommando {key} fehlgeschlagen: {result}") + if result and result not in ("ACCEPTED", "IN_PROGRESS", "PERFORMING", "UNKNOWN"): + return result + raise PorscheApiError(f"Kommando {key}: kein Ergebnis innerhalb {poll_timeout}s.") + + def direct_charge(self, configured_vin: Optional[str], on: bool) -> str: + """Startet (on=True) oder stoppt (on=False) das sofortige Laden.""" + key = "DIRECT_CHARGING_START" if on else "DIRECT_CHARGING_STOP" + return self.send_command(configured_vin, key) + + @staticmethod + def _extract_timestamp(measurement: Dict) -> Optional[float]: + ts = (measurement.get("status", {}).get("updatedAt") + or measurement.get("value", {}).get("timestamp")) + if not ts: + return None + if isinstance(ts, (int, float)): + return float(ts) / 1000 if ts > 1e10 else float(ts) + # ISO-8601 String -> Unix-Sekunden + try: + from datetime import datetime + return datetime.fromisoformat(ts.replace("Z", "+00:00")).timestamp() + except (ValueError, AttributeError): + return None diff --git a/packages/modules/vehicles/porsche/cli.py b/packages/modules/vehicles/porsche/cli.py new file mode 100644 index 0000000000..29030013cd --- /dev/null +++ b/packages/modules/vehicles/porsche/cli.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +"""Kleines CLI zum Live-Test des Porsche-Connect-Moduls. + +Testet den echten Login und SoC-Abruf mit deinen My-Porsche-Zugangsdaten, +bevor das Modul auf der openWB konfiguriert wird. + +Beispiele: + python cli.py soc --email you@example.com --password 'geheim' + python cli.py soc --email you@example.com --password 'geheim' --vin WP0ZZZ... + python cli.py list --email you@example.com --password 'geheim' + python cli.py charge on --email ... --password ... --vin WP0ZZZ... + +Zugangsdaten koennen auch per Umgebungsvariablen PORSCHE_EMAIL / PORSCHE_PASSWORD +gesetzt werden (dann muessen --email/--password nicht angegeben werden). +""" +import argparse +import json +import logging +import os +import sys +from pathlib import Path + +# api.py als Nachbar-Modul importierbar machen (standalone lauffaehig) +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import api # noqa: E402 + + +def build_client(args) -> "api.PorscheConnectApi": + email = args.email or os.environ.get("PORSCHE_EMAIL") + password = args.password or os.environ.get("PORSCHE_PASSWORD") + if not email or not password: + sys.exit("Bitte --email/--password angeben (oder PORSCHE_EMAIL/PORSCHE_PASSWORD setzen).") + return api.PorscheConnectApi(email, password, vehicle_id=args.vehicle_id) + + +def main(argv=None): + parser = argparse.ArgumentParser(description="Porsche-Connect-Modul Live-Test") + parser.add_argument("--email") + parser.add_argument("--password") + parser.add_argument("--vin", default=None) + parser.add_argument("--vehicle-id", type=int, default=0, help="openWB-Fahrzeug-ID fuer den Token-Cache") + parser.add_argument("-v", "--verbose", action="store_true", help="Debug-Logging") + sub = parser.add_subparsers(dest="cmd", required=True) + sub.add_parser("soc", help="SoC/Reichweite/Kilometerstand abrufen") + sub.add_parser("list", help="Fahrzeuge im Konto auflisten") + charge = sub.add_parser("charge", help="Sofortladen starten/stoppen") + charge.add_argument("state", choices=["on", "off"]) + args = parser.parse_args(argv) + + logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, + format="%(levelname)s %(name)s: %(message)s") + + client = build_client(args) + try: + if args.cmd == "list": + print(json.dumps(client.list_vehicles(), indent=2, ensure_ascii=False)) + elif args.cmd == "soc": + soc, range_km, ts, odometer = client.fetch_soc(args.vin) + print(f"SoC: {soc} %") + print(f"Reichweite: {range_km} km") + print(f"Kilometerstand: {odometer} km") + print(f"Zeitstempel: {ts}") + elif args.cmd == "charge": + result = client.direct_charge(args.vin, args.state == "on") + print(f"Direct Charging {args.state}: {result}") + except api.PorscheCaptchaRequired as e: + sys.exit(f"CAPTCHA erforderlich: {e}") + except api.PorscheWrongCredentials as e: + sys.exit(f"Zugangsdaten falsch: {e}") + except api.PorscheApiError as e: + sys.exit(f"Fehler: {e}") + + +if __name__ == "__main__": + main() diff --git a/packages/modules/vehicles/porsche/config.py b/packages/modules/vehicles/porsche/config.py new file mode 100644 index 0000000000..15c7bb3fb4 --- /dev/null +++ b/packages/modules/vehicles/porsche/config.py @@ -0,0 +1,30 @@ +from typing import Optional + +from helpermodules.auto_str import auto_str + + +@auto_str +class PorscheConnectConfiguration: + def __init__(self, + email: Optional[str] = None, # My-Porsche E-Mail (Porsche ID) + password: Optional[str] = None, # My-Porsche Passwort + vin: Optional[str] = None, # optional; leer -> erstes Fahrzeug im Konto + calculate_soc: bool = False): + self.email = email + self.password = password + self.vin = vin + self.calculate_soc = calculate_soc + + +@auto_str +class PorscheConnect: + def __init__(self, + name: str = "Porsche Connect", + type: str = "porsche", + official: bool = False, + configuration: PorscheConnectConfiguration = None) -> None: + self.name = name + self.type = type + # official=False: inoffizielle, reverse-engineerte Porsche-Connect-Schnittstelle + self.official = official + self.configuration = configuration or PorscheConnectConfiguration() diff --git a/packages/modules/vehicles/porsche/porsche_test.py b/packages/modules/vehicles/porsche/porsche_test.py new file mode 100644 index 0000000000..bc93366783 --- /dev/null +++ b/packages/modules/vehicles/porsche/porsche_test.py @@ -0,0 +1,114 @@ +import time + +import pytest +import requests_mock + +from modules.vehicles.porsche import api +from modules.vehicles.porsche.api import PorscheApiError, PorscheConnectApi + + +def make_api(monkeypatch, tmp_path): + # Token-Persistenz in ein temporaeres Verzeichnis umlenken + monkeypatch.setattr(api, "_DATA_PATH", tmp_path) + client = PorscheConnectApi("mail@example.com", "secret", vehicle_id=0) + # gueltiges Token vorsetzen -> OAuth-Flow wird uebersprungen + client._token = {"access_token": "tok", "refresh_token": "ref", + "expires_at": int(time.time()) + 3600} + return client + + +def status_sample(): + return { + "vin": "WP0ZZZ12345678901", + "modelName": "Macan 4", + "measurements": [ + {"key": "BATTERY_LEVEL", "status": {"isEnabled": True, "updatedAt": "2026-08-26T10:00:00Z"}, + "value": {"percent": 72}}, + {"key": "E_RANGE", "status": {"isEnabled": True}, "value": {"kilometers": 310}}, + {"key": "MILEAGE", "status": {"isEnabled": True}, "value": {"kilometers": 12345}}, + ], + } + + +def test_fetch_soc_parses_values(monkeypatch, tmp_path): + client = make_api(monkeypatch, tmp_path) + vin = "WP0ZZZ12345678901" + with requests_mock.Mocker() as m: + m.get(f"{api.API_BASE_URL}/connect/v1/vehicles/{vin}", json=status_sample()) + soc, range_km, soc_ts, odometer = client.fetch_soc(vin) + assert soc == 72.0 + assert range_km == 310 + assert odometer == 12345 + assert soc_ts == pytest.approx(1787738400.0, abs=1) # 2026-08-26T10:00:00Z + + +def test_fetch_soc_missing_battery_raises(monkeypatch, tmp_path): + client = make_api(monkeypatch, tmp_path) + vin = "WP0ZZZ12345678901" + with requests_mock.Mocker() as m: + m.get(f"{api.API_BASE_URL}/connect/v1/vehicles/{vin}", + json={"vin": vin, "measurements": []}) + with pytest.raises(PorscheApiError): + client.fetch_soc(vin) + + +def test_resolve_vin_uses_configured(monkeypatch, tmp_path): + client = make_api(monkeypatch, tmp_path) + # kein HTTP noetig, wenn VIN konfiguriert ist + assert client.resolve_vin("wp0zzz12345678901") == "WP0ZZZ12345678901" + + +def test_resolve_vin_single_vehicle(monkeypatch, tmp_path): + client = make_api(monkeypatch, tmp_path) + with requests_mock.Mocker() as m: + m.get(f"{api.API_BASE_URL}/connect/v1/vehicles", + json=[{"vin": "WP0ZZZ12345678901"}]) + assert client.resolve_vin(None) == "WP0ZZZ12345678901" + + +def test_resolve_vin_multiple_requires_config(monkeypatch, tmp_path): + client = make_api(monkeypatch, tmp_path) + with requests_mock.Mocker() as m: + m.get(f"{api.API_BASE_URL}/connect/v1/vehicles", + json=[{"vin": "WP0AAA"}, {"vin": "WP0BBB"}]) + with pytest.raises(PorscheApiError): + client.resolve_vin(None) + + +def test_direct_charge_polls_until_done(monkeypatch, tmp_path): + monkeypatch.setattr(api.time, "sleep", lambda *_: None) # nicht wirklich warten + client = make_api(monkeypatch, tmp_path) + vin = "WP0ZZZ12345678901" + with requests_mock.Mocker() as m: + m.post(f"{api.API_BASE_URL}/connect/v1/vehicles/{vin}/commands", + json={"status": {"id": "cmd1", "result": "ACCEPTED"}}) + m.get(f"{api.API_BASE_URL}/connect/v1/vehicles/{vin}/commands/cmd1", + json={"status": {"result": "SUCCESS"}}) + result = client.direct_charge(vin, on=True) + assert result == "SUCCESS" + + +def test_direct_charge_error_raises(monkeypatch, tmp_path): + monkeypatch.setattr(api.time, "sleep", lambda *_: None) + client = make_api(monkeypatch, tmp_path) + vin = "WP0ZZZ12345678901" + with requests_mock.Mocker() as m: + m.post(f"{api.API_BASE_URL}/connect/v1/vehicles/{vin}/commands", + json={"status": {"id": "cmd1", "result": "ACCEPTED"}}) + m.get(f"{api.API_BASE_URL}/connect/v1/vehicles/{vin}/commands/cmd1", + json={"status": {"result": "ERROR"}}) + with pytest.raises(api.PorscheApiError): + client.direct_charge(vin, on=False) + + +def test_expired_token_triggers_refresh(monkeypatch, tmp_path): + client = make_api(monkeypatch, tmp_path) + client._token["expires_at"] = int(time.time()) - 10 # abgelaufen + vin = "WP0ZZZ12345678901" + with requests_mock.Mocker() as m: + m.post(api.TOKEN_URL, json={"access_token": "new", "refresh_token": "ref2", + "expires_in": 3600}) + m.get(f"{api.API_BASE_URL}/connect/v1/vehicles/{vin}", json=status_sample()) + soc, *_ = client.fetch_soc(vin) + assert soc == 72.0 + assert client._token["access_token"] == "new" diff --git a/packages/modules/vehicles/porsche/soc.py b/packages/modules/vehicles/porsche/soc.py new file mode 100644 index 0000000000..35e487d1d9 --- /dev/null +++ b/packages/modules/vehicles/porsche/soc.py @@ -0,0 +1,35 @@ +#!/usr/bin/env python3 +import logging + +from modules.common.abstract_device import DeviceDescriptor +from modules.common.abstract_vehicle import VehicleUpdateData +from modules.common.component_state import CarState +from modules.common.configurable_vehicle import ConfigurableVehicle +from modules.vehicles.porsche.api import PorscheConnectApi +from modules.vehicles.porsche.config import PorscheConnect + +log = logging.getLogger(__name__) + + +def create_vehicle(vehicle_config: PorscheConnect, vehicle: int): + api = None + + def initializer(): + nonlocal api + api = PorscheConnectApi(vehicle_config.configuration.email, + vehicle_config.configuration.password, + vehicle) + + def updater(vehicle_update_data: VehicleUpdateData) -> CarState: + soc, range, soc_ts, odometer = api.fetch_soc(vehicle_config.configuration.vin) + log.debug(f"Porsche SoC={soc}%, range={range}km, ts={soc_ts}, odo={odometer}km") + return CarState(soc=soc, range=range, soc_timestamp=soc_ts, odometer=odometer) + + return ConfigurableVehicle(vehicle_config=vehicle_config, + component_updater=updater, + vehicle=vehicle, + initializer=initializer, + calc_while_charging=vehicle_config.configuration.calculate_soc) + + +device_descriptor = DeviceDescriptor(configuration_factory=PorscheConnect)