diff --git a/packages/control/ev/ev.py b/packages/control/ev/ev.py index 7e24057424..418adbeab9 100644 --- a/packages/control/ev/ev.py +++ b/packages/control/ev/ev.py @@ -112,6 +112,9 @@ def soc_interval_expired(self, vehicle_update_data: VehicleUpdateData) -> bool: interval = self.soc_module.general_config.request_interval_charging else: interval = self.soc_module.general_config.request_interval_not_charging + override = self.soc_module.get_request_interval_override() + if override is not None: + interval = min(interval, override) # Zeitstempel prüfen, ob wieder abgefragt werden muss. if timecheck.check_timestamp(self.data.get.soc_request_timestamp, interval-5) is False: # Zeit ist abgelaufen diff --git a/packages/modules/common/configurable_vehicle.py b/packages/modules/common/configurable_vehicle.py index 773b97889c..b1a347a12e 100644 --- a/packages/modules/common/configurable_vehicle.py +++ b/packages/modules/common/configurable_vehicle.py @@ -1,5 +1,6 @@ from enum import Enum import logging +import math import time from typing import Optional, TypeVar, Generic, Callable from helpermodules import timecheck @@ -37,8 +38,10 @@ def __init__(self, calc_while_charging: bool = False, general_config: Optional[GeneralVehicleConfig] = None, calculated_soc_state: Optional[CalculatedSocState] = None, - initializer: Callable = lambda: None) -> None: + initializer: Callable = lambda: None, + request_interval_override: Optional[Callable[[], Optional[float]]] = None) -> None: self.__component_updater = component_updater + self.__request_interval_override = request_interval_override self.vehicle_config = vehicle_config self.calculated_soc_state = calculated_soc_state self.__initializer = initializer @@ -97,7 +100,23 @@ def update(self, vehicle_update_data: VehicleUpdateData): Pub().pub(f"openWB/set/vehicle/{self.vehicle}/soc_module/calculated_soc_state", asdict(self.calculated_soc_state)) + def get_request_interval_override(self) -> Optional[float]: + """Return a module's current maximum request interval in seconds. + + None keeps the regular schedule. A positive interval can only advance + requests and also selects the API instead of calculated/chargepoint SoC. + The callback must be a quick, read-only lookup of module-owned state. + Keep it and the updater in the same closure so their cache/error state + remains shared when the SoC scheduler deep-copies this vehicle object. + """ + interval = self.__request_interval_override() if self.__request_interval_override is not None else None + if interval is not None and (not math.isfinite(interval) or interval <= 0): + raise ValueError("Vehicle request interval override must be finite and greater than zero") + return interval + def _get_carstate_source(self, vehicle_update_data: VehicleUpdateData) -> SocSource: + if self.get_request_interval_override() is not None: + return SocSource.API # Kein SoC vom LP vorhanden oder erwünscht if (vehicle_update_data.soc_from_cp is None or self.general_config.use_soc_from_cp is False or # oder aktueller manueller SoC vorhanden (ausgelesenen SoC während der Ladung korrigieren) diff --git a/packages/modules/common/request_interval_override_test.py b/packages/modules/common/request_interval_override_test.py new file mode 100644 index 0000000000..553646cbaa --- /dev/null +++ b/packages/modules/common/request_interval_override_test.py @@ -0,0 +1,136 @@ +import copy +from unittest.mock import Mock + +import pytest + +from control.ev.ev import Ev +from helpermodules import timecheck +from modules.common import store +from modules.common.abstract_vehicle import GeneralVehicleConfig, VehicleUpdateData +from modules.common.component_state import CarState +from modules.common.configurable_vehicle import ConfigurableVehicle, SocSource +from modules.vehicles.mqtt.config import MqttSocSetup + +NOW = 1652683252 # packages/conftest.py fixes the clock to this timestamp. + + +def vehicle(updater=None, override=None, **kwargs): + return ConfigurableVehicle(MqttSocSetup(), updater or Mock(return_value=CarState(soc=50)), vehicle=0, + request_interval_override=override, **kwargs) + + +@pytest.mark.parametrize("interval", [0, -1, float("inf"), float("nan")]) +def test_invalid_override_is_rejected(interval): + with pytest.raises(ValueError, match="finite and greater than zero"): + vehicle(override=lambda: interval).get_request_interval_override() + + +def test_default_has_no_override(): + assert vehicle().get_request_interval_override() is None + + +@pytest.mark.parametrize("counter, fault_state, charging, override, expected_interval", [ + (0, 0, False, None, 600), + (0, 0, True, None, 300), + (0, 2, False, None, 300), + (3, 2, False, None, 600), + (10, 1, False, 60, 60), # Cache warning still gets one-minute retries. + (10, 2, False, 60, 60), # Independent of the three-error limit. + (0, 0, True, 60, 60), + (0, 0, False, 1200, 600), # Cannot postpone a regular request. + (0, 0, True, 1200, 300), +]) +def test_scheduler_applies_override_after_regular_interval_selection( + counter, fault_state, charging, override, expected_interval, monkeypatch): + ev = Ev(0) + ev.soc_module = vehicle(override=lambda: override, general_config=GeneralVehicleConfig( + request_interval_charging=300, request_interval_not_charging=600)) + ev.data.get.soc_request_timestamp = NOW - 30 + ev.data.get.fault_state = fault_state + ev.data.set.soc_error_counter = counter + check = Mock(return_value=True) + monkeypatch.setattr(timecheck, "check_timestamp", check) + + assert not ev.soc_interval_expired(VehicleUpdateData(charge_state=charging)) + check.assert_called_once_with(NOW - 30, expected_interval - 5) + + +@pytest.mark.parametrize("elapsed, expected", [(30, False), (60, True)]) +def test_retry_uses_last_real_request_time(elapsed, expected): + ev = Ev(0) + ev.soc_module = vehicle(override=lambda: 60) + ev.data.get.soc_request_timestamp = NOW - elapsed + ev.data.get.fault_state = 1 + ev.data.set.soc_error_counter = 10 + assert ev.soc_interval_expired(VehicleUpdateData()) is expected + assert ev.data.get.soc_request_timestamp == NOW - elapsed + + +def test_only_plugged_policy_is_preserved(): + ev = Ev(0) + ev.soc_module = vehicle(override=lambda: 60, general_config=GeneralVehicleConfig(request_only_plugged=True)) + ev.data.get.soc_request_timestamp = NOW - 120 + assert not ev.soc_interval_expired(VehicleUpdateData(plug_state=False)) + assert ev.soc_interval_expired(VehicleUpdateData(plug_state=True)) + + +@pytest.mark.parametrize("override, cp_soc, expected", [ + (60, None, SocSource.API), + (60, 50, SocSource.API), + (None, None, SocSource.CALCULATION), + (None, 50, SocSource.CP), +]) +def test_retry_uses_api_even_when_soc_would_normally_be_calculated(override, cp_soc, expected): + module = vehicle(override=lambda: override, calc_while_charging=True, + general_config=GeneralVehicleConfig(use_soc_from_cp=True)) + update = VehicleUpdateData(charge_state=True, soc_from_cp=cp_soc, timestamp_soc_from_cp=NOW) + assert module._get_carstate_source(update) == expected + + +@pytest.mark.parametrize("has_cache", [False, True]) +def test_module_error_state_and_cache_survive_scheduler_copies(has_cache, monkeypatch): + last_request_failed = False + cached_state = (CarState(soc=50, soc_timestamp=NOW - 300, warning="Portal temporarily unavailable") + if has_cache else None) + portal = Mock(side_effect=[RuntimeError("Portal unavailable"), CarState(soc=60, soc_timestamp=NOW)]) + + def fetch(update): + nonlocal last_request_failed, cached_state + try: + cached_state = portal() + except RuntimeError: + last_request_failed = True + if cached_state is None: + raise + else: + last_request_failed = False + return cached_state + + value_store = Mock() + monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=value_store)) + module = vehicle(updater=fetch, override=lambda: 60 if last_request_failed else None, calc_while_charging=True) + copied = copy.deepcopy(module) + if has_cache: + state = copied._get_carstate_by_source(VehicleUpdateData(), SocSource.API) + assert state.soc == 50 + assert state.soc_timestamp == NOW - 300 + assert state.warning == "Portal temporarily unavailable" + else: + with pytest.raises(Exception, match="Portal unavailable"): + copied._get_carstate_by_source(VehicleUpdateData(), SocSource.API) + assert module.get_request_interval_override() == 60 + + # The next scheduler copy observes the failure, calls the API during charging, + # and its success restores the normal schedule/source on the original object. + next_copy = copy.deepcopy(module) + next_copy.update(VehicleUpdateData(charge_state=True)) + assert portal.call_count == 2 + assert module.get_request_interval_override() is None + assert module._get_carstate_source(VehicleUpdateData(charge_state=True)) == SocSource.CALCULATION + + +def test_new_module_instance_does_not_inherit_old_override(): + old_module = vehicle(override=lambda: 60) + replacement = vehicle() + assert old_module.get_request_interval_override() == 60 + assert replacement.get_request_interval_override() is None diff --git a/packages/modules/vehicles/vweuda/README.txt b/packages/modules/vehicles/vweuda/README.txt deleted file mode 100644 index 2bef5e60d6..0000000000 --- a/packages/modules/vehicles/vweuda/README.txt +++ /dev/null @@ -1,41 +0,0 @@ - -Das ist ein Startpunkt und wird noch Verbesserungen und Korrekturen benötigen. -Dazu werden Rückmeldungen aus der Community benötigt. - -Dazu wird im SoC-Log einiges an Ausgaben auf Level Info ausgegeben. -In einigen Fällen könnte es auch nötig sein, Level Debug/Details einzustellen. - -Die Daten im eu-data-act Portal sind bei meinem Fahrzeug mittlerweile einigermaßen aktuell. -Wenn das Fahrzeug gefahren oder geladen wurde ca. 2 Stunden. - -Vor Nutzung muss die kontinuierliche Abfrage (alle Daten, 15-min) im Portal angestoßen werden. -https://eu-data-act.drivesomethinggreater.com/de/de/eu-data-act.html -Benutzerdefinierte Daten abrufen - All Data - 15 min -Es kann dann mehrere Stunden dauern, bis erste nicht-leere Datenpakete ankommen. - -Das Modul startet je konfiguriertem Fahrzeug einen parallelen Prozess um das Portal zu pollen. -Der Thread prüft im Minutentakt auf Bereitschaft des Portals und lädt dann die neueste zip-Datei herunter. -Die 5 letzten in den zip's enthaltenen json-Dateien werden in ramdisk/vweuda gespeichert zu evtl. Analyse. - -Aus der json wird versucht, folgende Daten zu extrahieren: soc, range, soc_timestamp, odometer. -soc scheint immer vorhanden zu sein. -Range habe ich noch nie bekommen. -Als soc_timestamp nehme ich das maximum aller car_captured_time Felder. -odometer kommt mal und mal nicht. - -Diese Daten werden je VIN für die Abfrage durch die openwb bereitgestellt. - -Der jeweils letzte Stand, der an die openwb gemeldet wird, wird auch persistent gehalten. -Dieser Stand wird bei jeder Abfrage der openwb mit den letzten aus den Portal gelieferten Daten ergänzt. -Damit wird ein Feld nicht ungültig, wenn es vom Portal nicht gemeldet wurde. -Wenn die Reichweite/range leer ist, wird diese aus Batterie-Kapazität, SoC und Durchschnittsverbrauch berechnet. - -Solange keine Daten vom Portal kommen werden entsprechende Meldungen im soc-log ausgegeben. - -Die Konfiguration des Moduls ist gleichgeblieben. - -Nach dem ersten Start ist es normal, daß erst mal keine Daten vorhanden sind. -Auch wenn das Portal schon nicht-leere Datenpakete liefert, kann es mehrere Abfrage-Intervalle dauern, bis erste Ergebnisse in openWB ankommen. - -Quellen: - https://github.com/mikrohard/hass-vw-eu-data-act diff --git a/packages/modules/vehicles/vweuda/api.py b/packages/modules/vehicles/vweuda/api.py new file mode 100644 index 0000000000..52c8b99bea --- /dev/null +++ b/packages/modules/vehicles/vweuda/api.py @@ -0,0 +1,238 @@ +"""Synchronous VW EU Data Act login and dataset downloads.""" + +import io +import json +import re +import zipfile +from html.parser import HTMLParser +from typing import Optional, Union +from urllib.parse import urlencode, urljoin, urlparse + +from requests import HTTPError, RequestException, Response, Session + +BASE_URL = "https://eu-data-act.drivesomethinggreater.com" +OIDC_AUTHORIZE_URL = "https://identity.vwgroup.io/oidc/v1/authorize" +CALLBACK_LOGIN_PATH = "/services/callbacklogin" +METADATA_PATH = "/proxy_api/euda-apim/datarequest/vehicles/{vin}/metadata/partial" +LIST_PATH = "/proxy_api/euda-apim/datadelivery/vehicles/{vin}/{identifier}/list" +DOWNLOAD_PATH = "/proxy_api/euda-apim/datadelivery/vehicles/{vin}/{identifier}/download" +NO_CONTENT_SUFFIX = "_no_content_found.zip" +HTTP_TIMEOUT = (5, 15) +USER_AGENT = "vw_eu_data_act/0.3.3" + +# OIDC client ID and portal brand state. SEAT and CUPRA share a client ID. +BRANDS = { + "volkswagen": ("9b58543e-1c15-4193-91d5-8a14145bebb0@apps_vw-dilab_com", "VOLKSWAGEN_PASSENGER_CARS"), + "audi": ("cc29b87a-5e9a-4362-aecf-5adea6b01bbb@apps_vw-dilab_com", "AUDI"), + "skoda": ("3ea88bf9-1d4e-4a68-b3ad-4098c1f1d246@apps_vw-dilab_com", "SKODA"), + "seat": ("f85e5b69-e3b2-43aa-9c0d-1b7d0e0b576f@apps_vw-dilab_com", "SEAT"), + "cupra": ("f85e5b69-e3b2-43aa-9c0d-1b7d0e0b576f@apps_vw-dilab_com", "CUPRA"), +} +BRAND_PREFIXES = { + "audi": ("WAU", "WA1", "WUA", "WU1", "99A", "AAA", "TRU"), + "skoda": ("NAD", "TMB", "Y6U"), + "cupra": ("VSS",), # SEAT/CUPRA cannot be distinguished by this prefix. +} + + +def brand_for_vin(vin: str) -> str: + return next((brand for brand, prefixes in BRAND_PREFIXES.items() if vin[:3] in prefixes), "volkswagen") + + +class ApiError(Exception): + """Portal/network failure, with HTTP status when available.""" + + def __init__(self, message: str, *, status: Optional[int] = None) -> None: + super().__init__(message) + self.status = status + + +class AuthError(ApiError): + """Login failed or the identity provider's form changed.""" + + +class DatasetError(ApiError): + """Malformed archive; it must not prevent processing newer datasets.""" + + +class _LoginForm(HTMLParser): + """Collect the first HTML form and its input fields.""" + + def __init__(self, html: str) -> None: + super().__init__() + self.action = None + self.fields = {} + self._inside = False + self._done = False + self.feed(html) + + def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None: + attrs = dict(attrs) + if tag == "form" and not self._done and not self._inside: + self.action = attrs.get("action") + self._inside = True + elif tag == "input" and self._inside and attrs.get("name"): + self.fields[attrs["name"]] = attrs.get("value") or "" + + def handle_endtag(self, tag: str) -> None: + if tag == "form" and self._inside: + self._inside, self._done = False, True + + +def _template_model(html: str) -> dict: + # raw_decode also handles braces inside JSON strings, unlike brace counting. + index = html.find("templateModel") + start = html.find("{", index) if index >= 0 else -1 + try: + return json.JSONDecoder().raw_decode(html[start:])[0] if start >= 0 else {} + except ValueError: + return {} + + +def _login_error(html: str) -> Optional[str]: + model = _template_model(html) + error = model.get("error") or model.get("errorCode") + if isinstance(error, dict): + return error.get("text") or error.get("errorCode") or str(error) + return str(error) if error else None + + +def _login_fields(html: str) -> tuple[dict, Optional[str]]: + form = _LoginForm(html) + model = _template_model(html) + for key in ("hmac", "relayState"): + if model.get(key): + form.fields[key] = model[key] + csrf = re.search(r"csrf_token\s*[:=]\s*['\"]([^'\"]+)['\"]", html) + if csrf: + form.fields.setdefault("_csrf", csrf.group(1)) + if not all(key in form.fields for key in ("hmac", "_csrf")): + raise AuthError(_login_error(html) or "Could not parse identity login form") + return form.fields, form.action + + +def _check_login_status(response: Response) -> None: + status = response.status_code + if status >= 500 or status in (404, 429): + raise ApiError(f"Identity provider error (HTTP {status})", status=status) + if status >= 400: + raise AuthError(_login_error(response.text) or f"Login rejected (HTTP {status})") + + +class EudaApiClient: + """The synchronous caller owns and closes the session.""" + + def __init__(self, session: Session, email: str, password: str, brand: str = "volkswagen") -> None: + self._session = session + self._email, self._password, self._brand = email, password, brand + self._logged_in = False + + def _request(self, method: str, url: str, **kwargs) -> Response: + headers = {"User-Agent": USER_AGENT, **kwargs.pop("headers", {})} + try: + return self._session.request(method, url, headers=headers, timeout=HTTP_TIMEOUT, **kwargs) + except HTTPError as error: + # req.get_http_session() raises in its response hook. Keep the full + # response for VW login status handling and the 401/403 retry below. + if error.response is None: + raise ApiError(f"HTTP error during {method}: {error}") from error + return error.response + except RequestException as error: + raise ApiError(f"Network error during {method}: {error}") from error + + @staticmethod + def _build_authorize_url(brand: str = "volkswagen") -> str: + client_id, state = BRANDS.get(brand, BRANDS["volkswagen"]) + # The portal's redirect/authentication servlet fails for non-browser clients. + return OIDC_AUTHORIZE_URL + "?" + urlencode({ + "client_id": client_id, "state": f"de__en__{state}", "response_type": "code", + "scope": "openid cars profile", "redirect_uri": BASE_URL + "/login", "prompt": "login", + }) + + def _post_login_form(self, url: str, *, data: dict, referer: str) -> Response: + """Follow at most ten redirects, stopping at the session-setting callback.""" + response = self._request("POST", url, data=data, headers={"Referer": referer}, allow_redirects=False) + for hop in range(11): + location = response.headers.get("Location") + parsed = urlparse(response.url) + callback = parsed.netloc == urlparse(BASE_URL).netloc and parsed.path.startswith(CALLBACK_LOGIN_PATH) + if callback or response.status_code not in (301, 302, 303) or not location: + return response + previous_url = response.url + next_url = urljoin(previous_url, location) + response.close() + if hop == 10: + raise ApiError("Login redirect chain exceeded 10 hops") + response = self._request("GET", next_url, headers={"Referer": previous_url}, allow_redirects=False) + + def login(self) -> None: + """Authorize → email form → password form → portal callback.""" + self._logged_in = False + with self._request("GET", self._build_authorize_url(self._brand)) as response: + _check_login_status(response) + signin_url = response.url + fields, action = _login_fields(response.text) + + fields["email"] = self._email + with self._request("POST", urljoin(signin_url, action or ""), data=fields, + headers={"Referer": signin_url}) as response: + _check_login_status(response) + authenticate_url = response.url + fields, action = _login_fields(response.text) + + fields.update(email=self._email, password=self._password) + # relayState belongs in the POST body, not in the action URL as well. + action_url = urljoin(authenticate_url, action) if action else authenticate_url.split("?", 1)[0] + with self._post_login_form(action_url, data=fields, referer=authenticate_url) as response: + _check_login_status(response) + parsed = urlparse(response.url) + if parsed.netloc != urlparse(BASE_URL).netloc or not parsed.path.startswith(CALLBACK_LOGIN_PATH): + raise AuthError(_login_error(response.text) or "Login did not reach the portal callback") + if not response.cookies.get("access_token"): + raise AuthError("Login callback did not establish a session (access_token cookie missing)") + self._logged_in = True + + def _authenticated_get(self, url: str, headers: Optional[dict] = None) -> Response: + """Share the same one-time reauthentication for JSON and ZIP requests.""" + if not self._logged_in: + self.login() + for attempt in range(2): + response = self._request("GET", url, headers=headers or {}) + status = response.status_code + if status in (401, 403) and attempt == 0: + response.close() + self.login() + continue + if status >= 400: + response.close() + raise ApiError(f"Portal request failed (HTTP {status})", status=status) + return response + + def _get_json(self, url: str, headers: Optional[dict] = None) -> Union[dict, list]: + with self._authenticated_get(url, headers) as response: + try: + return response.json() + except ValueError as error: + raise ApiError("Invalid JSON from portal") from error + + def get_metadata(self, vin: str) -> dict: + return self._get_json(BASE_URL + METADATA_PATH.format(vin=vin)) + + def list_datasets(self, vin: str, identifier: str) -> list[dict]: + url = BASE_URL + LIST_PATH.format(vin=vin, identifier=identifier) + # The endpoint returns HTTP 500 without the matching request type. + data = self._get_json(url, headers={"type": "partial"}) + return data if isinstance(data, list) else data.get("files", []) + + def download_dataset(self, vin: str, identifier: str, name: str) -> tuple[str, dict]: + url = BASE_URL + DOWNLOAD_PATH.format(vin=vin, identifier=identifier) + with self._authenticated_get(url, {"filename": name, "type": "partial"}) as response: + try: + with zipfile.ZipFile(io.BytesIO(response.content)) as archive: + member = next((name for name in archive.namelist() if name.lower().endswith(".json")), None) + if member is None: + raise DatasetError("No JSON inside downloaded dataset") + data = json.loads(archive.read(member).decode("utf-8")) + except (zipfile.BadZipFile, ValueError) as error: + raise DatasetError("Could not read downloaded dataset") from error + return name.removesuffix(".zip") + ".json", data diff --git a/packages/modules/vehicles/vweuda/api_test.py b/packages/modules/vehicles/vweuda/api_test.py new file mode 100644 index 0000000000..1a05fb24fc --- /dev/null +++ b/packages/modules/vehicles/vweuda/api_test.py @@ -0,0 +1,102 @@ +import io +import json +import zipfile +from unittest.mock import Mock +from urllib.parse import parse_qs + +import pytest +from modules.common.req import get_http_session + +from modules.vehicles.vweuda import api + + +def test_login_stops_at_callback_and_posts_relay_state(requests_mock): + signin = "https://identity.vwgroup.io/signin-service/login" + authenticate = signin + "/authenticate" + callback = api.BASE_URL + api.CALLBACK_LOGIN_PATH + "?code=secret" + cms = api.BASE_URL + "/missing-cms-page" + form = '
' + form += '' + requests_mock.get(api.OIDC_AUTHORIZE_URL, status_code=302, headers={"Location": signin}) + requests_mock.get(signin, text=form) + password_page = '' + requests_mock.post(signin, status_code=302, headers={"Location": authenticate + "?relayState=relay"}) + requests_mock.get(authenticate, text=password_page) + credentials = requests_mock.post(authenticate, status_code=302, headers={"Location": callback}) + requests_mock.get(callback, status_code=302, headers={ + "Location": cms, "Set-Cookie": "access_token=token; Path=/; Secure", + }) + with get_http_session() as session: + client = api.EudaApiClient(session, "test@example.com", "password") + client.login() + assert client._logged_in + assert all(request.url != cms for request in requests_mock.request_history) + assert credentials.last_request.query == "" + fields = parse_qs(credentials.last_request.text) + assert fields["password"] == ["password"] + assert fields["relayState"] == ["relay"] + + +def test_callback_without_cookie_is_not_successful(requests_mock, monkeypatch): + callback = api.BASE_URL + api.CALLBACK_LOGIN_PATH + with get_http_session() as session: + client = api.EudaApiClient(session, "test@example.com", "password") + form = '
' + requests_mock.get(api.OIDC_AUTHORIZE_URL, text=form) + requests_mock.post(api.OIDC_AUTHORIZE_URL, text=form) + requests_mock.get(callback, status_code=302) + monkeypatch.setattr(client, "_post_login_form", lambda *args, **kwargs: session.get(callback)) + with pytest.raises(api.AuthError, match="cookie missing"): + client.login() + assert not client._logged_in + + +@pytest.mark.parametrize("status", [404, 429, 500, 503]) +def test_transient_login_errors_remain_retryable(requests_mock, status): + requests_mock.get(api.OIDC_AUTHORIZE_URL, status_code=status) + with get_http_session() as session: + with pytest.raises(api.ApiError) as error: + api.EudaApiClient(session, "test@example.com", "password").login() + assert not isinstance(error.value, api.AuthError) + assert error.value.status == status + + +def test_json_session_expiry_retries_only_once(requests_mock, monkeypatch): + url = api.BASE_URL + "/test-json" + response = requests_mock.get(url, [{"status_code": 401}, {"status_code": 403}]) + with get_http_session() as session: + client = api.EudaApiClient(session, "test@example.com", "password") + client._logged_in = True + login = Mock() + monkeypatch.setattr(client, "login", login) + with pytest.raises(api.ApiError) as error: + client._get_json(url) + assert error.value.status == 403 + login.assert_called_once() + assert response.call_count == 2 + + +def test_helper_http_error_preserves_login_error_body(requests_mock): + requests_mock.get(api.OIDC_AUTHORIZE_URL, status_code=400, + text='') + with get_http_session() as session: + with pytest.raises(api.AuthError, match="Account locked"): + api.EudaApiClient(session, "test@example.com", "password").login() + + +def test_download_session_expiry_preserves_zip_decoding(requests_mock, monkeypatch): + payload = {"vin": "WVW00000000000001", "Data": []} + archive = io.BytesIO() + with zipfile.ZipFile(archive, "w") as handle: + handle.writestr("data.json", json.dumps(payload)) + url = api.BASE_URL + api.DOWNLOAD_PATH.format(vin=payload["vin"], identifier="id") + requests_mock.get(url, [{"status_code": 401}, {"content": archive.getvalue()}]) + with get_http_session() as session: + client = api.EudaApiClient(session, "test@example.com", "password") + client._logged_in = True + login = Mock() + monkeypatch.setattr(client, "login", login) + name, data = client.download_dataset(payload["vin"], "id", "data.zip") + assert name == "data.json" + assert data == payload + login.assert_called_once() diff --git a/packages/modules/vehicles/vweuda/libeuda.py b/packages/modules/vehicles/vweuda/libeuda.py old mode 100755 new mode 100644 index 6ac67824e0..8f64e2b151 --- a/packages/modules/vehicles/vweuda/libeuda.py +++ b/packages/modules/vehicles/vweuda/libeuda.py @@ -1,1248 +1,112 @@ -#!/usr/bin/env python3 -"""Client for the VW EU Data Act portal (OIDC login + data delivery).""" +"""Synchronous portal requests with a module-owned in-memory cache.""" -from __future__ import annotations - -from typing import Union -from asyncio import new_event_loop, set_event_loop -import uuid import logging -import aiohttp -import io -import json import re -import zipfile -from html.parser import HTMLParser -from urllib.parse import urlencode, urljoin, urlparse -from datetime import datetime, timezone, timedelta import time -import threading -import asyncio -# from helpermodules.pub import Pub -import glob -import os -import os.path -from pathlib import Path -from collections import deque -from helpermodules.constants import RAMDISK_PATH -from modules.common.abstract_vehicle import VehicleUpdateData -from modules.vehicles.vweuda.config import VWEUDA - -"""Constants for the VW EU Data Act integration.""" - -# --- Portal / OIDC endpoints --------------------------------------------- -BASE_URL = "https://eu-data-act.drivesomethinggreater.com" -IDENTITY_BASE = "https://identity.vwgroup.io" - -# Brand is part of the OIDC state; VW passenger cars by default. -# BRAND = "VOLKSWAGEN_PASSENGER_CARS" -CALLBACK_LOGIN_PATH = "/services/callbacklogin" - -BRANDS: dict[str, dict[str, str]] = { - "volkswagen": { - "display_name": "Volkswagen", - "client_id": "9b58543e-1c15-4193-91d5-8a14145bebb0@apps_vw-dilab_com", - "state": "VOLKSWAGEN_PASSENGER_CARS", - }, - "audi": { - "display_name": "Audi", - "client_id": "cc29b87a-5e9a-4362-aecf-5adea6b01bbb@apps_vw-dilab_com", - "state": "AUDI", - }, - "skoda": { - "display_name": "Škoda", - "client_id": "3ea88bf9-1d4e-4a68-b3ad-4098c1f1d246@apps_vw-dilab_com", - "state": "SKODA", - }, - "seat": { - "display_name": "SEAT", - "client_id": "f85e5b69-e3b2-43aa-9c0d-1b7d0e0b576f@apps_vw-dilab_com", - "state": "SEAT", - }, - "cupra": { - "display_name": "CUPRA", - "client_id": "f85e5b69-e3b2-43aa-9c0d-1b7d0e0b576f@apps_vw-dilab_com", - "state": "CUPRA", - }, -} - - -DEFAULT_BRAND = "volkswagen" -DEFAULT_COUNTRY = "de" -DEFAULT_LANGUAGE = "en" - - -def ano_part(original): - return original[0] + ('*' * (len(original) - 2)) + original[-1] - - -def ano_email(original): - user, at, domain = original.partition('@') - domain_name, dot, tld = domain.rpartition('.') - return '{}@{}.{}'.format(ano_part(user), ano_part(domain_name), tld) - - -def ano_vin(vin: str) -> str: - return vin[:-9] + "*********" - - -def get_oidc_client_id(brand: str = DEFAULT_BRAND) -> str: - """Return the OIDC client_id for the given brand.""" - return BRANDS.get(brand, BRANDS[DEFAULT_BRAND])["client_id"] - - -def get_oidc_state(brand: str = DEFAULT_BRAND) -> str: - """Return the OIDC state for the given brand.""" - brand_state = BRANDS.get(brand, BRANDS[DEFAULT_BRAND])["state"] - return f"{DEFAULT_COUNTRY}__{DEFAULT_LANGUAGE}__{brand_state}" - - -# OIDC: we build the authorize URL directly instead of using the portal's -# /services/redirect/authentication servlet, which returns HTTP 500 for -# non-browser clients (it depends on AEM browser session state). -OIDC_AUTHORIZE_URL = IDENTITY_BASE + "/oidc/v1/authorize" -# OIDC_CLIENT_ID = "9b58543e-1c15-4193-91d5-8a14145bebb0@apps_vw-dilab_com" -OIDC_SCOPE = "openid cars profile" -OIDC_REDIRECT_URI = BASE_URL + "/login" -# state encodes country__language__brand (echoed back to the portal callback). -# DEFAULT_COUNTRY = "si" -# DEFAULT_LANGUAGE = "sl" -CONF_BRAND = "brand" -# OIDC_STATE = f"{DEFAULT_COUNTRY}__{DEFAULT_LANGUAGE}__{BRAND}" - -# Legacy constants for backward compatibility (default to VW) -OIDC_CLIENT_ID = BRANDS[DEFAULT_BRAND]["client_id"] -OIDC_STATE = get_oidc_state(DEFAULT_BRAND) - -# proxy_api paths (relative to BASE_URL) -VEHICLES_PATH = "/proxy_api/consent/me/vehicles" -RELATION_PATH = "/proxy_api/vum/v2/users/me/relations/{vin}" -METADATA_PATH = "/proxy_api/euda-apim/datarequest/vehicles/{vin}/metadata/partial" -LIST_PATH = "/proxy_api/euda-apim/datadelivery/vehicles/{vin}/{identifier}/list" -DOWNLOAD_PATH = "/proxy_api/euda-apim/datadelivery/vehicles/{vin}/{identifier}/download" - -DOMAIN = "vw_eu_data_act" -VERSION = "0.3.3" -USER_AGENT = f"{DOMAIN}/{VERSION}" - -# --- Config entry keys ---------------------------------------------------- -CONF_EMAIL = "email" -CONF_PASSWORD = "password" -CONF_VIN = "vin" -CONF_IDENTIFIER = "identifier" -CONF_NICKNAME = "nickname" - -# --- Scheduling ----------------------------------------------------------- -# Datasets land ~every 15 min; refresh shortly after the next expected drop. -DATASET_INTERVAL = timedelta(minutes=15) -POST_DATASET_BUFFER = timedelta(seconds=45) -RETRY_INTERVAL = timedelta(minutes=1) -MIN_INTERVAL = timedelta(seconds=30) - -# Files with this suffix carry no payload and are skipped. -NO_CONTENT_SUFFIX = "_no_content_found.zip" - -POLL_INTERVAL = 60 # polling interval in seconds -CYCLE_INTERVAL = 600 # cycle interval in seconds -INITIAL_RESULT_WAIT = 30 # seconds to wait for the first background result -EUDA_THREADNAME = "soc_bt_ev" -UTC = None -KEEP_JSON = 5 -DATA_PATH = Path(__file__).resolve().parents[4] / "data" / "modules" / "vweuda" -JSON_PATH = Path(str(RAMDISK_PATH) + '/vweuda') -storeFileName = '/data_' -MODULE_TYPE = 'vweuda' - -# VIN-Brand map -VIN_BRAND_MAP = { - "WAU": "audi", # Audi car - "WA1": "audi", # Audi SUV - "WUA": "audi", # Audi Sport car - "WU1": "audi", # Audi Sport SUV - "99A": "audi", # Audi 2016- - "AAA": "audi", # Audi South-Africa- - "TRU": "audi", # Audi Hungary - "VSS": "cupra", # Seat/Cupra - assume cupra as default - "NAD": "skoda", # Skoda - "TMB": "skoda", # Skoda (Czech Republic) - "Y6U": "skoda", # Skoda Auto made by Eurocar (Ukraine) - "VWV": "volkswagen", # Volkswagen Spain - "WVG": "volkswagen", # SUV/Touran - "WVW": "volkswagen", # Passenger Cars - "WV1": "volkswagen", # Commercial Vehicles - "WV2": "volkswagen", # Commercial Vehicles - "WV3": "volkswagen", # Commercial Vehicles - "WV4": "volkswagen", # Commercial Vehicles - "WV5": "volkswagen", # Commercial Vehicles -} - - -_LOGGER = logging.getLogger(__name__) - - -class ApiError(Exception): - """Generic API failure. - - Carries the HTTP ``status`` when the failure came from an HTTP response, so - callers can branch on it (e.g. retry 5xx) without grepping the message - string. ``None`` for non-HTTP failures (connection errors, bad JSON, …). - """ - - def __init__(self, message: str, *, status: int | None = None) -> None: - super().__init__(message) - self.status = status - - -class AuthError(ApiError): - """Authentication failed or session expired.""" - - -# Non-5xx statuses the identity provider returns for reasons unrelated to the -# credentials: 404 when the sign-in service is (re)deploying / a step URL is -# temporarily missing, 429 when we are rate limited. -_TRANSIENT_LOGIN_STATUSES = frozenset({404, 429}) - - -def _is_transient_login_status(status: int) -> bool: - """True when a login-step HTTP status is an upstream problem, not bad credentials.""" - return status >= 500 or status in _TRANSIENT_LOGIN_STATUSES - - -# Redirect statuses the identity provider / portal use while bouncing the -# browser back after the credentials POST. All of them are followed with a GET. -_LOGIN_REDIRECT_STATUSES = frozenset({301, 302, 303}) -# Observed chain is 6 hops (authenticate -> oauth/sso -> consent -> oauth -# callback/success -> portal /login -> /services/callbacklogin); anything much -# longer is a loop. -_MAX_LOGIN_REDIRECTS = 10 -# Cookie the portal callback sets once the OIDC code has been exchanged. -_SESSION_COOKIE = "access_token" - - -def _redact_query(url: str) -> str: - """Drop the query string for logging: redirect URLs carry the OIDC code.""" - return url.split("?", 1)[0] - - -def _is_portal_callback(url: str) -> bool: - """True if ``url`` is the portal's OIDC callback (``/services/callbacklogin``).""" - parsed = urlparse(url) - return parsed.netloc == urlparse(BASE_URL).netloc and parsed.path.startswith( - CALLBACK_LOGIN_PATH - ) - - -class _FormParser(HTMLParser): - """Extract the first
action and all hidden/input fields.""" - - def __init__(self) -> None: - super().__init__() - self.action: str | None = None - self.fields: dict[str, str] = {} - self._in_form = False - self._done = False # only capture the first form - - def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: - if self._done: - return - a = dict(attrs) - if tag == "form" and self.action is None: - self.action = a.get("action") - self._in_form = True - elif tag == "input" and self._in_form: - name = a.get("name") - if name: - self.fields[name] = a.get("value") or "" - - def handle_endtag(self, tag: str) -> None: - if tag == "form" and self._in_form: - self._in_form = False - self._done = True - - -def _parse_form(html: str) -> _FormParser: - p = _FormParser() - p.feed(html) - return p - - -def _extract_template_model(html: str) -> dict: - """Extract the VW identity ``templateModel`` JSON embedded in the page. - - The signin/authenticate pages carry their form state (hmac, relayState, - prefilled email, postAction, error) in a JS object rather than HTML inputs: - - window._IDK = { templateModel: { ... }, csrf_token: '...' } - """ - idx = html.find("templateModel") - if idx == -1: - return {} - brace = html.find("{", idx) - if brace == -1: - return {} - depth = 0 - for i in range(brace, len(html)): - c = html[i] - if c == "{": - depth += 1 - elif c == "}": - depth -= 1 - if depth == 0: - try: - return json.loads(html[brace: i + 1]) - except ValueError: - return {} - return {} - - -def _extract_csrf(html: str) -> str | None: - """Pull the csrf_token out of the identity page's JS.""" - m = re.search(r"csrf_token\s*[:=]\s*['\"]([^'\"]+)['\"]", html) - return m.group(1) if m else None - - -def _login_fields(html: str) -> tuple[dict[str, str], str | None]: - """Collect the fields needed to POST a VW identity login step. - - Merges HTML hidden inputs with the JS templateModel/csrf so it works - whether the page renders inputs server-side (email step) or via JS - (password step). Returns (fields, form_action).""" - form = _parse_form(html) - fields: dict[str, str] = dict(form.fields) - model = _extract_template_model(html) - if model: - for key in ("hmac", "relayState"): - if model.get(key): - fields[key] = model[key] - email = (model.get("emailPasswordForm") or {}).get("email") - if email: - fields.setdefault("email", email) - csrf = _extract_csrf(html) - if csrf: - fields.setdefault("_csrf", csrf) - return fields, form.action - - -def _login_error(html: str) -> str | None: - """Return a human-readable login error from the page, if present.""" - model = _extract_template_model(html) - err = model.get("error") or model.get("errorCode") - if isinstance(err, dict): - return err.get("text") or err.get("errorCode") or str(err) - return str(err) if err else None - - -def _extract_vins(payload) -> list[dict]: - """Best-effort extraction of vehicles from the (undocumented) vehicles body. - - Returns a list of {vin, nickname?} dicts. Walks the JSON for any 17-char - VIN-like identifier so it is robust to wrapper shape ({vehicles:[]}, list, …). - """ - vins: dict[str, dict] = {} - - def walk(node): - if isinstance(node, dict): - vin = node.get("vin") or node.get("vehicleIdentificationNumber") - if isinstance(vin, str) and len(vin) == 17: - vins.setdefault(vin, {"vin": vin}) - nick = node.get("vehicleNickname") or node.get("nickname") or node.get("modelName") - if nick: - vins[vin]["nickname"] = nick - for v in node.values(): - walk(v) - elif isinstance(node, list): - for v in node: - walk(v) - - walk(payload) - return list(vins.values()) - - -class EudaApiClient: - """Authenticated client for the EU Data Act portal.""" - - def __init__(self, session: aiohttp.ClientSession, email: str, password: str, brand: str = "volkswagen") -> None: - self._session = session - self._email = email - self._password = password - self._brand = brand - self._logged_in = False - - # -- low level --------------------------------------------------------- - - async def _get(self, url: str, *, headers: dict | None = None, allow_redirects: bool = True): - h = {"User-Agent": USER_AGENT, **(headers or {})} - return await self._session.get(url, headers=h, allow_redirects=allow_redirects) - - async def _post_login_form(self, url: str, *, data: dict[str, str], referer: str): - """POST a login form and follow redirects only as far as the portal callback. - - The browser flow after the credentials POST is a redirect chain that ends - on ``/services/callbacklogin``, which exchanges the OIDC code, sets the - ``access_token`` session cookie and redirects once more to a localized - CMS landing page. That last page is irrelevant to authentication (and is - missing for some locales), so redirects are followed by hand and the - chain is cut at the callback. aiohttp stores each hop's cookies in the - session jar whether or not it follows the redirect itself. - - Returns the callback response, or the first non-redirect response if - the chain ends elsewhere (bad credentials, identity-provider error). - """ - resp = await self._session.post( - url, - data=data, - headers={"User-Agent": USER_AGENT, "Referer": referer}, - allow_redirects=False, - ) - for redirects in range(_MAX_LOGIN_REDIRECTS + 1): - if _is_portal_callback(str(resp.url)): - return resp - if resp.status not in _LOGIN_REDIRECT_STATUSES: - return resp - if redirects == _MAX_LOGIN_REDIRECTS: - resp.release() - raise ApiError(f"Login redirect chain exceeded {_MAX_LOGIN_REDIRECTS} hops") - location = resp.headers.get("Location") - if not location: - return resp - next_url = urljoin(str(resp.url), location) - previous_url = str(resp.url) - _LOGGER.debug("login step4: %s -> %s", resp.status, _redact_query(next_url)) - resp.release() - resp = await self._get( - next_url, - headers={"Referer": previous_url}, - allow_redirects=False, - ) - - # -- authentication ---------------------------------------------------- - - async def async_login(self) -> None: - """Run the full OIDC login, populating the session cookie jar.""" - try: - await self._do_login() - except aiohttp.ClientError as err: - raise ApiError(f"Network error during login: {err}") from err - self._logged_in = True - - async def _do_login(self) -> None: - # 1. Start the OIDC flow directly at the identity provider. We build the - # authorize URL ourselves because the portal's - # /services/redirect/authentication servlet returns HTTP 500 for - # non-browser clients. - authorize_url = self._build_authorize_url(self._brand) - _LOGGER.debug("login step1: authorize url = %s", authorize_url) - async with await self._get(authorize_url) as resp: - signin_url = str(resp.url) - signin_html = await resp.text() - if _is_transient_login_status(resp.status): - raise ApiError( - f"Identity provider error on sign-in page (HTTP {resp.status})", - status=resp.status, - ) - _LOGGER.debug("login step2: signin page = %s (%d bytes)", signin_url, len(signin_html)) - - # 2. POST the email (identifier step). Fields come from HTML inputs - # and/or the JS templateModel (hmac, _csrf, relayState). - fields, action = _login_fields(signin_html) - _LOGGER.debug("login step2: action=%s fields=%s", action, sorted(fields)) - if "hmac" not in fields or "_csrf" not in fields: - raise AuthError( - f"Could not parse the sign-in form (fields found: {sorted(fields)})" - ) - fields["email"] = self._email - identifier_action = urljoin(signin_url, action or "") - async with self._session.post( - identifier_action, - data=fields, - headers={"User-Agent": USER_AGENT, "Referer": signin_url}, - ) as resp: - authenticate_url = str(resp.url) - authenticate_html = await resp.text() - status = resp.status - _LOGGER.debug( - "login step3: after identifier POST status=%s url=%s", status, authenticate_url - ) - if _is_transient_login_status(status): - raise ApiError( - f"Identity provider error on identifier step (HTTP {status})", - status=status, - ) - - # 3. The identifier step lands on the password (authenticate) page, - # whose hidden fields live in the JS templateModel, not HTML inputs. - fields2, action2 = _login_fields(authenticate_html) - _LOGGER.debug("login step3: action=%s fields=%s", action2, sorted(fields2)) - if "hmac" not in fields2 or "_csrf" not in fields2: - err = _login_error(authenticate_html) - raise AuthError( - err - or "Identity portal did not return the password form - check the " - "email address (or the login flow changed)" - ) - fields2["email"] = self._email - fields2["password"] = self._password - # The browser posts to the clean /login/authenticate URL with relayState - # in the body; posting to authenticate_url (which carries ?relayState=) - # duplicates it and is rejected with HTTP 400. Strip the query. - if action2: - authenticate_action = urljoin(authenticate_url, action2) - else: - authenticate_action = authenticate_url.split("?", 1)[0] - _LOGGER.debug("login step4: POST credentials to %s", authenticate_action) - - # 4. POST credentials; follow the redirect chain back to the portal - # and stop at /services/callbacklogin, which sets the session cookie. - async with await self._post_login_form( - authenticate_action, - data=fields2, - referer=authenticate_url, - ) as resp: - landing = str(resp.url) - if _is_portal_callback(landing) and resp.status < 400: - # The callback's own redirect (to the CMS landing page) is - # deliberately not followed; the access_token cookie it set is - # the proof the login worked. A failing callback (e.g. 5xx on - # the code exchange) goes through the error handling below. - cookie = resp.cookies.get(_SESSION_COOKIE) - if not getattr(cookie, "value", cookie): - raise AuthError( - "Login callback did not establish a session " - f"({_SESSION_COOKIE} cookie missing)" - ) - _LOGGER.debug("login step4: callback set %s cookie", _SESSION_COOKIE) - return - landing_html = await resp.text() - if resp.status >= 400: - _LOGGER.debug( - "login step4: HTTP %s body[:500]=%s", resp.status, landing_html[:500] - ) - # 5xx / 404 / 429 is the identity provider failing (or - # throttling us), not our credentials — surface it as a - # retryable ApiError so the coordinator keeps the entry alive - # instead of demanding reauthentication. - if _is_transient_login_status(resp.status): - raise ApiError( - f"Identity provider error (HTTP {resp.status})", - status=resp.status, - ) - err = _login_error(landing_html) - raise AuthError(err or f"Login rejected (HTTP {resp.status})") - _LOGGER.debug("login step4: landed on %s", landing) - - # A completed flow returns above from the portal callback. Anything else - # that ended the chain is a failure: bad credentials re-render the - # identity sign-in page (URL still on identity.vwgroup.io/signin-service). - if "signin-service" in landing or "/error" in landing: - raise AuthError("Login failed - check email and password") - raise AuthError(f"Login did not complete (ended at {_redact_query(landing)})") - - @staticmethod - def _build_authorize_url(brand: str = "volkswagen") -> str: - """Construct the OIDC authorize URL (bypasses the broken AEM servlet).""" - params = { - "client_id": get_oidc_client_id(brand), - "response_type": "code", - "scope": OIDC_SCOPE, - "state": get_oidc_state(brand), - "redirect_uri": OIDC_REDIRECT_URI, - "prompt": "login", - } - return f"{OIDC_AUTHORIZE_URL}?{urlencode(params)}" - - # -- authenticated requests ------------------------------------------- - - async def _get_json(self, url: str, *, headers: dict | None = None, _retry: bool = True): - async with await self._get(url, headers=headers) as resp: - if resp.status in (401, 403) and _retry: - _LOGGER.debug("Session expired (%s) for %s; re-authenticating", resp.status, url) - self._logged_in = False - await self.async_login() - return await self._get_json(url, headers=headers, _retry=False) - if resp.status >= 400: - raise ApiError(f"GET {url} -> HTTP {resp.status}") - text = await resp.text() - try: - return json.loads(text) - except ValueError as err: - raise ApiError(f"Invalid JSON from {url}: {err}") from err - - async def async_ensure_login(self) -> None: - if not self._logged_in: - await self.async_login() - - async def async_list_vehicles(self) -> list[dict]: - await self.async_ensure_login() - payload = await self._get_json(f"{BASE_URL}{VEHICLES_PATH}?viewPosition=FRONT_LEFT") - vehicles = _extract_vins(payload) - # Always enrich with the friendly vehicleNickname from the relation - # endpoint (the authoritative source, e.g. "ID.3"). - for veh in vehicles: - try: - rel = await self.async_get_relation(veh["vin"]) - nickname = (rel.get("relation") or {}).get("vehicleNickname") - _LOGGER.debug("relation for %s: nickname=%r", veh["vin"], nickname) - if nickname: - veh["nickname"] = nickname - except ApiError as err: - _LOGGER.debug("Could not fetch nickname for %s: %s", veh["vin"], err) - return vehicles - - async def async_get_relation(self, vin: str) -> dict: - await self.async_ensure_login() - # The relation endpoint requires a traceid header; it returns HTTP 400 - # without one. - headers = {"traceid": f"vehicle-relation-fetch-{uuid.uuid4()}"} - return await self._get_json( - f"{BASE_URL}{RELATION_PATH.format(vin=vin)}", headers=headers - ) - - async def async_get_metadata(self, vin: str) -> dict: - """Return the data-request metadata; ``Identifier`` is needed downstream.""" - await self.async_ensure_login() - return await self._get_json(f"{BASE_URL}{METADATA_PATH.format(vin=vin)}") - - async def async_list_datasets(self, vin: str, identifier: str) -> list[dict]: - """Return the rolling list of available zips: [{name, createdOn, size}].""" - await self.async_ensure_login() - url = f"{BASE_URL}{LIST_PATH.format(vin=vin, identifier=identifier)}" - # The list endpoint requires the data-request type header (matching - # metadata/partial); without it the backend returns HTTP 500. - data = await self._get_json(url, headers={"type": "partial"}) - return data if isinstance(data, list) else data.get("files", []) - - async def async_download_dataset(self, vin: str, identifier: str, name: str) -> Union[str, dict]: - """Download a specific zip by name and return the parsed JSON inside it.""" - await self.async_ensure_login() - if name.endswith(NO_CONTENT_SUFFIX): - raise ApiError(f"{name} contains no content") - url = f"{BASE_URL}{DOWNLOAD_PATH.format(vin=vin, identifier=identifier)}" - headers = {"filename": name, "type": "partial"} - async with await self._get(url, headers=headers) as resp: - if resp.status in (401, 403): - self._logged_in = False - await self.async_login() - async with await self._get(url, headers=headers) as resp2: - if resp2.status >= 400: - raise ApiError(f"Download {name} -> HTTP {resp2.status}") - raw = await resp2.read() - elif resp.status >= 400: - raise ApiError(f"Download {name} -> HTTP {resp.status}") - else: - raw = await resp.read() - _data = self._unzip_json(raw, name) - _name = name.replace('.zip', '.json') - return _name, _data +from typing import Optional - @staticmethod - def _unzip_json(raw: bytes, name: str) -> dict: - try: - with zipfile.ZipFile(io.BytesIO(raw)) as zf: - members = [n for n in zf.namelist() if n.lower().endswith(".json")] - if not members: - raise ApiError(f"No JSON inside {name}") - with zf.open(members[0]) as fh: - return json.loads(fh.read().decode("utf-8")) - except (zipfile.BadZipFile, ValueError) as err: - raise ApiError(f"Could not read {name}: {err}") from err - - -async def _async_try_login(session, client) -> str | None: - """Attempt login + vehicle discovery; return an error key or None.""" - try: - await client.async_login() - _vehicles = await client.async_list_vehicles() - _LOGGER.info(f"_vehicles={_vehicles}") - except AuthError: - return "invalid_auth" - except ApiError: - return "cannot_connect" - except Exception: # noqa: BLE001 - _LOGGER.exception("Unexpected error during login") - return "unknown" - return None - - -async def _async_update_data(client, vin, identifier) -> Union[str, str]: - await client.async_ensure_login() - # await client.async_login() - try: - listing = await client.async_list_datasets(vin, identifier) - except AuthError as err: - # Retry soon rather than waiting the full ~15-min cadence. - # self.update_interval = RETRY_INTERVAL - raise ApiError(f"Authentication failed: {err}") from err - except ApiError as err: - # self.update_interval = RETRY_INTERVAL - if "HTTP 400" in str(err): - _LOGGER.info(f"err={err}") - # The data-delivery endpoint returns 400 until the portal has - # finished provisioning a newly enabled continuous data request, - # which can take a few hours. HA keeps retrying until it's ready. - raise ApiError( - "Data delivery not ready yet (HTTP 400). If you just enabled " - "the continuous data request on the portal, it can take a few " - "hours to start; will keep retrying." - ) from err - raise ApiError(str(err)) from err - - # content datasets, oldest -> newest by createdOn - content = sorted( - (e for e in listing if e.get("name") and not e["name"].endswith(NO_CONTENT_SUFFIX)), - key=lambda e: _created_on(e) or datetime.min.replace(tzinfo=timezone.utc), - ) - _LOGGER.debug("refresh: %d listed, %d with content", len(listing), len(content)) - - if not content: - raise ApiError("No datasets with content available yet") - - # Load the newest dataset for live state. (We don't backfill history into - # statistics: importing into recorder-managed sensor entities collides - # with the recorder's own statistics and can corrupt unrelated ones.) - newest = content[-1] - - try: - name, payload = await client.async_download_dataset( - vin, identifier, newest["name"] - ) - except ApiError as err: - raise ApiError(f"Could not download newest dataset: {err}") from err - - return name, payload - - -def _filename_timestamp(name: str) -> datetime | None: - """Parse a YYYYMMDDhhmmss segment from a dataset filename. - - Handles both layouts seen in the wild ("TIMESTAMP_VIN.zip" and - "VIN_TIMESTAMP.zip") by scanning the underscore-separated parts - right-to-left for the first one that parses as a timestamp. - """ - stem = name.rsplit(".", 1)[0] - for part in reversed(stem.split("_")): - try: - return datetime.strptime(part, "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc) - except ValueError: - continue - return None - - -def _created_on(entry: dict) -> datetime | None: - raw = entry.get("createdOn") - if not raw: - return _filename_timestamp(entry.get("name", "")) - try: - return datetime.fromisoformat(raw.replace("Z", "+00:00")) - except ValueError: - return _filename_timestamp(entry.get("name", "")) - - -def get_field_value_by_fieldname(D: dict, field: str) -> str: - ret = None - for f in D: - if f['dataFieldName'] == field: - ret = f['value'] - return ret - - -def get_field_value_by_key(D: dict, key: str, field: str) -> str: - ret = None - for f in D: - if f['key'] == key: - ret = f['value'] - if ret: - _LOGGER.info(f"get_field_value_by_key: field: {field}, {key} -> {ret}") - return ret - - -def get_field_timestamp_by_key(D: dict, key: str) -> str: - ret = None - for f in D: - if f['key'] == key: - ret = f['timestampUtc'] - if ret: - _LOGGER.info(f"get_field_timestamp_by_key {key} -> {ret}") - return ret - - -CAR_TIMESTAMP1 = "car_captured_time" -CAR_TIMESTAMP2 = "timestamp" - - -def get_max_value_by_fieldname(D: dict, field: str) -> str: - ret = "-" - key = None - for f in D: - if f['dataFieldName'] == field: - v = f['value'] - if ret == "-": - ret = v - key = f['key'] - else: - if v > ret: - ret = v - key = f['key'] - if ret == "-": - _LOGGER.info(f"get_max_value_by_fieldname {field}: no match") - ret = None - else: - _LOGGER.info(f"get_max_value_by_fieldname {field} -> {ret}, key={key}") - return ret - - -def utc_to_timestamp(d: str) -> float: - _epoch = datetime(1970, 1, 1) - _utcformat = "%Y-%m-%dT%H:%M:%SZ" - _d = re.sub(r'\....Z', 'Z', d) - _dt = datetime.strptime(_d, _utcformat) - _ts = (_dt - _epoch).total_seconds() - return _ts - - -def parse_vehicle_data(payload: dict) -> dict: - """Extract normalized SoC fields from an EUDA JSON payload.""" - data = payload.get('Data', []) - - # special case state_of_charge with timestamp (Caddy) - soc = get_field_value_by_key(data, 'ae0294b4-1286-3e98-a818-1485b8d88430', 'soc') - soc_timestamp_str = None - if soc is not None: - _LOGGER.info(f"soc {soc} found in state_of_charge") - _ts = get_field_timestamp_by_key(data, 'ae0294b4-1286-3e98-a818-1485b8d88430') - soc_timestamp_str = re.sub(r'\....Z', 'Z', _ts) - - # try to get soc_timestamp as max of all car_captured_time or timestamp fields - if soc_timestamp_str is None: - soc_timestamp_str = get_max_value_by_fieldname(data, CAR_TIMESTAMP1) - if soc_timestamp_str is None: - soc_timestamp_str = get_max_value_by_fieldname(data, CAR_TIMESTAMP2) - - # if soc is None, try sveral other fields - if soc is None: - soc = get_field_value_by_key(data, 'ac1108b1-b8cc-3db9-a663-03d387e42223', 'soc') - if soc is None: - soc = get_field_value_by_key(data, '0a18a053-b4b0-3db1-be44-a6c5dba629b1', 'soc') # Skoda? - if soc is None: - soc = get_field_value_by_key(data, 'f89ed652-d104-3fa6-b7e2-ab7543309e7b', 'soc') - if soc is None: - soc = get_field_value_by_key(data, '506cb83e-f99f-3af3-bbeb-0429b69a78d9', 'soc') - - range = get_field_value_by_key(data, '153e8c40-4c6c-3c17-a11b-0ecc35d55b81', 'range') - if range is None: - range = get_field_value_by_key(data, '0ca40e18-0564-3eda-bcc0-7aee9ef44f04', 'range') - if range is None: - range = get_field_value_by_key(data, '55e0d40b-38ed-3cb5-9dcd-6193df6fc493', 'range') - - odometer = get_field_value_by_key(data, '41c0805c-43e5-313e-9dfb-356cb8d20f7c', 'odometer') - if odometer is None: - odometer = get_field_value_by_key(data, '30cc36fd-71ca-3c09-9296-e94ebd47bd2b', 'odometer') - - if soc_timestamp_str: - soc_timestamp = utc_to_timestamp(soc_timestamp_str) - if soc_timestamp > 1e10: - soc_timestamp = soc_timestamp / 1000 - else: - _LOGGER.warning("soc_timestamp not found!") - - warning = None - if get_field_value_by_fieldname(data, 'setting.bcam_activation') == 'BCAM_ACTIVATION_ACTIVATED': - bcam_threshold = get_field_value_by_fieldname(data, 'battery_care_mode.charge_bcam_threshold') - if bcam_threshold: - warning = f"Battery Care Mode ist im Fahrzeug aktiv und begrenzt die Ladung selbst auf {bcam_threshold}%." - - return { - 'soc': soc, - 'range': range, - 'soc_timestamp': soc_timestamp, - 'soc_timestamp_str': soc_timestamp_str, - 'odometer': odometer, - 'warning': warning, - } - - -class euda(): - - # _LOGGER.setLevel(logging.DEBUG) - client = {} - thread = {} - result = {} - files = {} - tests_done = False - - def __init__(self): - # make sure required folders are there - try: - _LOGGER.debug("DATA_PATH=" + str(DATA_PATH)) - DATA_PATH.mkdir(parents=True, exist_ok=True) - except Exception as e: - _LOGGER.exception("init: dataPath creation failed, dataPath: " + - str(DATA_PATH) + ", error=" + str(e)) - - try: - _LOGGER.debug("JSON_PATH=" + str(JSON_PATH)) - JSON_PATH.mkdir(parents=True, exist_ok=True) - except Exception as e: - _LOGGER.exception("init: JSON_PATH creation failed, dataPath: " + - str(JSON_PATH) + ", error=" + str(e)) - - def save_json_file(self, _name: str, vin: str, _data: dict) -> bool: - status = False - if vin not in euda.files: - euda.files[vin] = deque(maxlen=KEEP_JSON) - if _name not in euda.files[vin]: - euda.files[vin].append(_name) - status = True - - fname = str(JSON_PATH) + '/' + _name - if not os.path.isfile(fname): - with open(fname, 'w') as f: - _LOGGER.debug(f"save json file {fname.replace(vin, ano_vin(vin))}") - json.dump(_data, f, indent=4) - # status = True - else: - _LOGGER.debug(f"file {fname.replace(vin, ano_vin(vin))} not saved because is exists already") - - # cleanup old file except the latest KEEP_JSON! - _l = glob.glob(str(JSON_PATH) + '/*_' + vin + '.json') - _l.sort() - _len = len(_l) - _del = _len - KEEP_JSON - - _la = [] - for _x in _l: - _la.append(_x.replace(vin, ano_vin(vin))) - - _LOGGER.debug(f"cleanup: KEEP_JSON={KEEP_JSON}, _l={_la}\n _len ={_len}, _del={_del}") - if _del > 0: - _del_list = _l[0:_del] - _da = [] - for _x in _del_list: - _da.append(_x.replace(vin, ano_vin(vin))) - _LOGGER.debug(f"cleanup: _del_list={_da}") - for f in _del_list: - os.remove(f) - _LOGGER.debug(f"delete json file {f.replace(vin, ano_vin(vin))}") - - return status +from modules.common.abstract_vehicle import VehicleUpdateData +from modules.common.component_state import CarState +from modules.common import req +from modules.vehicles.vweuda.api import ( + ApiError, AuthError, DatasetError, EudaApiClient, NO_CONTENT_SUFFIX, brand_for_vin, +) +from modules.vehicles.vweuda.vehicle_data import dataset_time, number, parse_vehicle_data +from modules.vehicles.vweuda.config import VWEUDAConfiguration - def update_result_from_payload(self, payload: dict, source: str) -> dict: - """Parse an EUDA payload and publish it in the in-memory result cache.""" - vin = payload['vin'] - result = parse_vehicle_data(payload) - _valid = True - _LOGGER.info(f"latest result=\n{json.dumps(result, indent=4)}") - if vin in euda.result: - _LOGGER.info(f"cache result=\n{json.dumps(euda.result[vin], indent=4)}") - if result['soc_timestamp'] < euda.result[vin]['soc_timestamp']: - _LOGGER.info("thread result skipped, soc_timestamp too old") - _valid = False - if result['soc'] is None: - _LOGGER.info("thread result skipped, no soc found") - _valid = False - if _valid: - cached_odometer = euda.result[vin].get('odometer') - if result['odometer'] is None: - # newer payload carries no odometer reading - keep the - # last known value instead of dropping the whole update - result['odometer'] = cached_odometer - elif cached_odometer is not None: - try: - if float(result['odometer']) < float(cached_odometer): - _LOGGER.info("odometer less than earlier - keep earlier value") - result['odometer'] = cached_odometer - except (TypeError, ValueError): - _LOGGER.warning( - f"could not compare odometer values " - f"(new={result['odometer']!r}, cached={cached_odometer!r}); " - "keeping new value" - ) - euda.result[vin] = result - _LOGGER.info("thread result is valid") - else: - euda.result[vin] = result - _LOGGER.info(f"new VIN {ano_vin(vin)} initialized in euda.result") +log = logging.getLogger(__name__) - _ano_j = {ano_vin(vin): euda.result[vin]} - _LOGGER.info(f"new cache result:\n{json.dumps(_ano_j, indent=4)}") - return result +class EudaVehicle: + """One cache shared by the updater and interval callback, without a worker.""" - def load_latest_json_result(self, vin: str) -> bool: - """Load and parse the newest already downloaded EUDA JSON for a VIN.""" - files = glob.glob(str(JSON_PATH) + '/*_' + vin + '.json') - files.sort() - if not files: - return False - latest = files[-1] - try: - with open(latest) as f: - payload = json.load(f) - self.update_result_from_payload(payload, latest) - return True - except Exception as err: - _LOGGER.exception(f"failed to load latest EUDA JSON {latest}: {err}") + def __init__(self, configuration: VWEUDAConfiguration, vehicle: int) -> None: + self.configuration = configuration + self.vehicle = vehicle + self.vin = configuration.vin + self.last_request_failed = False + self._result: Optional[dict] = None + self._processed: set[str] = set() + + def get_request_interval_override(self) -> Optional[float]: + return 60 if self.last_request_failed else None + + def _merge(self, result: dict) -> bool: + result = dict(result) + for field in ("soc", "soc_timestamp", "range", "odometer"): + result[field] = number(result.get(field)) + if result["soc"] is None or result["soc"] > 100 or not result["soc_timestamp"]: return False - - async def get_module_type(self, vehicle: int) -> str: - topic = f"openWB/vehicle/{vehicle}/soc_module/config" - conf = os.popen(f"mosquitto_sub -C 1 -t {topic}").read() - _LOGGER.debug(f"thread loop: conf={conf}") - _type = json.loads(conf)['type'] - return _type - - # eudaThread - async def async_eudaThread(self, username: str, password: str, vin: str, vehicle: int): - if vin[0:3] not in VIN_BRAND_MAP: - _LOGGER.warning(f"VIN {vin[0:3]} not in brand map, use {DEFAULT_BRAND}") - brand = VIN_BRAND_MAP.get(vin[0:3], DEFAULT_BRAND) - _LOGGER.info(f"async Thread started, brand={brand}") - try: - async with aiohttp.ClientSession(headers={'Connection': 'keep-alive'}, - connector_owner=False) as session: - client_id = f"{vehicle}" - _k = str(euda.client.keys()) - _LOGGER.info(f"libeuda.Thread client at entry: euda.client.keys={_k}") - if client_id not in euda.client: - _LOGGER.debug(f"create new client, key={client_id}") - euda.client[client_id] = {} - euda.client[client_id] = EudaApiClient(session, username, password, brand) - _k = str(euda.client.keys()) - _LOGGER.info(f"libeuda.Thread client: euda.client.keys={_k}") - - meta = None - while meta is None: - try: - meta = await euda.client[client_id].async_get_metadata(vin) - except ApiError as err: - if "HTTP 500" in str(err): - _LOGGER.info(f"Portal not ready/get_metadata, wait {POLL_INTERVAL} seconds") - else: - _LOGGER.exception(f"APIError/get_metadata: {err}") - meta = None - except Exception as err: - _LOGGER.exception(f"Exception/get_metadata: {err}") - meta = None - if meta is None: - time.sleep(POLL_INTERVAL) - - identifier = meta.get("Identifier") - - # thread main loop - _active = True - while _active: - _type = await self.get_module_type(vehicle) - _LOGGER.info(f"thread loop: ev{vehicle} module type={_type}") - if _type != MODULE_TYPE: - _LOGGER.info(f"vehicle {vehicle} is not using module vweuda: terminate now") - _active = False - continue - try: - _data = None - while _data is None: - try: - _name, _data = await _async_update_data(euda.client[client_id], vin, identifier) - except ApiError as err: - if "HTTP 500" in str(err): - _LOGGER.info(f"Portal not ready/update_data, wait {POLL_INTERVAL} seconds") - _data = None - else: - _LOGGER.exception(f"thread APIError {err}") - _data = None - except Exception as err: - _LOGGER.exception(f"thread Exception {err}") - _data = None - if _data is None: - await asyncio.sleep(POLL_INTERVAL) - - vin = _data['vin'] - self.save_json_file(_name, vin, _data) - self.update_result_from_payload(_data, _name) - _LOGGER.info(f"sleep {CYCLE_INTERVAL} seconds") - await asyncio.sleep(CYCLE_INTERVAL) - - except Exception as e: - _LOGGER.exception(f"thread loop failed 0, exception={e}") - - except Exception as e: - _LOGGER.exception(f"thread body failed 0, exception={e}") - - def eudaThread(self, username: str, password: str, vin: str, vehicle: int): - _LOGGER.info(f"sync libeuda.eudaThread {threading.current_thread().name} started") - asyncio.run(self.async_eudaThread(username, password, vin, vehicle)) - _LOGGER.info(f"sync libeuda.eudaThread {threading.current_thread().name} ended") - - def check_tests(self): - _l = glob.glob(str(DATA_PATH) + '/test_*' + '.json') - for _t in _l: - _vin = _t[_t.index('_')+1:].replace('.json', '') - _LOGGER.info(f"found test file: {_t}, vin={_vin}") - with open(_t) as f: - payload = json.load(f) + if self._result is not None: + if result["soc_timestamp"] < self._result["soc_timestamp"]: + return False + if result["range"] is None: + result["range"] = self._result.get("range") + previous_odometer = self._result.get("odometer") + if previous_odometer is not None: + result["odometer"] = max(result["odometer"] or 0, previous_odometer) + self._result = result + return True + + def _poll(self, client: EudaApiClient) -> None: + identifier = client.get_metadata(self.vin).get("Identifier") + if not identifier: + raise ApiError("Kontinuierlicher Datenabruf im Portal noch nicht bereit") + listing = client.list_datasets(self.vin, identifier) + content = sorted((entry for entry in listing + if entry.get("name") and not entry["name"].endswith(NO_CONTENT_SUFFIX)), key=dataset_time) + if not content: + raise ApiError("Im Portal sind noch keine Datenpakete mit Inhalt verfügbar") + self._processed.intersection_update(entry["name"] for entry in content) + for entry in content: + name = entry["name"] + if name in self._processed: + continue + try: + _, payload = client.download_dataset(self.vin, identifier, name) + if payload.get("vin") != self.vin: + raise ValueError("VIN mismatch") result = parse_vehicle_data(payload) - - test_result = {} - test_result[_vin] = result - _ano_j = json.dumps(test_result, indent=4).replace(_vin, ano_vin(_vin)) - _LOGGER.info(f"test_result, vin={_vin}:\n{_ano_j}") - - async def get_status(self, - conf: VWEUDA, - vehicle: int, - vehicle_update_data: VehicleUpdateData) -> Union[int, float, str, float, float, str]: - - # error codes SOCERR-xx raised: - # SOCERR-00: general error - # SOCERR-01: login problem, username, password wrong, account locked, etc. - # SOCERR-02: vehicle not (yet) found in portal, VIN wrong? - - if not euda.tests_done: - self.check_tests() - euda.tests_done = True - - self.username = conf.configuration.user_id - self.password = conf.configuration.password - self.vin = conf.configuration.vin - self.vehicle = vehicle - _LOGGER.debug(f"update_data:{vehicle_update_data}") - self.average_consump = vehicle_update_data.average_consump - self.battery_capacity = vehicle_update_data.battery_capacity - + self._merge(result) + except (DatasetError, ValueError, TypeError, KeyError, AttributeError): + log.warning("EUDA ev%s: Ungültiges Datenpaket übersprungen", self.vehicle) + self._processed.add(name) + time.sleep(.5) + + def fetch(self, update: VehicleUpdateData) -> CarState: + error_message = None try: - self.storeFile = str(DATA_PATH) + storeFileName + str(self.vin) + '.json' - if os.path.isfile(self.storeFile): - _LOGGER.debug(f"load data from {self.storeFile.replace(self.vin, ano_vin(self.vin))}") - with open(self.storeFile) as f: - data = json.load(f) - else: - data = {} - data['currentSOC_pct'] = str(0) - data['cruisingRangeElectric_km'] = str(0) - data['carCapturedTimestamp'] = "1970-01-01T00:00:00Z" - data['soc_timestamp'] = str(0) - data['odometer'] = str(0) - - thread_id = f"{self.vehicle}" - euda.thread[thread_id] = {} - euda.thread[thread_id]['name'] = f"{EUDA_THREADNAME}{self.vehicle}" - euda.thread[thread_id]['thread'] = None - for t in threading.enumerate(): - if t.name == euda.thread[thread_id]['name']: - _LOGGER.debug(f"thread {t.name} exists already") - euda.thread[thread_id]['thread'] = t - if euda.thread[thread_id]['thread'] is None: - _LOGGER.debug(f"{euda.thread[thread_id]['name']} not found: starting now") - euda.thread[thread_id]['thread'] = threading.Thread(target=self.eudaThread, - name=euda.thread[thread_id]['name'], - args=(self.username, - self.password, - self.vin, - self.vehicle), - daemon=True) - euda.thread[thread_id]['thread'].start() - - if self.vin not in euda.result: - self.load_latest_json_result(self.vin) - - wait_until = time.time() + INITIAL_RESULT_WAIT - while self.vin not in euda.result and time.time() < wait_until: - _LOGGER.info(f"wait for first EUDA result for VIN {ano_vin(self.vin)}") - time.sleep(1) - - warning = None - if self.vin in euda.result: - _LOGGER.debug(f"vehicle match: {ano_vin(self.vin)}") - _ano_j = {} - for vin in euda.result: - _ano_j[ano_vin(vin)] = euda.result[vin] - _LOGGER.info(f"result from thread:\n{json.dumps(_ano_j, indent=4)}") - warning = euda.result[self.vin].get('warning') - soc = euda.result[self.vin]['soc'] - range = euda.result[self.vin]['range'] - try: - soc_pct = float(soc) if soc is not None else 0.0 - except (TypeError, ValueError): - soc_pct = 0.0 - if soc_pct > 0 and range is None: - range = int(soc_pct * float(self.battery_capacity) / float(self.average_consump)) - _LOGGER.warning(f"no range delivered, calculate range = {range}km") - - ts = euda.result[self.vin]['soc_timestamp'] - ts_str = euda.result[self.vin]['soc_timestamp_str'] - odometer = euda.result[self.vin]['odometer'] - - _LOGGER.debug(f"vin = {self.vin}") - _LOGGER.debug(f"soc = {soc}") - _LOGGER.debug(f"range = {range}") - _LOGGER.debug(f"soc_timestamp = {ts}") - _LOGGER.debug(f"soc_timestamp_str = {ts_str}") - _LOGGER.debug(f"odometer = {odometer}") - - data_modified = False - if soc and str(soc) != data['currentSOC_pct']: - data['currentSOC_pct'] = str(soc) - data_modified = True - if range and str(range) != data['cruisingRangeElectric_km']: - data['cruisingRangeElectric_km'] = str(range) - data_modified = True - if ts and str(ts) != data['soc_timestamp']: - data['soc_timestamp'] = str(ts) - data_modified = True - if ts_str and str(ts_str) != data['carCapturedTimestamp']: - data['carCapturedTimestamp'] = str(ts_str) - data_modified = True - if odometer and str(odometer) != data['odometer']: - data['odometer'] = str(odometer) - data_modified = True - - # save data to file if modified - if data_modified or not os.path.isfile(self.storeFile): - _LOGGER.info(f"save data to {self.storeFile}") - with open(self.storeFile, 'w') as f: - json.dump(data, f, indent=4) - - else: - _LOGGER.error(f"SOCERR-02: Für VIN {ano_vin(self.vin)} wurden (noch) keine Daten gefunden") - # raise Exception(f"SOCERR-02: Für VIN {self.vin} wurden (noch) keine Daten gefunden") - - _LOGGER.info(f"return data:\n{json.dumps(data, indent=4)}") - soc = data['currentSOC_pct'] - range = data['cruisingRangeElectric_km'] - ts = data['soc_timestamp'] - ts_str = data['carCapturedTimestamp'] - odometer = data['odometer'] - _LOGGER.info(f"get_status: soc={soc}, range={range}, ts={ts}, ts_str={ts_str}, odometer={odometer}") - - # for test only: - # set soc_timestamp to 0 to avoid computed state being later than this reported state - # topic = f"openWB/vehicle/{self.vehicle}/get/soc_timestamp" - # ep0 = 0 - # _LOGGER.info(f"get_status: publish soc_timestamp as 0: topic: {topic}, message: {ep0}") - # Pub().pub(topic, ep0) - - return float(soc), float(range), float(ts), ts_str, float(odometer), warning - except Exception as e: - _LOGGER.exception(f"get_status failed 0, exception={e}") - # if exception is a SOCERR reraise it, otherwise raise general SOCERR-00 - if "SOCERR" in str(e): - raise e + conf = self.configuration + if not conf.user_id or not conf.password: + raise AuthError("SOCERR-01: Benutzername und Kennwort müssen konfiguriert sein") + if not isinstance(self.vin, str) or not re.fullmatch(r"[A-HJ-NPR-Z0-9]{17}", self.vin): + raise ApiError("SOCERR-02: Bitte eine gültige 17-stellige VIN konfigurieren") + # A fresh session per scheduled attempt is closed after the request. + with req.get_http_session() as session: + self._poll(EudaApiClient(session, conf.user_id, conf.password, brand_for_vin(self.vin))) + if self._result is None: + raise ApiError("Es wurden noch keine gültigen Fahrzeugdaten empfangen") + except Exception as error: + self.last_request_failed = True + error_message = str(error) + if isinstance(error, ApiError): + log.warning("EUDA ev%s: Abruf fehlgeschlagen: %s", self.vehicle, error_message) else: - _t = f"SOCERR-00: Für User {ano_email(self.username)}" - _t = _t + f" und VIN {ano_vin(self.vin)} wurden keine Daten empfangen" - raise Exception(f"{_t} {e}") - - -# sync function -def fetch_soc(conf: VWEUDA, - vehicle: int, - vehicle_update_data: VehicleUpdateData) -> Union[float, float, float, str, float, str]: - - # prepare and call async method - loop = new_event_loop() - set_event_loop(loop) - - # get soc, range from server - a = euda() - soc, range, soc_ts, soc_tsX, odometer, warning =\ - loop.run_until_complete(a.get_status(conf, vehicle, vehicle_update_data)) - - return soc, range, soc_ts, soc_tsX, odometer, warning + log.exception("EUDA ev%s: Unerwarteter Fehler beim Abruf", self.vehicle) + if self._result is None: + raise ApiError(f"SOCERR-02: {error_message}") from error + else: + self.last_request_failed = False + + result = self._result + range_km = result.get("range") + if range_km is None: + capacity, consumption = number(update.battery_capacity), number(update.average_consump) + if capacity is not None and consumption is not None and consumption > 0: + range_km = result["soc"] * capacity / consumption + warning = result.get("warning") + if error_message: + message = f"EUDA-Abruf fehlgeschlagen; letzter gültiger Stand: {error_message}" + warning = " ".join(filter(None, [warning, message])) + return CarState(soc=result["soc"], range=range_km, soc_timestamp=result["soc_timestamp"], + odometer=result.get("odometer"), warning=warning) diff --git a/packages/modules/vehicles/vweuda/libeuda_test.py b/packages/modules/vehicles/vweuda/libeuda_test.py index ec39aa2c00..918d30fdb5 100644 --- a/packages/modules/vehicles/vweuda/libeuda_test.py +++ b/packages/modules/vehicles/vweuda/libeuda_test.py @@ -46,3 +46,69 @@ def test_parse_vehicle_data_with_bcam_active_but_no_threshold_returns_no_warning result = parse_vehicle_data({"Data": data}) assert result["warning"] is None + + +def test_caddy_uses_its_measurement_timestamp_instead_of_latest_other_field(): + data = BASE_DATA + [{ + "key": "ae0294b4-1286-3e98-a818-1485b8d88430", "dataFieldName": "state_of_charge", + "value": 0, "timestampUtc": "2026-09-15T08:00:00Z", + }] + result = parse_vehicle_data({"Data": data}) + assert result["soc"] == 0 + assert result["soc_timestamp_str"] == "2026-09-15T08:00:00Z" + assert result["soc_timestamp"] == 1789459200 + + +def test_fallback_fields_and_latest_capture_time_are_preserved(): + data = [ + {"key": "506cb83e-f99f-3af3-bbeb-0429b69a78d9", "dataFieldName": "soc", "value": "42"}, + {"key": "55e0d40b-38ed-3cb5-9dcd-6193df6fc493", "dataFieldName": "range", "value": 0}, + {"key": "30cc36fd-71ca-3c09-9296-e94ebd47bd2b", "dataFieldName": "odometer", "value": "1000"}, + {"key": "new", "dataFieldName": "car_captured_time", "value": "2026-09-15T09:00:00Z"}, + {"key": "old", "dataFieldName": "car_captured_time", "value": "2026-09-15T08:00:00Z"}, + ] + result = parse_vehicle_data({"Data": data}) + assert (result["soc"], result["range"], result["odometer"]) == ("42", 0, "1000") + assert result["soc_timestamp_str"] == "2026-09-15T09:00:00Z" + + +def test_primary_zero_soc_takes_precedence_over_fallback(): + data = [dict(field) for field in BASE_DATA] + data[1]["value"] = 0 + data.append({"key": "506cb83e-f99f-3af3-bbeb-0429b69a78d9", "dataFieldName": "soc", "value": "99"}) + assert parse_vehicle_data({"Data": data})["soc"] == 0 + + +def test_timestamp_fallback_uses_latest_value_without_car_captured_time(): + data = [BASE_DATA[1]] + [ + {"key": "latest", "dataFieldName": "timestamp", "value": "2026-09-15T09:00:00Z"}, + {"key": "earlier", "dataFieldName": "timestamp", "value": "2026-09-15T08:00:00Z"}, + ] + result = parse_vehicle_data({"Data": data}) + assert result["soc"] == "55.0" + assert result["soc_timestamp_str"] == "2026-09-15T09:00:00Z" + assert result["soc_timestamp"] == 1789462800 + + +def test_car_captured_time_has_priority_over_newer_timestamp_fallback(): + data = BASE_DATA + [ + {"key": "fallback", "dataFieldName": "timestamp", "value": "2026-09-15T10:00:00Z"}, + ] + result = parse_vehicle_data({"Data": data}) + assert result["soc_timestamp_str"] == "2026-09-15T09:00:00Z" + + +def test_caddy_timestamp_has_priority_over_timestamp_fallback(): + data = [ + {"key": "ae0294b4-1286-3e98-a818-1485b8d88430", "dataFieldName": "state_of_charge", + "value": 55, "timestampUtc": "2026-09-15T08:00:00Z"}, + {"key": "fallback", "dataFieldName": "timestamp", "value": "2026-09-15T10:00:00Z"}, + ] + result = parse_vehicle_data({"Data": data}) + assert result["soc_timestamp_str"] == "2026-09-15T08:00:00Z" + + +def test_missing_timestamp_fields_returns_no_timestamp(): + result = parse_vehicle_data({"Data": [BASE_DATA[1]]}) + assert result["soc_timestamp"] is None + assert result["soc_timestamp_str"] is None diff --git a/packages/modules/vehicles/vweuda/scheduler_test.py b/packages/modules/vehicles/vweuda/scheduler_test.py new file mode 100644 index 0000000000..f85f55f62b --- /dev/null +++ b/packages/modules/vehicles/vweuda/scheduler_test.py @@ -0,0 +1,171 @@ +import copy +from unittest.mock import MagicMock, Mock + +import pytest + +from modules.common.abstract_vehicle import VehicleUpdateData +from modules.common.configurable_vehicle import SocSource +from modules.vehicles.vweuda import libeuda, soc +from modules.vehicles.vweuda.config import VWEUDA, VWEUDAConfiguration + +VIN = "WVW00000000000001" +SOC_KEY = "ac1108b1-b8cc-3db9-a663-03d387e42223" + + +def payload(value=55, timestamp="2026-10-19T10:00:00Z", odometer=None): + return {"vin": VIN, "Data": [ + {"key": SOC_KEY, "dataFieldName": "soc", "value": value}, + {"key": "ts", "dataFieldName": "car_captured_time", "value": timestamp}, + {"key": "41c0805c-43e5-313e-9dfb-356cb8d20f7c", "dataFieldName": "odometer", "value": odometer}, + ]} + + +@pytest.fixture +def config(): + return VWEUDA(configuration=VWEUDAConfiguration(user_id="test@example.com", password="test", vin=VIN)) + + +def test_history_keeps_valid_data_and_processes_each_dataset_once(config): + portal = libeuda.EudaVehicle(config.configuration, 1) + client = Mock() + client.get_metadata.return_value = {"Identifier": "id"} + client.list_datasets.return_value = [ + {"name": "new.zip", "createdOn": "2026-10-19T12:00:00Z"}, + {"name": "empty_no_content_found.zip"}, + {"name": "old.zip", "createdOn": "2026-10-19T10:00:00Z"}, + {"name": "incomplete.zip", "createdOn": "2026-10-19T11:00:00Z"}, + ] + contents = {"old.zip": payload(odometer=1000), "incomplete.zip": payload(value=None), + "new.zip": payload(value=60, timestamp="2026-10-19T12:00:00Z")} + client.download_dataset.side_effect = lambda vin, identifier, name: (VIN + name + ".json", contents[name]) + portal._poll(client) + assert [call.args[2] for call in client.download_dataset.call_args_list] == [ + "old.zip", "incomplete.zip", "new.zip"] + assert portal._result["soc"] == 60 + assert portal._result["odometer"] == 1000 + portal._poll(client) + assert client.download_dataset.call_count == 3 + + +@pytest.mark.parametrize("value", [None, "bad", -1, 101, "NaN", "Infinity"]) +def test_invalid_data_never_advances_valid_timestamp(config, value): + portal = libeuda.EudaVehicle(config.configuration, 1) + portal._merge(libeuda.parse_vehicle_data(payload())) + before = dict(portal._result) + assert not portal._merge(libeuda.parse_vehicle_data(payload(value, "2026-10-19T12:00:00Z"))) + assert portal._result == before + + +def test_missing_timestamp_and_older_data_do_not_replace_cache(config): + portal = libeuda.EudaVehicle(config.configuration, 1) + portal._merge(libeuda.parse_vehicle_data(payload())) + assert not portal._merge(libeuda.parse_vehicle_data(payload(60, None))) + assert not portal._merge(libeuda.parse_vehicle_data(payload(60, "2026-10-19T09:00:00Z"))) + assert portal._result["soc"] == 55 + + +def test_zero_values_and_missing_optional_fields_are_preserved_in_memory(config): + portal = libeuda.EudaVehicle(config.configuration, 1) + first = libeuda.parse_vehicle_data(payload(odometer=1000)) + first["range"] = 200 + portal._merge(first) + newer = libeuda.parse_vehicle_data(payload(0, "2026-10-19T11:00:00Z", odometer=900)) + portal._merge(newer) + assert portal._result["range"] == 200 + newer["range"] = 0 + portal._merge(newer) + assert portal._result["soc"] == 0 + assert portal._result["range"] == 0 + assert portal._result["odometer"] == 1000 + + +@pytest.mark.parametrize("has_cache", [False, True]) +def test_scheduler_copies_share_cache_retry_flag_and_recovery(config, monkeypatch, has_cache): + config.configuration.calculate_soc = True + requests = [] + + def poll(portal, client): + requests.append(client) + if len(requests) == 1: + if has_cache: + portal._merge(libeuda.parse_vehicle_data(payload())) + raise libeuda.ApiError("Portal unavailable") + portal._merge(libeuda.parse_vehicle_data(payload(60, "2026-10-19T11:00:00Z"))) + + monkeypatch.setattr(libeuda.EudaVehicle, "_poll", poll) + session_factory = MagicMock() + monkeypatch.setattr(libeuda.req, "get_http_session", session_factory) + module = soc.create_vehicle(config, 1) + first = copy.deepcopy(module) + if has_cache: + state = first._get_carstate_by_source(VehicleUpdateData(), SocSource.API) + assert state.soc == 55 + assert state.soc_timestamp == 1792404000 + assert "Portal unavailable" in state.warning + else: + with pytest.raises(Exception, match="Portal unavailable"): + first._get_carstate_by_source(VehicleUpdateData(), SocSource.API) + assert module.get_request_interval_override() == 60 + second = copy.deepcopy(module) + assert second._get_carstate_source(VehicleUpdateData(charge_state=True)) == SocSource.API + state = second._get_carstate_by_source(VehicleUpdateData(), SocSource.API) + assert state.soc == 60 + assert state.warning is None + assert module.get_request_interval_override() is None + assert module._get_carstate_source(VehicleUpdateData(charge_state=True)) == SocSource.CALCULATION + assert session_factory.call_count == 2 + assert session_factory.return_value.__exit__.call_count == 2 + assert soc.create_vehicle(config, 1).get_request_interval_override() is None + + +def test_empty_portal_response_without_cache_is_error_and_requests_retry(config, monkeypatch): + client = Mock() + client.get_metadata.return_value = {"Identifier": "id"} + client.list_datasets.return_value = [] + monkeypatch.setattr(libeuda, "EudaApiClient", Mock(return_value=client)) + portal = libeuda.EudaVehicle(config.configuration, 1) + with pytest.raises(libeuda.ApiError, match="keine Datenpakete"): + portal.fetch(VehicleUpdateData()) + assert portal.get_request_interval_override() == 60 + + +def test_foreign_and_corrupt_datasets_do_not_block_later_valid_data(config): + client = Mock() + client.get_metadata.return_value = {"Identifier": "id"} + client.list_datasets.return_value = [{"name": name} for name in ("bad.zip", "foreign.zip", "good.zip")] + foreign = dict(payload(), vin="WVW00000000000002") + client.download_dataset.side_effect = [libeuda.DatasetError("bad archive"), + ("foreign.json", foreign), (VIN + ".json", payload())] + portal = libeuda.EudaVehicle(config.configuration, 1) + portal._poll(client) + assert portal._result["soc"] == 55 + + +def test_memory_cache_used_with_original_timestamp_and_calculated_range_on_failure(config, monkeypatch): + portal = libeuda.EudaVehicle(config.configuration, 1) + portal._merge(libeuda.parse_vehicle_data(payload())) + timestamp = portal._result["soc_timestamp"] + monkeypatch.setattr(libeuda.EudaVehicle, "_poll", Mock(side_effect=libeuda.ApiError("offline"))) + state = portal.fetch(VehicleUpdateData(battery_capacity=80, average_consump=20)) + assert state.soc == 55 + assert state.range == 220 + assert state.soc_timestamp == timestamp + assert "offline" in state.warning + assert portal.get_request_interval_override() == 60 + + # A fresh module starts without measurements and must fetch from the portal. + replacement = libeuda.EudaVehicle(config.configuration, 1) + with pytest.raises(libeuda.ApiError, match="offline"): + replacement.fetch(VehicleUpdateData()) + assert replacement.get_request_interval_override() == 60 + + +def test_unexpected_error_logs_traceback_and_keeps_cache_retry(config, monkeypatch, caplog): + portal = libeuda.EudaVehicle(config.configuration, 1) + portal._merge(libeuda.parse_vehicle_data(payload())) + monkeypatch.setattr(portal, "_poll", Mock(side_effect=RuntimeError("unexpected response shape"))) + state = portal.fetch(VehicleUpdateData()) + assert state.soc == 55 + assert "unexpected response shape" in state.warning + assert portal.get_request_interval_override() == 60 + assert any(record.exc_info and "Unerwarteter Fehler" in record.getMessage() for record in caplog.records) diff --git a/packages/modules/vehicles/vweuda/soc.py b/packages/modules/vehicles/vweuda/soc.py old mode 100755 new mode 100644 index fc4d905abc..5de0ecb88e --- a/packages/modules/vehicles/vweuda/soc.py +++ b/packages/modules/vehicles/vweuda/soc.py @@ -1,33 +1,25 @@ -# import aiohttp -import logging -# from asyncio import new_event_loop, set_event_loop -# from typing import Union - 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.vweuda.config import VWEUDA -from modules.vehicles.vweuda import libeuda -# from modules.vehicles.vwgroup.vwgroup import VwGroup - -log = logging.getLogger(__name__) +from modules.vehicles.vweuda.libeuda import EudaVehicle -def fetch(vehicle_update_data: VehicleUpdateData, config: VWEUDA, vehicle: int) -> CarState: - soc, range, soc_ts, soc_tsX, odometer, warning = libeuda.fetch_soc(config, vehicle, vehicle_update_data) - log.debug(f"soc return: soc={soc}, range={range}, soc_ts={soc_ts}, soc_tsX={soc_tsX}, odometer={odometer}") - return CarState(soc=soc, range=range, soc_timestamp=soc_ts, odometer=odometer, warning=warning) +def create_vehicle(vehicle_config: VWEUDA, vehicle: int) -> ConfigurableVehicle[VWEUDA]: + portal = EudaVehicle(vehicle_config.configuration, vehicle) - -def create_vehicle(vehicle_config: VWEUDA, vehicle: int): + # Functions are shared by deepcopy: both callbacks keep the same portal cache. def updater(vehicle_update_data: VehicleUpdateData) -> CarState: - return fetch(vehicle_update_data, vehicle_config, vehicle) - - return ConfigurableVehicle(vehicle_config=vehicle_config, - component_updater=updater, - vehicle=vehicle, - calc_while_charging=vehicle_config.configuration.calculate_soc) + return portal.fetch(vehicle_update_data) + + return ConfigurableVehicle( + vehicle_config=vehicle_config, + component_updater=updater, + vehicle=vehicle, + calc_while_charging=vehicle_config.configuration.calculate_soc, + request_interval_override=lambda: portal.get_request_interval_override(), + ) device_descriptor = DeviceDescriptor(configuration_factory=VWEUDA) diff --git a/packages/modules/vehicles/vweuda/vehicle_data.py b/packages/modules/vehicles/vweuda/vehicle_data.py new file mode 100644 index 0000000000..aa209761ef --- /dev/null +++ b/packages/modules/vehicles/vweuda/vehicle_data.py @@ -0,0 +1,75 @@ +"""Decode portal datasets and normalize vehicle measurements.""" + +import math +from datetime import datetime, timezone +from typing import Optional + +# Candidate order is significant: keep established fields ahead of fallbacks. +CADDY_SOC = "ae0294b4-1286-3e98-a818-1485b8d88430" +FIELD_KEYS = { + "soc": ( + CADDY_SOC, + "ac1108b1-b8cc-3db9-a663-03d387e42223", + "0a18a053-b4b0-3db1-be44-a6c5dba629b1", + "f89ed652-d104-3fa6-b7e2-ab7543309e7b", + "506cb83e-f99f-3af3-bbeb-0429b69a78d9", + ), + "range": ( + "153e8c40-4c6c-3c17-a11b-0ecc35d55b81", + "0ca40e18-0564-3eda-bcc0-7aee9ef44f04", + "55e0d40b-38ed-3cb5-9dcd-6193df6fc493", + ), + "odometer": ("41c0805c-43e5-313e-9dfb-356cb8d20f7c", "30cc36fd-71ca-3c09-9296-e94ebd47bd2b"), +} + + +def number(value) -> Optional[float]: + try: + value = float(value) + return value if math.isfinite(value) and value >= 0 else None + except (TypeError, ValueError): + return None + + +def parse_datetime(value: str) -> datetime: + value = datetime.fromisoformat(value.replace("Z", "+00:00")) + return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value + + +def dataset_time(entry: dict) -> datetime: + try: + return parse_datetime(entry["createdOn"]) + except (KeyError, ValueError, TypeError, AttributeError): + # Both TIMESTAMP_VIN.zip and VIN_TIMESTAMP.zip occur in the portal. + for part in reversed(entry["name"].rsplit(".", 1)[0].split("_")): + try: + return datetime.strptime(part, "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc) + except ValueError: + continue + return datetime.min.replace(tzinfo=timezone.utc) + + +def parse_vehicle_data(payload: dict) -> dict: + data = payload.get("Data", []) + by_key = {field["key"]: field for field in data} + by_name = {field["dataFieldName"]: field.get("value") for field in data} + + def first_value(keys): + return next((by_key[key]["value"] for key in keys + if key in by_key and by_key[key].get("value") is not None), None) + + result = {name: first_value(keys) for name, keys in FIELD_KEYS.items()} + timestamp = None + if by_key.get(CADDY_SOC, {}).get("value") is not None: + timestamp = by_key[CADDY_SOC].get("timestampUtc") + # Portal variants use different names; timestamp is only a fallback, + # even when it contains a later value than car_captured_time (PR #4127). + for name in ("car_captured_time", "timestamp"): + if not timestamp: + timestamp = max((field["value"] for field in data if field["dataFieldName"] == name), default=None) + result.update(soc_timestamp=parse_datetime(timestamp).timestamp() if timestamp else None, + soc_timestamp_str=timestamp, warning=None) + threshold = by_name.get("battery_care_mode.charge_bcam_threshold") + if by_name.get("setting.bcam_activation") == "BCAM_ACTIVATION_ACTIVATED" and threshold: + result["warning"] = f"Battery Care Mode ist im Fahrzeug aktiv und begrenzt die Ladung selbst auf {threshold}%." + return result