From 82d9ed18259d45f5e2e3ae5713c4a494d71b2271 Mon Sep 17 00:00:00 2001 From: tietjen Date: Wed, 11 Mar 2026 13:41:51 +0100 Subject: [PATCH] Support Porsche mobile auth flow and Macan EV data --- pyporscheconnectapi/account.py | 34 +++++- pyporscheconnectapi/cli.py | 8 +- pyporscheconnectapi/connection.py | 56 ++++++++- pyporscheconnectapi/const.py | 9 +- pyporscheconnectapi/exceptions.py | 7 +- pyporscheconnectapi/oauth2.py | 188 ++++++++++++++++++++++++++++-- pyporscheconnectapi/vehicle.py | 121 +++++++++++++++---- 7 files changed, 374 insertions(+), 49 deletions(-) diff --git a/pyporscheconnectapi/account.py b/pyporscheconnectapi/account.py index ee43e7c..4503c5e 100644 --- a/pyporscheconnectapi/account.py +++ b/pyporscheconnectapi/account.py @@ -4,12 +4,40 @@ import logging -from pyporscheconnectapi.connection import Connection -from pyporscheconnectapi.vehicle import PorscheVehicle +from .connection import Connection +from .vehicle import PorscheVehicle _LOGGER = logging.getLogger(__name__) +def _normalize_engine(vehicle: dict) -> str: + """Best-effort mapping of the portal vehicle payload to drivetrain type.""" + model_type = vehicle.get("modelType", {}) + if model_type.get("engine"): + return model_type["engine"] + description = str(vehicle.get("modelDescription", "")).lower() + if description in {"macan", "taycan"}: + return "BEV" + return "COMBUSTION" + + +def _normalize_vehicle(vehicle: dict) -> dict: + """Normalize the portal vehicle payload to the legacy library shape.""" + model_name = vehicle.get("modelDescription") or vehicle.get("modelName") or vehicle.get("vin", "Porsche") + return { + "vin": vehicle["vin"], + "name": model_name, + "modelName": model_name, + "modelType": { + "year": vehicle.get("modelYear") or vehicle.get("modelType", {}).get("year", "not available"), + "engine": _normalize_engine(vehicle), + }, + "systemInfo": vehicle.get("systemInfo", {}), + "timestamp": vehicle.get("validFrom") or vehicle.get("timestamp"), + "portalVehicle": vehicle, + } + + class PorscheConnectAccount: """Establishes a connection to a Porsche Connect account.""" @@ -39,7 +67,7 @@ async def _init_vehicles(self) -> None: _LOGGER.debug("Got vehicle %s", vehicle) v = PorscheVehicle( vin=vehicle["vin"], - data=vehicle, + data=_normalize_vehicle(vehicle), status={}, connection=self.connection, ) diff --git a/pyporscheconnectapi/cli.py b/pyporscheconnectapi/cli.py index 8bf4ce9..f3b9173 100755 --- a/pyporscheconnectapi/cli.py +++ b/pyporscheconnectapi/cli.py @@ -13,10 +13,10 @@ import aiofiles -from pyporscheconnectapi.account import PorscheConnectAccount -from pyporscheconnectapi.connection import Connection -from pyporscheconnectapi.exceptions import PorscheCaptchaRequiredError, PorscheWrongCredentialsError -from pyporscheconnectapi.remote_services import RemoteServices +from .account import PorscheConnectAccount +from .connection import Connection +from .exceptions import PorscheCaptchaRequiredError, PorscheWrongCredentialsError +from .remote_services import RemoteServices vehicle_commands = { "battery": "Prints the main battery level (BEV)", diff --git a/pyporscheconnectapi/connection.py b/pyporscheconnectapi/connection.py index e2a14c3..17375c9 100644 --- a/pyporscheconnectapi/connection.py +++ b/pyporscheconnectapi/connection.py @@ -5,10 +5,11 @@ import asyncio import logging +import uuid import httpx -from .const import API_BASE_URL, TIMEOUT, USER_AGENT, X_CLIENT_ID +from .const import API_BASE_URL, DCGW_BASE_URL, TIMEOUT, USER_AGENT, X_CLIENT_ID from .exceptions import PorscheExceptionError from .oauth2 import Captcha, Credentials, OAuth2Client, OAuth2Token @@ -39,6 +40,8 @@ def __init__( captcha_code: str | None = None, state: str | None = None, async_client=httpx.AsyncClient(), + cookies: list[dict] | None = None, + code_verifier: str | None = None, token=None, leeway: int = 60, ) -> None: @@ -47,15 +50,33 @@ def __init__( token = {} self.asyncClient = async_client self.token_lock = asyncio.Lock() + self.country_code = "de" + self.language_code = "de_DE" + + if cookies: + for cookie in cookies: + self.asyncClient.cookies.set( + cookie["name"], + cookie["value"], + domain=cookie.get("domain"), + path=cookie.get("path", "/"), + ) self.token = OAuth2Token(token) - self.headers = {"User-Agent": USER_AGENT, "X-Client-ID": X_CLIENT_ID} + self.headers = { + "User-Agent": USER_AGENT, + "X-Client-ID": X_CLIENT_ID, + "Accept-Language": "de-DE", + "Origin": "https://security.porsche.com", + "Referer": "https://security.porsche.com/", + } self.oauth2_client = OAuth2Client( self.asyncClient, Credentials(email, password), Captcha(captcha_code, state), + code_verifier, leeway, ) @@ -69,6 +90,18 @@ async def get(self, url, params=None): """Make a GET request to the Porsche Connect API.""" return await self.request("GET", url, params=params) + async def portal_get(self, url, params=None): + """Make a GET request to the Porsche DCGW portal API.""" + return await self.absolute_request("GET", f"{DCGW_BASE_URL}{url}", params=params) + + async def get_portal_config(self): + """Fetch and cache portal localization/config data.""" + config = await self.portal_get(f"/core/config/v1/{self.country_code}/{self.country_code}/") + localization = config.get("localization", {}) + self.country_code = str(localization.get("countryCode", self.country_code)).lower() + self.language_code = localization.get("languageCode", self.language_code) + return config + async def post(self, url, data=None, json=None): """Make a POST request to the Porsche Connect API.""" return await self.request("POST", url, data=data, json=json) @@ -83,20 +116,33 @@ async def delete(self, url, data=None, json=None): async def request(self, method, url, **kwargs): """Create a request to the Porsche Connect API.""" + return await self.absolute_request(method, f"{API_BASE_URL}{url}", **kwargs) + + async def absolute_request(self, method, url, **kwargs): + """Create a request to an absolute Porsche API URL.""" try: async with self.token_lock: await self.oauth2_client.ensure_valid_token(self.token) + headers = self.headers | { + "Authorization": f"Bearer {self.token.access_token}", + "X-TRACE-ID": f"PCCK-PORTAL-{uuid.uuid4()}", + } resp = await self.asyncClient.request( method, - f"{API_BASE_URL}{url}", - headers=self.headers | {"Authorization": f"Bearer {self.token.access_token}"}, + url, + headers=headers, timeout=TIMEOUT, **kwargs, ) resp.raise_for_status() # A common error seem to be: httpx.HTTPStatusError: Server error '504 Gateway Time-out' return resp.json() except httpx.HTTPStatusError as exc: - raise PorscheExceptionError(exc.response.status_code) from exc + response_text = exc.response.text[:1000] if exc.response.text else None + raise PorscheExceptionError( + exc.response.status_code, + response_body=response_text, + request_url=str(exc.request.url), + ) from exc async def close(self): """Close the asyncClient connection.""" diff --git a/pyporscheconnectapi/const.py b/pyporscheconnectapi/const.py index 3e2af41..69aa7f1 100644 --- a/pyporscheconnectapi/const.py +++ b/pyporscheconnectapi/const.py @@ -1,12 +1,13 @@ """Client configuration constants.""" AUTHORIZATION_SERVER = "identity.porsche.com" -REDIRECT_URI = "my-porsche-app://auth0/callback" +REDIRECT_URI = "https://security.porsche.com/auth/en-GB/app/callback" AUDIENCE = "https://api.porsche.com" -CLIENT_ID = "XhygisuebbrqQ80byOuU5VncxLIm8E6H" -X_CLIENT_ID = "41843fb4-691d-4970-85c7-2673e8ecef40" -USER_AGENT = "pyporscheconnectapi/0.2.0" +CLIENT_ID = "qIkoJqlAXvbj4R3j12ct3zdinPId0Zbl" +X_CLIENT_ID = "09fcb5d8-d4ad-48e8-a0e8-a9c7cb1b9cbc" +USER_AGENT = "de.porsche.one/18.26.09-row+162630 (android)" API_BASE_URL = "https://api.ppa.porsche.com/app" +DCGW_BASE_URL = "https://dgw.p-fra.portal.aws.porsche.cloud" AUTHORIZATION_URL = f"https://{AUTHORIZATION_SERVER}/authorize" TOKEN_URL = f"https://{AUTHORIZATION_SERVER}/oauth/token" TIMEOUT = 90 diff --git a/pyporscheconnectapi/exceptions.py b/pyporscheconnectapi/exceptions.py index ada14b0..46d55d3 100644 --- a/pyporscheconnectapi/exceptions.py +++ b/pyporscheconnectapi/exceptions.py @@ -9,9 +9,11 @@ class PorscheExceptionError(Exception): """Class of Porsche API exceptions.""" - def __init__(self, code=None, *args, **kwargs) -> None: + def __init__(self, code=None, response_body=None, request_url=None, *args, **kwargs) -> None: """Initialize exceptions for the Porsche API.""" self.message = "" + self.response_body = response_body + self.request_url = request_url super().__init__(*args, **kwargs) if code is not None: self.code = code @@ -41,6 +43,9 @@ def __init__(self, code=None, *args, **kwargs) -> None: elif self.code > 299: self.message = f"UNKNOWN_ERROR_{self.code}" + if self.response_body: + self.message = f"{self.message}: {self.response_body}" + class PorscheWrongCredentialsError(PorscheExceptionError): """Class of exceptions for incomplete credentials.""" diff --git a/pyporscheconnectapi/oauth2.py b/pyporscheconnectapi/oauth2.py index 4e1d67a..f790ac8 100644 --- a/pyporscheconnectapi/oauth2.py +++ b/pyporscheconnectapi/oauth2.py @@ -2,10 +2,15 @@ # SPDX-License-Identifier: Apache-2.0 import asyncio +import base64 +import hashlib +import json import logging +import os +import re import time from typing import NamedTuple -from urllib.parse import parse_qs, urlparse +from urllib.parse import parse_qs, urljoin, urlparse import httpx from bs4 import BeautifulSoup @@ -102,6 +107,7 @@ def __init__( client: httpx.AsyncClient, credentials: Credentials, captcha: Captcha, + code_verifier: str | None = None, leeway: int = 60, ): """Initialise the oauth2 client.""" @@ -109,8 +115,49 @@ def __init__( self.credentials = credentials self.captcha = captcha self.leeway = leeway + self.code_verifier = code_verifier self.headers = {"User-Agent": USER_AGENT, "X-Client-ID": X_CLIENT_ID} + def _generate_pkce_verifier(self) -> str: + """Generate an Auth0-compatible PKCE verifier.""" + return base64.urlsafe_b64encode(os.urandom(64)).rstrip(b"=").decode("ascii") + + def _build_pkce_challenge(self, verifier: str) -> str: + """Generate a S256 code challenge from the verifier.""" + digest = hashlib.sha256(verifier.encode("ascii")).digest() + return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") + + def _extract_universal_login_context(self, html: str) -> dict | None: + """Extract Auth0 universal login context from inline base64 JSON.""" + match = re.search(r'atob\("([A-Za-z0-9+/=]+)"\)', html) + if not match: + return None + + try: + payload = base64.b64decode(match.group(1)) + return json.loads(payload.decode("utf-8")) + except (ValueError, json.JSONDecodeError): + return None + + def _extract_captcha_from_login_html(self, html: str) -> str | None: + """Extract captcha image data from Porsche/Auth0 login HTML. + + Older pages exposed the captcha as a regular captcha. The + current Porsche login page embeds the full login context as a base64 + payload in `window.universal_login_context`, including + `screen.captcha.image`. + """ + soup = BeautifulSoup(html, "html.parser") + captcha_img = soup.find("img", {"alt": "captcha"}) + if captcha_img and captcha_img.get("src"): + return captcha_img["src"] + + context = self._extract_universal_login_context(html) + if context is None: + return None + + return context.get("screen", {}).get("captcha", {}).get("image") + async def ensure_valid_token(self, token: OAuth2Token): """Ensure the access_token is valid, logging in or refreshing if necessary.""" token_is_expired = token.is_expired(self.leeway) @@ -144,6 +191,7 @@ async def fetch_authorization_code(self): _LOGGER.debug("Fetching authorization code.") # first request to get the code + self.code_verifier = self._generate_pkce_verifier() params = await self.get_and_extract_location_params( AUTHORIZATION_URL, params={ @@ -152,6 +200,9 @@ async def fetch_authorization_code(self): "redirect_uri": REDIRECT_URI, "audience": AUDIENCE, "scope": SCOPE, + "response_mode": "query", + "code_challenge": self._build_pkce_challenge(self.code_verifier), + "code_challenge_method": "S256", "state": "pyporscheconnectapi", }, ) @@ -170,10 +221,9 @@ async def fetch_authorization_code(self): resume_path = await self.login_with_identifier(params["state"][0]) # completed the Identifier First flow, now resume the auth code request - params = await self.get_and_extract_location_params( - f"https://{AUTHORIZATION_SERVER}{resume_path}", + authorization_code = await self.resume_authorization_code_flow( + self._resolve_resume_url(resume_path), ) - authorization_code = params.get("code", [None])[0] except httpx.HTTPStatusError as exc: raise PorscheExceptionError(exc.response.status_code) from exc @@ -184,11 +234,12 @@ async def fetch_authorization_code(self): else: try: + if self.code_verifier is None: + self.code_verifier = self._generate_pkce_verifier() resume_path = await self.login_with_identifier(self.captcha.state) - params = await self.get_and_extract_location_params( - f"https://{AUTHORIZATION_SERVER}{resume_path}", + authorization_code = await self.resume_authorization_code_flow( + self._resolve_resume_url(resume_path), ) - authorization_code = params.get("code", [None])[0] except httpx.HTTPStatusError as exc: raise PorscheExceptionError(exc.response.status_code) from exc @@ -235,6 +286,120 @@ def _merge_query_params(self, url: str, params: dict[str, str]) -> dict[str, str new_query.update(params) return new_query + def _resolve_resume_url(self, resume_path: str) -> str: + """Resolve resume targets returned by Porsche/Auth0.""" + if resume_path.startswith("http://") or resume_path.startswith("https://"): + return resume_path + return f"https://{AUTHORIZATION_SERVER}{resume_path}" + + async def _skip_passkey_enrollment(self, url: str, html: str | None = None) -> str: + """Skip optional Auth0/Porsche passkey enrollment.""" + if html is None: + resp = await self.client.get( + url, + timeout=TIMEOUT, + headers=self.headers, + follow_redirects=False, + ) + resp.raise_for_status() + html = resp.text + + context = self._extract_universal_login_context(html) + if context is None: + msg = "PASSKEY_ENROLLMENT_CONTEXT_MISSING" + raise PorscheExceptionError(msg) + + transaction_state = context.get("transaction", {}).get("state") + if not transaction_state: + msg = "PASSKEY_ENROLLMENT_STATE_MISSING" + raise PorscheExceptionError(msg) + + submitted_form_data = context.get("untrustedData", {}).get("submittedFormData") or {} + data = dict(submitted_form_data) + data.update( + { + "state": transaction_state, + "action": "abort-passkey-enrollment", + "acul-sdk": "@auth0/auth0-acul-js@1.2.0", + }, + ) + + resp = await self.client.post( + url, + data=data, + timeout=TIMEOUT, + headers=self.headers, + follow_redirects=False, + ) + if resp.status_code not in (302, 303): + msg = "PASSKEY_ENROLLMENT_SKIP_FAILED" + raise PorscheExceptionError(msg) + + return urljoin(url, resp.headers["Location"]) + + async def resume_authorization_code_flow(self, url: str) -> str: + """Follow Auth0 redirects until the final authorization code is returned.""" + current_url = url + + for _ in range(10): + parsed = urlparse(current_url) + code = parse_qs(parsed.query).get("code", [None])[0] + if code is not None: + return code + + resp = await self.client.get( + current_url, + timeout=TIMEOUT, + headers=self.headers, + follow_redirects=False, + ) + + if resp.status_code in (302, 303, 307, 308): + location = urljoin(str(resp.url), resp.headers["Location"]) + parsed = urlparse(location) + code = parse_qs(parsed.query).get("code", [None])[0] + if code is not None: + return code + if "/u/passkey-enrollment" in parsed.path: + current_url = await self._skip_passkey_enrollment(location) + continue + current_url = location + continue + + if resp.status_code == 200 and "/u/passkey-enrollment" in resp.url.path: + current_url = await self._skip_passkey_enrollment(str(resp.url), resp.text) + continue + + # Porsche may bounce back to the SPA root after login. At this point the + # Auth0 session is established, so a fresh /authorize request should now + # yield the final authorization code. + if resp.status_code == 200 and resp.url.host == "my.porsche.com": + params = await self.get_and_extract_location_params( + AUTHORIZATION_URL, + params={ + "response_type": "code", + "client_id": CLIENT_ID, + "redirect_uri": REDIRECT_URI, + "audience": AUDIENCE, + "scope": SCOPE, + "response_mode": "query", + "code_challenge": self._build_pkce_challenge(self.code_verifier), + "code_challenge_method": "S256", + "state": "pyporscheconnectapi", + }, + ) + code = params.get("code", [None])[0] + if code is not None: + return code + msg = "AUTHORIZATION_CODE_MISSING_AFTER_PORTAL_RESUME" + raise PorscheExceptionError(msg) + + msg = "Could not fetch authorization code" + raise PorscheExceptionError(msg) + + msg = "AUTHORIZATION_CODE_REDIRECT_LOOP" + raise PorscheExceptionError(msg) + async def login_with_identifier(self, state: str): """Log into the Identifier First flow. @@ -283,8 +448,10 @@ async def login_with_identifier(self, state: str): # In case captcha verification is required, the response code is 400 and the captcha is provided as a svg image if resp.status_code == 400: _LOGGER.debug("Captcha required.") - soup = BeautifulSoup(resp.text, "html.parser") - captcha_img = soup.find("img", {"alt": "captcha"})["src"] + captcha_img = self._extract_captcha_from_login_html(resp.text) + if captcha_img is None: + msg = "CAPTCHA_REQUIRED_BUT_NOT_PARSEABLE" + raise PorscheExceptionError(msg) _LOGGER.debug("Parsed out SVG captcha: %s", captcha_img) raise PorscheCaptchaRequiredError(captcha=captcha_img, state=state) @@ -334,6 +501,8 @@ async def fetch_access_token(self, authorization_code): "code": authorization_code, "redirect_uri": REDIRECT_URI, } + if self.code_verifier is not None: + data["code_verifier"] = self.code_verifier try: _LOGGER.debug("Exchanging the authorization code for an access token.") @@ -347,6 +516,7 @@ async def fetch_access_token(self, authorization_code): resp.raise_for_status() return resp.json() except httpx.HTTPStatusError as exc: + _LOGGER.debug("Token exchange failed: %s", exc.response.text) raise PorscheExceptionError(exc.response.status_code) from exc async def refresh_token(self, refresh_token): diff --git a/pyporscheconnectapi/vehicle.py b/pyporscheconnectapi/vehicle.py index 94760cc..7d8499d 100644 --- a/pyporscheconnectapi/vehicle.py +++ b/pyporscheconnectapi/vehicle.py @@ -8,9 +8,9 @@ import re import uuid -from pyporscheconnectapi.connection import Connection -from pyporscheconnectapi.exceptions import PorscheExceptionError -from pyporscheconnectapi.remote_services import RemoteServices +from .connection import Connection +from .exceptions import PorscheExceptionError +from .remote_services import RemoteServices from .const import COMMANDS, MEASUREMENTS, TIRE_PRESSURE_TOLERANCE, TRIP_STATISTICS @@ -19,6 +19,83 @@ BASE_DATA = ["vin", "modelName", "modelType", "systemInfo", "timestamp"] +def _normalize_engine(vehicle: dict) -> str: + """Best-effort mapping of the portal vehicle payload to drivetrain type.""" + description = str(vehicle.get("modelDescription", "")).lower() + if description in {"macan", "taycan"}: + return "BEV" + return "COMBUSTION" + + +def _normalize_portal_vehicle(vehicle: dict) -> dict: + """Normalize the portal vehicle payload to the legacy library shape.""" + model_name = vehicle.get("modelDescription") or vehicle.get("modelName") or vehicle.get("vin", "Porsche") + return { + "vin": vehicle["vin"], + "name": model_name, + "modelName": model_name, + "modelType": { + "year": vehicle.get("modelYear", "not available"), + "engine": _normalize_engine(vehicle), + }, + "systemInfo": {}, + "timestamp": vehicle.get("validFrom"), + "portalVehicle": vehicle, + } + + +def _remote_access_enabled(connect_capability: dict) -> bool | None: + """Translate connect capability remote access state to the legacy flag shape.""" + remote_access = connect_capability.get("remoteAccess", {}) + if not remote_access: + return None + return remote_access.get("supported") is True and remote_access.get("status") == "ACTIVE" and remote_access.get("userIsActive") is True + + +def _privacy_mode_enabled(services: dict) -> bool | None: + """Translate service disabled reasons to the legacy privacy mode flag.""" + items = services.get("services") + if items is None: + return None + return any(service.get("disabledReason") == "PRIVACY_MODE" for service in items) + + +def _connectivity_state(connectivity: dict) -> dict | None: + """Normalize connectivity data into a legacy-friendly nested node.""" + if not connectivity: + return None + return { + "supported": connectivity.get("supported"), + "status": connectivity.get("status"), + "connectivityStatus": connectivity.get("connectivityStatus"), + "provider": connectivity.get("provider"), + } + + +def _pairing_state(pairing: dict, connect_capability: dict) -> dict | None: + """Normalize pairing state from available core endpoints.""" + login = connect_capability.get("login", {}) if connect_capability else {} + if not pairing and not login: + return None + return { + "status": pairing.get("status") or login.get("pairingStatus"), + "pairingCode": pairing.get("pairingCode") or login.get("pairingCode"), + "canSendPairingCode": pairing.get("canSendPairingCode", login.get("canSendPairingCode")), + "method": login.get("method"), + "porscheId": login.get("porscheId"), + } + + +def _permissions_state(permissions: dict) -> dict | None: + """Normalize permissions data into a stable node.""" + if not permissions: + return None + return { + "userIsActive": permissions.get("userIsActive"), + "userRoleStatus": permissions.get("userRoleStatus"), + } + + class PorscheVehicle: """Representation of a Porsche Connect vehicle.""" @@ -203,14 +280,14 @@ def location(self) -> tuple[float | None, float | None, int | None]: async def get_stored_overview(self) -> None: """Return stored vechicle status overview.""" - measurements = "mf=" + "&mf=".join(MEASUREMENTS) - try: _LOGGER.debug("Getting stored status for vehicle %s", self.vin) - self.status = await self.connection.get( - f"/connect/v1/vehicles/{self.vin}?{measurements}", - ) - self._update_vehicle_data() + overview = await self.connection.get(f"/connect/v1/vehicles/{self.vin}") + self.status = { + "appVehicle": overview, + } + normalized = _normalize_portal_vehicle(overview) + self.data = self.data | normalized except PorscheExceptionError as err: _LOGGER.exception( "Could not get stored overview, error communicating with API: '%s", @@ -267,18 +344,7 @@ async def get_trip_statistics(self) -> None: async def get_picture_locations(self) -> None: """Return list of uri's to vechicle pictures.""" - try: - _LOGGER.debug("Getting picture urls for vehicle %s", self.vin) - resp = await self.connection.get( - f"/connect/v1/vehicles/{self.vin}/pictures", - ) - for p in resp: - self.picture_locations[p["view"]] = p["url"] - except PorscheExceptionError as err: - _LOGGER.exception( - "Could not get capabilities, error communicating with API: %s", - err.message, - ) + _LOGGER.debug("Skipping picture lookup for vehicle %s; legacy picture endpoint is no longer used", self.vin) def __repr__(self) -> str: """Return a printable representation of the Porsche Connect Vehicle object.""" @@ -321,7 +387,11 @@ def _update_vehicle_data(self) -> None: # This attribute gives the chargingRate in the odd unit kilometers per minute. We add a km/h attribute. mdata["CHARGING_RATE"]["chargingRate-kph"] = mdata["CHARGING_RATE"]["chargingRate"] * 60 - if "CHARGING_SUMMARY" in mdata and mdata.get("CHARGING_SUMMARY", {}).get("mode") == "PROFILE": + if "CHARGING_SUMMARY" in mdata and mdata.get("CHARGING_SUMMARY", {}).get("targetSoC") is not None: + # Current app/connect payload already exposes the active target SoC explicitly. + mdata["CHARGING_SUMMARY"]["minSoC"] = mdata["CHARGING_SUMMARY"]["targetSoC"] + + elif "CHARGING_SUMMARY" in mdata and mdata.get("CHARGING_SUMMARY", {}).get("mode") == "PROFILE": # If charging profiles are enabled, get minSoC from this dict. mdata["CHARGING_SUMMARY"]["minSoC"] = mdata["CHARGING_SUMMARY"]["chargingProfile"]["minSoC"] @@ -329,7 +399,12 @@ def _update_vehicle_data(self) -> None: # If charging on departures are enabled, get minSoC from the CHARGING_SETTINGS dict. mdata["CHARGING_SUMMARY"]["minSoC"] = mdata["CHARGING_SETTINGS"]["targetSoc"] - if "DEPARTURES" not in mdata and "CHARGING_SUMMARY" in mdata and mdata.get("CHARGING_SUMMARY", {}).get("mode") == "DIRECT": + if ( + "DEPARTURES" not in mdata + and "CHARGING_SUMMARY" in mdata + and mdata.get("CHARGING_SUMMARY", {}).get("mode") == "DIRECT" + and mdata.get("CHARGING_SUMMARY", {}).get("minSoC") is None + ): # If direct charging is ongoing, minSoC is set to None in the API. We set it till 100 instead. mdata["CHARGING_SUMMARY"]["minSoC"] = 100