diff --git a/docs/SoC-Cupra-settings-1.png b/docs/SoC-Cupra-settings-1.png new file mode 100644 index 0000000000..5064108227 Binary files /dev/null and b/docs/SoC-Cupra-settings-1.png differ diff --git a/docs/SoC-Cupra-settings-2.png b/docs/SoC-Cupra-settings-2.png new file mode 100644 index 0000000000..219e090f75 Binary files /dev/null and b/docs/SoC-Cupra-settings-2.png differ diff --git a/docs/SoC-Cupra.md b/docs/SoC-Cupra.md new file mode 100644 index 0000000000..fbea412a55 --- /dev/null +++ b/docs/SoC-Cupra.md @@ -0,0 +1,82 @@ +# SoC-Modul Cupra + +Das SoC-Modul Cupra gibt es in openWB 2.x. + +## Konfiguration + +Die Konfiguration erfolgt im Bereich Einstellungen - Konfiguration - Fahrzeuge: + +![Allgemeine Konfiguration](SoC-Cupra-settings-1.png) + +![Spezielle Konfiguration](SoC-Cupra-settings-2.png) + +## Hinweise + +Für nicht-Cupra Fahrzeuge (Audi, Skoda, VW, etc.) funktioniert das Modul nicht. + +Erfolgreich getestet u.a. für folgende Fahrzeuge: Cupra Born. + +**Wichtig für alle Fahrzeuge:** +Es muss ein aktives Konto im Seat ID Portal vorhanden sein und die "My SEAT App" muss eingerichtet sein. Es muss explizit die "My SEAT App" eingerichtet werden und nicht die "My CUPRA App". + +**WICHTIG:** +Seat/Cupra ändert gelegentlich die Bedingungen für die Nutzung der Online-Services. + +Diese müssen bestätigt werden. Wenn der SOC-Abruf plötzlich nicht mehr funktioniert, VOR dem Posten bitte Schritt 1 ausführen. + +Bei Problemen zunächst bitte diese Schritte durchführen: + +1. sicherstellen, dass auf dieser Seat-Seite alle Einverständnisse gegeben wurden. + + + + In einigen Fällen wurden die Einverständnisse gegeben und trotzdem funktionierte die Abfrage nicht. + Hier hat folgendes Vorgehen geholfen: Im Seat Konto das Land temporär umstellen, d.h. + - auf ein anderes Land als das eigene ändern + - sichern + - zurück auf das eigene Land ändern + - sichern. + +2. Nach einem manuellen SOC-Abruf (Kreispfeil hinter dem SOC klicken) auf der Status - Seite EV-SOC Log und Debug log auf Fehler kontrollieren + +3. Falls im Ev-Soc Log Fehler 303 (unknown redirect) gemeldet wird: + - Ursache 1: Bestimmte Sonderzeichen im Passwort funktionieren nicht mit dem Modul. Bitte das Passwort auf eines ohne Sonderzeichen ändern und testen. + - Ursache 2: Falsche Email, Passwort oder VIN eingegeben. Alle 3 löschen, speichern, neu eingeben, speichern und testen. + +4. Falls eine Firewall im Spiel ist: Es gab einzelne Probleme beim Internet-Zugriff der openWB auf Python Archive und Fahrzeug-Server wenn IPV6 aktiv ist. + +5. Nach Neustart bzw. Änderung der LP-Konfiguration werden im EV-Soc-Log Fehler ausgegeben (permission oder fehlende Datei). + + Diese Fehler sind normal und können ignoriert werden. Leider wird im Debug Mode 0 keine Positiv-Meldung ausgegeben. + Empfehlung: + - In Einstellungen - System - Fehlersuche dies einstellen: Debug Level/Details + - dann einen manuellen SOC-Abruf durchführen (im Dashboard auf Kreispfeil klicken). + - danach sollte im EV-SOC-Log eine Zeile ähnlich dieser kommen: + + `2023-02-12 11:57:14 INFO:soc_skoda:Lp1 SOC: 61%@2023-02-12T11:53:20` + + Diese Zeile zeigt folgende Information: + + `2023-02-12 11:57:14` *- Timestamp des SOC-Abrufs* + + `INFO` *- Debug Level INFO* + + `soc_skoda` *- SOC-Modul* + + `Lp1` *- Ladepunkt* + + `SOC: 61%` *- SOC Stand* + + `@2025-02-12T11:53:20` *- Timestamp des Updates vom EV zum Seat Cloud-Server* + +6. Falls diese Schritte nicht zum Erfolg führen, das Problem im [Support Thema](https://forum.openwb.de/viewforum.php?f=12) mit Angabe relevanter Daten posten + - oWB SW Version + - oWB gekauft oder selbst installiert + - wenn selbst installiert: welches OS(Stretch/Buster) + - welches Fahrzeug + - falls vorhanden Angaben über Firewall, VPN, etc., also Appliances, die den Internetzugang limitieren könnten + - relevante Abschnitte der Logs, vor allem Fehlermeldungen, als CODE-blocks (). + +Das SoC-Log mit evtl. Fehlermeldungen kann wie folgt eingesehen werden: + +- Einstellungen - System - Fehlersuche diff --git a/docs/_Sidebar.md b/docs/_Sidebar.md index 0c9da7ff8d..11676906b0 100644 --- a/docs/_Sidebar.md +++ b/docs/_Sidebar.md @@ -6,6 +6,7 @@ * [Fahrzeuge](https://github.com/openWB/core/wiki/Fahrzeuge) * [Manueller SoC](https://github.com/openWB/core/wiki/Manueller-SoC) * [SoC BMW & Mini](https://github.com/openWB/core/wiki/SoC-BMW-Mini) + * [SoC Cupra](https://github.com/openWB/core/wiki/SoC-Cupra) * [SoC OVMS](https://github.com/openWB/core/wiki/SoC-OVMS) * [SoC VWId](https://github.com/openWB/core/wiki/SoC-VWId) * [WiCAN OBD2-Dongle mit manuellem SoC](https://github.com/openWB/core/wiki/WiCAN) diff --git a/packages/modules/vehicles/cupra/__init__.py b/packages/modules/vehicles/cupra/__init__.py new file mode 100755 index 0000000000..e69de29bb2 diff --git a/packages/modules/vehicles/cupra/api.py b/packages/modules/vehicles/cupra/api.py new file mode 100755 index 0000000000..cb877b8e2b --- /dev/null +++ b/packages/modules/vehicles/cupra/api.py @@ -0,0 +1,33 @@ +#!/usr/bin/env python3 + +import aiohttp +from asyncio import new_event_loop, set_event_loop +from typing import Union +from modules.vehicles.cupra import libcupra +from modules.vehicles.cupra.config import Cupra +from modules.vehicles.vwgroup.vwgroup import VwGroup + + +class api(VwGroup): + + def __init__(self, conf: Cupra, vehicle: int): + super().__init__(conf, vehicle) + + # async method, called from sync fetch_soc, required because libvwid/libskoda/libcupra expect async environment + async def _fetch_soc(self) -> Union[int, float, str]: + async with aiohttp.ClientSession() as self.session: + cupra = libcupra.cupra(self.session) + return await super().request_data(cupra) + + +def fetch_soc(conf: Cupra, vehicle: int) -> Union[int, float, str]: + + # prepare and call async method + loop = new_event_loop() + set_event_loop(loop) + + # get soc, range from server + a = api(conf, vehicle) + soc, range, soc_ts, soc_tsX = loop.run_until_complete(a._fetch_soc()) + + return soc, range, soc_ts, soc_tsX diff --git a/packages/modules/vehicles/cupra/config.py b/packages/modules/vehicles/cupra/config.py new file mode 100755 index 0000000000..edcc535a33 --- /dev/null +++ b/packages/modules/vehicles/cupra/config.py @@ -0,0 +1,28 @@ +from typing import Optional + + +class CupraConfiguration: + def __init__(self, + user_id: Optional[str] = None, # show in UI + password: Optional[str] = None, # show in UI + vin: Optional[str] = None, # show in UI + refreshToken: Optional[str] = None, # DON'T show in UI! + calculate_soc: bool = False # show in UI + ): + self.user_id = user_id + self.password = password + self.vin = vin + self.refreshToken = refreshToken + self.calculate_soc = calculate_soc + + +class Cupra: + def __init__(self, + name: str = "Cupra", + type: str = "cupra", + official: bool = False, + configuration: CupraConfiguration = None) -> None: + self.name = name + self.type = type + self.official = official + self.configuration = configuration or CupraConfiguration() diff --git a/packages/modules/vehicles/cupra/libcupra.py b/packages/modules/vehicles/cupra/libcupra.py new file mode 100755 index 0000000000..ab5a7fa89c --- /dev/null +++ b/packages/modules/vehicles/cupra/libcupra.py @@ -0,0 +1,265 @@ +# A Python class to communicate with the "Cupra Connect" API. +# Adapted the libvwid.py module to cupra interface + +import secrets +import logging +import json +import uuid +import base64 +import hashlib + +from helpermodules.utils.error_handling import ImportErrorContext +with ImportErrorContext(): + import lxml.html + +# Constants +LOGIN_BASE = "https://identity.vwgroup.io/oidc/v1" +LOGIN_HANDLER_BASE = "https://identity.vwgroup.io" +API_BASE = "https://ola.prod.code.seat.cloud.vwgroup.com" +CLIENT_ID = "99a5b77d-bd88-4d53-b4e5-a539c60694a3@apps_vw-dilab_com" + + +class cupra: + def __init__(self, session): + self.session = session + self.headers = {} + self.log = logging.getLogger(__name__) + self.jobs_string = 'all' + + def form_from_response(self, text): + page = lxml.html.fromstring(text) + elements = page.xpath('//form//input[@type="hidden"]') + form = {x.attrib['name']: x.attrib['value'] for x in elements} + return (form, page.forms[0].action) + + def password_form(self, text): + page = lxml.html.fromstring(text) + elements = page.xpath('//script') + + # Todo: Find more elegant way parse this... + objects = {} + for a in elements: + if (a.text) and (a.text.find('window._IDK') != -1): + text = a.text.strip() + text = text[text.find('\n'):text.rfind('\n')].strip() + for line in text.split('\n'): + try: + (name, val) = line.strip().split(':', 1) + except ValueError: + continue + val = val.strip('\', ') + objects[name] = val + + json_model = json.loads(objects['templateModel']) + + if ('errorCode' in json_model): + self.log.error("Login error: %s", json_model['errorCode']) + return False + + try: + # Generate form + form = {} + form['relayState'] = json_model['relayState'] + form['hmac'] = json_model['hmac'] + form['email'] = json_model['emailPasswordForm']['email'] + form['_csrf'] = objects['csrf_token'] + + # Generate URL action + action = '/signin-service/v1/%s/%s'\ + % (json_model['clientLegalEntityModel']['clientId'], json_model['postAction']) + + return (form, action) + + except KeyError: + self.log.exception("Missing fields in response from Cupra API") + return False + + def set_vin(self, vin): + self.vin = vin + + def set_credentials(self, username, password): + self.username = username + self.password = password + + def set_jobs(self, jobs): + self.jobs_string = ','.join(jobs) + + def get_code_challenge(self): + code_verifier = secrets.token_urlsafe(64).replace('+', '-').replace('/', '_').replace('=', '') + code_challenge = base64.b64encode(hashlib.sha256(code_verifier.encode('utf-8')).digest()) + code_challenge = code_challenge.decode('utf-8').replace('+', '-').replace('/', '_').replace('=', '') + return (code_verifier, code_challenge) + + def convert_to_camel_case(self, json): + def to_camel_case(s): + parts = s.split('_') + return parts[0] + ''.join(word.capitalize() for word in parts[1:]) + + return {to_camel_case(k): v for k, v in json.items()} + + async def connect(self, username, password): + self.set_credentials(username, password) + return (await self.reconnect()) + + async def reconnect(self): + # Get code challenge and verifier + code_verifier, code_challenge = self.get_code_challenge() + + # Get authorize page + _scope = 'openid profile nickname birthdate phone' + payload = { + 'client_id': CLIENT_ID, + 'scope': _scope, + 'response_type': 'code', + 'nonce': secrets.token_urlsafe(12), + 'redirect_uri': 'seat://oauth-callback', + 'state': str(uuid.uuid4()), + 'code_challenge': code_challenge, + 'code_challenge_method': 'S256' + } + + response = await self.session.get(LOGIN_BASE + '/authorize', params=payload) + if response.status >= 400: + self.log.error(f"Authorize: Non-2xx response ({response.status})") + # Non 2xx response, failed + return False + + # Fill form with email (username) + (form, action) = self.form_from_response(await response.read()) + form['email'] = self.username + response = await self.session.post(LOGIN_HANDLER_BASE+action, data=form) + if response.status >= 400: + self.log.error("Email: Non-2xx response") + return False + + # Fill form with password + (form, action) = self.password_form(await response.read()) + form['password'] = self.password + url = LOGIN_HANDLER_BASE + action + response = await self.session.post(url, data=form, allow_redirects=False) + + # Can get a 303 redirect for a "terms and conditions" page + if (response.status == 303): + url = response.headers['Location'] + if ("terms-and-conditions" in url): + # Get terms and conditions page + url = LOGIN_HANDLER_BASE + url + response = await self.session.get(url, data=form, allow_redirects=False) + (form, action) = self.form_from_response(await response.read()) + + url = LOGIN_HANDLER_BASE + action + response = await self.session.post(url, data=form, allow_redirects=False) + + self.log.warning("Agreed to terms and conditions") + else: + self.log.error("Got unknown 303 redirect") + return False + + # Handle every single redirect and stop if the redirect + # URL uses the seat adapter. + while (True): + url = response.headers['Location'] + if (url.split(':')[0] == "seat"): + if not ('code' in url): + self.log.error("Missing authorization code") + return False + # Parse query string + query_string = url.split('?')[1] + query = {x[0]: x[1] for x in [x.split("=") for x in query_string.split("&")]} + break + + if (response.status != 302): + self.log.error("Not redirected, status %u" % response.status) + return False + + response = await self.session.get(url, data=form, allow_redirects=False) + + # Get final token + form = {} + form['code'] = query['code'] + form['client_id'] = CLIENT_ID + form['redirect_uri'] = "seat://oauth-callback" + form['grant_type'] = 'authorization_code' + form['code_verifier'] = code_verifier + headers = {} + headers['Content-Type'] = 'application/x-www-form-urlencoded' + headers['Authorization'] = 'BASIC OTlhNWI3N2QtYmQ4OC00ZDUzLWI0ZTUtYTUzOWM2MDY5NGEzQGFwcHNfdnctZGlsYWJfY29tOg==' + headers['User-Agent'] = ( + 'SEATApp/2.5.0 (com.seat.myseat.ola; build:202410171614; ' + 'iOS 15.8.3) Alamofire/5.7.0 Mobile' + ) + + response = await self.session.post(API_BASE + '/authorization/api/v1/token', + headers=headers, data=form) + if response.status >= 400: + self.log.error("Login: Non-2xx response") + # Non 2xx response, failed + return False + self.tokens = self.convert_to_camel_case(await response.json()) + + # Update header with final token + self.headers['Authorization'] = 'Bearer %s' % self.tokens["accessToken"] + + # Success + return True + + async def refresh_tokens(self): + if not self.headers: + return False + + # Use the refresh token + form = {} + form['client_id'] = CLIENT_ID + form['grant_type'] = 'refresh_token' + form['refresh_token'] = self.tokens["refreshToken"] + headers = {} + headers['Content-Type'] = 'application/x-www-form-urlencoded' + headers['User-Agent'] = ( + 'SEATApp/2.5.0 (com.seat.myseat.ola; build:202410171614; ' + 'iOS 15.8.3) Alamofire/5.7.0 Mobile' + ) + + response = await self.session.post(API_BASE + '/authorization/api/v1/token', + headers=headers, data=form) + if response.status >= 400: + return False + self.tokens = self.convert_to_camel_case(await response.json()) + + # Use the newly received access token + self.headers['Authorization'] = 'Bearer %s' % self.tokens["accessToken"] + + return True + + async def get_status(self): + status_url = f"{API_BASE}/vehicles/{self.vin}/charging/status" + response = await self.session.get(status_url, headers=self.headers) + + # If first attempt fails, try to refresh tokens + if response.status >= 400: + self.log.debug("Refreshing tokens") + if await self.refresh_tokens(): + response = await self.session.get(status_url, headers=self.headers) + + # If refreshing tokens failed, try a full reconnect + if response.status >= 400: + self.log.info("Reconnecting") + if await self.reconnect(): + response = await self.session.get(status_url, headers=self.headers) + else: + self.log.error("Reconnect failed") + return {} + + if response.status >= 400: + self.log.error("Get status failed") + return {} + + status_data = await response.json() + self.log.debug(f"Status data from Cupra API: {status_data}") + + return { + 'charging': { + 'batteryStatus': { + 'value': status_data['status']['battery'] + } + } + } diff --git a/packages/modules/vehicles/cupra/soc.py b/packages/modules/vehicles/cupra/soc.py new file mode 100755 index 0000000000..b5326c192b --- /dev/null +++ b/packages/modules/vehicles/cupra/soc.py @@ -0,0 +1,43 @@ +from typing import List + +import logging + +from helpermodules.cli import run_using_positional_cli_args +from modules.common import store +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.cupra import api +from modules.vehicles.cupra.config import Cupra, CupraConfiguration + + +log = logging.getLogger(__name__) + + +def fetch(vehicle_update_data: VehicleUpdateData, config: Cupra, vehicle: int) -> CarState: + soc, range, soc_ts, soc_tsX = api.fetch_soc(config, vehicle) + log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts) + return CarState(soc=soc, range=range, soc_timestamp=soc_tsX) + + +def create_vehicle(vehicle_config: Cupra, vehicle: int): + 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) + + +def cupra_update(user_id: str, password: str, vin: str, refreshToken: str, charge_point: int): + log.debug("cupra: user_id="+user_id+"vin="+vin+"charge_point="+str(charge_point)) + store.get_car_value_store(charge_point).store.set( + fetch(None, Cupra(configuration=CupraConfiguration(user_id, password, vin, refreshToken)), charge_point)) + + +def main(argv: List[str]): + run_using_positional_cli_args(cupra_update, argv) + + +device_descriptor = DeviceDescriptor(configuration_factory=Cupra)