Vehicle: Add Pleos (Hyundai/Kia/Genesis) EU Data Act API

packages/helpermodules/logger.py

@@ -20,7 +20,7 @@
 
 KNOWN_SENSITIVE_FIELDS = [
     'password', 'secret', 'token', 'apikey', 'access_token',
-    'refresh_token', 'accesstoken', 'refreshtoken'
+    'refresh_token', 'accesstoken', 'refreshtoken', 'client_secret'
 ]
 REDACTION_PATTERNS = [
     (r'({field})[=:]([^\s&]+)', r'\1=***REDACTED***'),  # field=value, i.e. for URL query parameters

packages/helpermodules/logger_test.py

@@ -44,3 +44,10 @@ def test_redact_url_credentials_with_sensitive_field_as_user_name():
     # If the user name matches an entry of KNOWN_SENSITIVE_FIELDS, the field pattern applies on
     # top and truncates the url. The password is removed in that case as well.
     assert redact_sensitive_info("ws://token:[email protected]/v16/") == "ws://token=***REDACTED***"
+
+
+def test_redact_client_secret():
+    message = '{"client_id": "abc", "client_secret": "xyz"}'
+    expected = '{"client_id": "abc", "client_secret": "***REDACTED***"}'
+
+    assert redact_sensitive_info(message) == expected

packages/modules/vehicles/pleos/api.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+from datetime import datetime
+import logging
+import time
+from typing import Optional
+
+from modules.common import req
+from modules.vehicles.pleos.config import PleosToken
+
+log = logging.getLogger(__name__)
+
+BASE_URL = "https://api.pleos.ai/v1"
+
+_UNIT_TO_KM = {"km": 1, "mile": 1.609344, "miles": 1.609344, "meter": 0.001, "feet": 0.0003048}
+
+
+def validate_token(brand: str, client_id: str, client_secret: str, token: PleosToken) -> PleosToken:
+    if token.access_token is not None and time.time() < token.created_at + token.expires_in:
+        return token
+    return request_token(brand, client_id, client_secret)
+
+
+def request_token(brand: str, client_id: str, client_secret: str) -> PleosToken:
+    response = req.get_http_session().post(
+        BASE_URL + "/auth/personal-token",
+        json={"clientId": client_id, "clientSecret": client_secret},
+        headers={"Brand": brand},
+        timeout=10,
+    )
+    data = response.json().get("data", {})
+    access_token = data.get("accessToken")
+    if not access_token:
+        raise Exception("Pleos: Token-Antwort enthält keinen accessToken.")
+    return PleosToken(access_token=access_token, expires_in=data.get("expiresIn") or 3600,
+                      created_at=int(time.time()))
+
+
+def _auth_headers(brand: str, token: PleosToken) -> dict:
+    return {"Brand": brand, "Authorization": "Bearer " + token.access_token}
+
+
+def get_battery(brand: str, vin: str, token: PleosToken) -> dict:
+    response = req.get_http_session().get(
+        BASE_URL + f"/vehicles/{vin}/batteries", headers=_auth_headers(brand, token), timeout=10)
+    return response.json().get("data", {})
+
+
+def get_distance_to_empty_km(brand: str, vin: str, token: PleosToken) -> Optional[float]:
+    response = req.get_http_session().get(
+        BASE_URL + f"/vehicles/{vin}/powertrains", headers=_auth_headers(brand, token), timeout=10)
+    for entry in response.json().get("data", {}).get("distanceToEmpties", []):
+        if entry.get("type") == "EV":
+            return _to_km(entry.get("value"), entry.get("unit"))
+    return None
+
+
+def get_odometer_km(brand: str, vin: str, token: PleosToken) -> Optional[float]:
+    response = req.get_http_session().get(
+        BASE_URL + f"/vehicles/{vin}/driving", headers=_auth_headers(brand, token), timeout=10)
+    odometer = response.json().get("data", {}).get("odometer", {})
+    return _to_km(odometer.get("value"), odometer.get("unit"))
+
+
+def _to_km(value, unit: str) -> Optional[float]:
+    try:
+        value = float(value)
+    except (TypeError, ValueError):
+        return None
+    return value * _UNIT_TO_KM.get(unit or "km", 1)
+
+
+def parse_timestamp(value: Optional[str]) -> Optional[float]:
+    # Pleos liefert idR den letzten vom Fahrzeug gemeldeten Stand, nicht live -
+    # ohne diesen Zeitstempel würde CarState "jetzt" annehmen und die
+    # Veraltungs-/Reihenfolge-Prüfung in ConfigurableVehicle wirkungslos machen.
+    if not value:
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
+    except ValueError:
+        return None

packages/modules/vehicles/pleos/api_test.py

@@ -0,0 +1,90 @@
+import time
+from unittest.mock import Mock
+
+import pytest
+
+from modules.vehicles.pleos import api
+from modules.vehicles.pleos.config import PleosToken
+
+
+def _session_returning(payload):
+    session = Mock()
+    response = Mock()
+    response.json.return_value = payload
+    session.get.return_value = response
+    session.post.return_value = response
+    return session
+
+
+class TestValidateToken:
+    def test_reuses_valid_token(self, monkeypatch):
+        session = _session_returning({})
+        monkeypatch.setattr(api.req, "get_http_session", lambda: session)
+        token = PleosToken(access_token="cached", expires_in=3600, created_at=int(time.time()))
+
+        result = api.validate_token("kia", "cid", "csec", token)
+
+        assert result.access_token == "cached"
+        session.post.assert_not_called()
+
+    def test_requests_new_token_when_expired(self, monkeypatch):
+        session = _session_returning({"data": {"accessToken": "fresh", "expiresIn": 3600}})
+        monkeypatch.setattr(api.req, "get_http_session", lambda: session)
+        token = PleosToken(access_token="old", expires_in=10, created_at=int(time.time()) - 100)
+
+        result = api.validate_token("kia", "cid", "csec", token)
+
+        assert result.access_token == "fresh"
+        session.post.assert_called_once()
+
+    def test_raises_when_no_access_token_in_response(self, monkeypatch):
+        session = _session_returning({"data": {}})
+        monkeypatch.setattr(api.req, "get_http_session", lambda: session)
+
+        with pytest.raises(Exception):
+            api.validate_token("kia", "cid", "csec", PleosToken())
+
+
+class TestGetBattery:
+    def test_returns_charge_data(self, monkeypatch):
+        payload = {"data": {"charge": {"stateOfCharge": 72.5}}}
+        session = _session_returning(payload)
+        monkeypatch.setattr(api.req, "get_http_session", lambda: session)
+
+        result = api.get_battery("kia", "VIN1", PleosToken(access_token="t"))
+
+        assert result["charge"]["stateOfCharge"] == 72.5
+
+
+class TestGetDistanceToEmptyKm:
+    def test_converts_miles_to_km(self, monkeypatch):
+        payload = {"data": {"distanceToEmpties": [{"type": "EV", "value": "100", "unit": "mile"}]}}
+        session = _session_returning(payload)
+        monkeypatch.setattr(api.req, "get_http_session", lambda: session)
+
+        result = api.get_distance_to_empty_km("kia", "VIN1", PleosToken(access_token="t"))
+
+        assert round(result, 2) == 160.93
+
+    def test_ignores_non_ev_entries(self, monkeypatch):
+        payload = {"data": {"distanceToEmpties": [{"type": "ICE", "value": "500", "unit": "km"}]}}
+        session = _session_returning(payload)
+        monkeypatch.setattr(api.req, "get_http_session", lambda: session)
+
+        assert api.get_distance_to_empty_km("kia", "VIN1", PleosToken(access_token="t")) is None
+
+    def test_returns_none_on_invalid_value(self, monkeypatch):
+        payload = {"data": {"distanceToEmpties": [{"type": "EV", "value": "invalid", "unit": "km"}]}}
+        session = _session_returning(payload)
+        monkeypatch.setattr(api.req, "get_http_session", lambda: session)
+
+        assert api.get_distance_to_empty_km("kia", "VIN1", PleosToken(access_token="t")) is None
+
+
+class TestGetOdometerKm:
+    def test_returns_value_in_km(self, monkeypatch):
+        payload = {"data": {"odometer": {"value": "15234", "unit": "km"}}}
+        session = _session_returning(payload)
+        monkeypatch.setattr(api.req, "get_http_session", lambda: session)
+
+        assert api.get_odometer_km("kia", "VIN1", PleosToken(access_token="t")) == 15234.0

packages/modules/vehicles/pleos/config.py

@@ -0,0 +1,37 @@
+from typing import Optional
+
+
+class PleosToken:
+    def __init__(self,
+                 access_token: Optional[str] = None,
+                 expires_in: int = 0,
+                 created_at: int = 0) -> None:
+        self.access_token = access_token
+        self.expires_in = expires_in
+        self.created_at = created_at
+
+
+class PleosConfiguration:
+    def __init__(self,
+                 brand: Optional[str] = None,
+                 client_id: Optional[str] = None,
+                 client_secret: Optional[str] = None,
+                 vin: Optional[str] = None,
+                 token: Optional[PleosToken] = None) -> None:
+        self.brand = brand
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.vin = vin
+        self.token = token or PleosToken()
+
+
+class Pleos:
+    def __init__(self,
+                 name: str = "Pleos (Hyundai/Kia/Genesis)",
+                 type: str = "pleos",
+                 official: bool = False,
+                 configuration: Optional[PleosConfiguration] = None) -> None:
+        self.name = name
+        self.type = type
+        self.official = official
+        self.configuration = configuration or PleosConfiguration()

packages/modules/vehicles/pleos/soc.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+import logging
+
+import requests
+
+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.pleos import api
+from modules.vehicles.pleos.config import Pleos
+
+log = logging.getLogger(__name__)
+
+
+def _get_battery_with_token_refresh(config) -> dict:
+    try:
+        return api.get_battery(config.brand, config.vin, config.token)
+    except requests.HTTPError as e:
+        if e.response is None or e.response.status_code != 401:
+            raise
+        config.token = api.request_token(config.brand, config.client_id, config.client_secret)
+        return api.get_battery(config.brand, config.vin, config.token)
+
+
+def fetch(vehicle_config: Pleos, vehicle_update_data: VehicleUpdateData) -> CarState:
+    config = vehicle_config.configuration
+    if not config.vin:
+        raise Exception("Pleos: VIN muss konfiguriert sein.")
+    config.token = api.validate_token(config.brand, config.client_id, config.client_secret, config.token)
+    battery = _get_battery_with_token_refresh(config)
+    charge = battery.get('charge', {})
+    soc = float(charge['stateOfCharge'])
+    soc_timestamp = api.parse_timestamp(battery.get('timestamp'))
+    if soc_timestamp is None:
+        raise Exception("Pleos: Zeitstempel des Ladezustands fehlt oder ist ungültig.")
+    target_soc = (charge.get('targetStateOfCharge') or {}).get('standard')
+    warning = (f"Das Fahrzeug begrenzt die Ladung selbst auf {target_soc}%."
+               if target_soc and target_soc < 100 else None)
+    try:
+        range = api.get_distance_to_empty_km(config.brand, config.vin, config.token)
+    except Exception as e:
+        log.debug(f"Pleos: Reichweite nicht verfügbar: {e}")
+        range = None
+    try:
+        odometer = api.get_odometer_km(config.brand, config.vin, config.token)
+    except Exception as e:
+        log.debug(f"Pleos: Kilometerstand nicht verfügbar: {e}")
+        odometer = None
+    return CarState(soc=soc, range=range, soc_timestamp=soc_timestamp, odometer=odometer, warning=warning)
+
+
+def create_vehicle(vehicle_config: Pleos, vehicle: int):
+    def updater(vehicle_update_data: VehicleUpdateData) -> CarState:
+        return fetch(vehicle_config, vehicle_update_data)
+    return ConfigurableVehicle(vehicle_config=vehicle_config,
+                               component_updater=updater,
+                               vehicle=vehicle)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=Pleos)