diff --git a/packages/helpermodules/update_config.py b/packages/helpermodules/update_config.py index 36735ee5be..6fe83c3caf 100644 --- a/packages/helpermodules/update_config.py +++ b/packages/helpermodules/update_config.py @@ -58,7 +58,7 @@ class UpdateConfig: - DATASTORE_VERSION = 134 + DATASTORE_VERSION = 135 valid_topic = [ "^openWB/bat/config/bat_control_activated$", @@ -3435,3 +3435,16 @@ def upgrade(topic: str, payload) -> Optional[dict]: return {topic: config} self._loop_all_received_topics(upgrade) self._append_datastore_version(134) + + def upgrade_datastore_135(self) -> None: + def upgrade(topic: str, payload) -> None: + if re.search("openWB/vehicle/[0-9]+/soc_module/config", topic) is not None: + configuration_payload = decode_payload(payload) + # replace cupra,skoda,vwid by vweuda + if configuration_payload.get("type") in ["cupra", "skoda", "vwid"]: + configuration_payload.update({"type": "vweuda"}) + if configuration_payload['configuration'].get('refreshToken'): + configuration_payload['configuration'].pop('refreshToken') + return {topic: configuration_payload} + self._loop_all_received_topics(upgrade) + self._append_datastore_version(135) diff --git a/packages/modules/vehicles/cupra/api.py b/packages/modules/vehicles/cupra/api.py deleted file mode 100755 index 7533cecc1d..0000000000 --- a/packages/modules/vehicles/cupra/api.py +++ /dev/null @@ -1,36 +0,0 @@ -#!/usr/bin/env python3 - -import aiohttp -from asyncio import new_event_loop, set_event_loop -from typing import Optional, Tuple -from modules.vehicles.cupra import libcupra -from modules.vehicles.cupra.config import Cupra -from modules.vehicles.vwgroup.vwgroup import VwGroup - - -SoCResult = Tuple[int, float, str, float, Optional[float]] - - -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) -> SoCResult: - 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) -> SoCResult: - - # prepare and call async method - loop = new_event_loop() - set_event_loop(loop) - - # get soc, range, timestamp, and odometer from server - a = api(conf, vehicle) - soc, range, soc_ts, soc_tsX, odometer = loop.run_until_complete(a._fetch_soc()) - - return soc, range, soc_ts, soc_tsX, odometer diff --git a/packages/modules/vehicles/cupra/config.py b/packages/modules/vehicles/cupra/config.py deleted file mode 100755 index edcc535a33..0000000000 --- a/packages/modules/vehicles/cupra/config.py +++ /dev/null @@ -1,28 +0,0 @@ -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 deleted file mode 100755 index 93589b219e..0000000000 --- a/packages/modules/vehicles/cupra/libcupra.py +++ /dev/null @@ -1,326 +0,0 @@ -# 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 datetime import datetime, timezone -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 = "3c756d46-f1ba-4d78-9f9a-cff0d5292d51@apps_vw-dilab_com" -USER_AGENT = "OLACupra/2.16.0 (Android 14; Pixel 8; Google) Mobile" - - -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()} - - def token_headers(self): - basic_auth = base64.b64encode(f"{CLIENT_ID}:".encode('utf-8')).decode('utf-8') - return { - 'Content-Type': 'application/x-www-form-urlencoded', - 'Authorization': f'Basic {basic_auth}', - 'User-Agent': USER_AGENT, - } - - 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() - self.log.debug("Starting Cupra reconnect/auth flow") - - # Get authorize page - _scope = ( - 'openid profile nickname birthdate phone mbb cars address ' - 'nationalIdentifier nationality profession badge driversLicense' - ) - payload = { - 'client_id': CLIENT_ID, - 'scope': _scope, - 'response_type': 'code', - 'nonce': secrets.token_urlsafe(12), - 'redirect_uri': 'cupra://oauth-callback', - 'state': str(uuid.uuid4()), - 'code_challenge': code_challenge, - 'code_challenge_method': 'S256' - } - - response = await self.session.get(LOGIN_BASE + '/authorize', params=payload) - self.log.debug("Authorize request finished with status=%s", response.status) - 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 - self.log.debug("Submitting email form to action=%s", action) - response = await self.session.post(LOGIN_HANDLER_BASE+action, data=form) - self.log.debug("Email form response status=%s", response.status) - 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()) - if not form or not action: - self.log.error("Password form parsing failed") - return False - form['password'] = self.password - url = LOGIN_HANDLER_BASE + action - self.log.debug("Submitting password form to url=%s", url) - response = await self.session.post(url, data=form, allow_redirects=False) - self.log.debug("Password form response status=%s", response.status) - - # Can get a 303 redirect for a "terms and conditions" page - if (response.status == 303): - url = response.headers['Location'] - self.log.debug("Received 303 redirect to %s", url) - if ("terms-and-conditions" in url): - # Get terms and conditions page - url = LOGIN_HANDLER_BASE + url - self.log.debug("Opening terms and conditions page: %s", 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 - self.log.debug("Submitting terms and conditions form to %s", url) - 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): - if 'Location' not in response.headers: - self.log.error("Redirect handling stopped: missing Location header (status=%s)", - response.status) - return False - - url = response.headers['Location'] - self.log.debug("Redirect: status=%s location=%s", response.status, url) - if (url.split(':')[0] == "cupra"): - 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("&")]} - self.log.debug("Authorization redirect reached") - break - - if (response.status != 302): - self.log.error("Not redirected, status=%u, last_url=%s", response.status, url) - 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'] = "cupra://oauth-callback" - form['grant_type'] = 'authorization_code' - form['code_verifier'] = code_verifier - headers = self.token_headers() - - response = await self.session.post(API_BASE + '/authorization/api/v1/token', - headers=headers, data=form) - self.log.debug("Authorization code exchange status=%s", response.status) - if response.status >= 400: - self.log.error("Login: Non-2xx response (%s), body=%s", - response.status, await response.text()) - # 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"] - self.log.debug("Cupra authorization completed successfully") - - # 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 = self.token_headers() - - response = await self.session.post(API_BASE + '/authorization/api/v1/token', - headers=headers, data=form) - if response.status >= 400: - self.log.error("Refresh token exchange failed (%s), body=%s", - response.status, await response.text()) - 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): - self.headers['user-agent'] = USER_AGENT - self.headers['app-brand'] = 'cupra' - self.headers['app-market'] = 'android' - self.headers['app-version'] = '2.16.0' - status_url = f"{API_BASE}/v1/vehicles/{self.vin}/charging/status" - statusv2_url = f"{API_BASE}/v2/vehicles/{self.vin}/status" - mileage_url = f"{API_BASE}/v1/vehicles/{self.vin}/mileage" - 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}") - - # Fetch mileage data - response = await self.session.get(mileage_url, headers=self.headers) - if response.status >= 400: - self.log.error("Get mileage failed") - odometer = None - else: - mileage_data = await response.json() - self.log.debug(f"Mileage data from Cupra API: {mileage_data}") - odometer = mileage_data.get('mileageKm', None) - - # Fetch additional status data from v2 endpoint - response = await self.session.get(statusv2_url, headers=self.headers) - if response.status >= 400: - self.log.error("Get status v2 failed") - else: - statusv2_data = await response.json() - self.log.debug(f"Status v2 data from Cupra API: {statusv2_data}") - # use current timestamp as a fallback, as the API field is missing - carCapturedTimestamp = statusv2_data.get('updatedAt', datetime.now(timezone.utc).isoformat()).split('.')[0] - if not carCapturedTimestamp.endswith('Z'): - carCapturedTimestamp += 'Z' - - battery_value = { - 'currentSOC_pct': status_data['battery']['currentSocPercentage'], - 'cruisingRangeElectric_km': status_data['battery']['estimatedRangeInKm'], - 'carCapturedTimestamp': status_data['battery'].get('updatedAt', carCapturedTimestamp), - 'odometer': odometer, - } - - return { - 'charging': { - 'batteryStatus': { - 'value': battery_value, - } - } - } diff --git a/packages/modules/vehicles/cupra/soc.py b/packages/modules/vehicles/cupra/soc.py deleted file mode 100755 index cc6cefb0c2..0000000000 --- a/packages/modules/vehicles/cupra/soc.py +++ /dev/null @@ -1,43 +0,0 @@ -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, odometer = api.fetch_soc(config, vehicle) - log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts + ", odometer=" + str(odometer)) - return CarState(soc=soc, range=range, soc_timestamp=soc_tsX, odometer=odometer) - - -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) diff --git a/packages/modules/vehicles/cupra/soc_test.py b/packages/modules/vehicles/cupra/soc_test.py deleted file mode 100644 index 6b4ca994a9..0000000000 --- a/packages/modules/vehicles/cupra/soc_test.py +++ /dev/null @@ -1,189 +0,0 @@ -from unittest.mock import Mock, MagicMock -import pytest -import asyncio -from modules.common import store -from modules.common.abstract_vehicle import VehicleUpdateData -from modules.common.component_context import SingleComponentUpdateContext -from modules.vehicles.cupra import api -from modules.vehicles.cupra.soc import create_vehicle -from modules.vehicles.cupra.config import Cupra, CupraConfiguration -from modules.vehicles.cupra.libcupra import cupra as CupraApi, CLIENT_ID, USER_AGENT -import base64 - - -class TestCupra: - @pytest.fixture(autouse=True) - def set_up(self, monkeypatch): - self.mock_context_exit = Mock(return_value=True) - self.mock_fetch_soc = Mock( - name="fetch_soc", - return_value=(60, 320, "2026-03-29T11:00:00Z", 1774782000.0, 45678), - ) - self.mock_value_store = Mock(name="value_store") - monkeypatch.setattr(api, "fetch_soc", self.mock_fetch_soc) - monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=self.mock_value_store)) - monkeypatch.setattr(SingleComponentUpdateContext, '__exit__', self.mock_context_exit) - - def test_update_updates_value_store(self): - # setup - config = Cupra(configuration=CupraConfiguration(user_id="test_user", password="test_password", vin="test_vin")) - - # execution - create_vehicle(config, 1).update(VehicleUpdateData()) - - # evaluation - self.assert_context_manager_called_with(None) - self.mock_fetch_soc.assert_called_once_with(config, 1) - assert self.mock_value_store.set.call_count == 1 - call_args = self.mock_value_store.set.call_args[0][0] - assert call_args.soc == 60 - assert call_args.range == 320 - assert call_args.soc_timestamp == 1774782000.0 - assert call_args.odometer == 45678 - - def test_update_passes_errors_to_context(self): - # setup - dummy_error = Exception("Der SoC kann nicht ausgelesen werden") - self.mock_fetch_soc.side_effect = dummy_error - config = Cupra(configuration=CupraConfiguration(user_id="test_user", password="test_password", vin="test_vin")) - - # execution - create_vehicle(config, 1).update(VehicleUpdateData()) - - # evaluation - self.assert_context_manager_called_with_substr(dummy_error) - - def assert_context_manager_called_with(self, error): - assert self.mock_context_exit.call_count == 1 - assert self.mock_context_exit.call_args[0][1] is error - - def assert_context_manager_called_with_substr(self, error): - assert self.mock_context_exit.call_count == 1 - assert str(error) in str(self.mock_context_exit.call_args[0][1]) - - -class MockAiohttpResponse: - def __init__(self, json_data, status_code): - self._json_data = json_data - self.status = status_code - - async def json(self): - return self._json_data - - def release(self): - pass - - async def __aenter__(self): - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - pass - - -class TestCupraGetStatus: - @pytest.fixture - def mock_session(self): - session = MagicMock() - - async def get_side_effect(*args, **kwargs): - return session.get.return_value - session.get = MagicMock(side_effect=get_side_effect) - return session - - @pytest.fixture - def cupra_instance(self, mock_session): - instance = CupraApi(mock_session) - instance.set_vin("test_vin") - instance.headers = { - "Authorization": "Bearer test_token", - "app-market": "android", - } - return instance - - def test_token_headers_use_dynamic_basic_auth_and_user_agent(self, cupra_instance): - headers = cupra_instance.token_headers() - expected_basic = base64.b64encode(f"{CLIENT_ID}:".encode("utf-8")).decode("utf-8") - - assert headers["Content-Type"] == "application/x-www-form-urlencoded" - assert headers["Authorization"] == f"Basic {expected_basic}" - assert headers["User-Agent"] == USER_AGENT - - def test_get_status_success(self, cupra_instance, mock_session): - # setup - status_response_data = { - "battery": { - "currentSocPercentage": 67, - "estimatedRangeInKm": 305 - } - } - mileage_response_data = { - "mileageKm": 77889 - } - statusv2_response_data = { - "updatedAt": "2026-03-29T11:20:00Z" - } - responses = [ - MockAiohttpResponse(status_response_data, 200), - MockAiohttpResponse(mileage_response_data, 200), - MockAiohttpResponse(statusv2_response_data, 200) - ] - - async def side_effect_func(*args, **kwargs): - return responses.pop(0) - mock_session.get.side_effect = side_effect_func - - # execution - status = asyncio.run(cupra_instance.get_status()) - - # evaluation - assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 67 - assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 305 - assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2026-03-29T11:20:00Z" - assert status['charging']['batteryStatus']['value']['odometer'] == 77889 - assert mock_session.get.call_count == 3 - mock_session.get.assert_any_call( - "https://ola.prod.code.seat.cloud.vwgroup.com/v1/vehicles/test_vin/charging/status", - headers=cupra_instance.headers - ) - mock_session.get.assert_any_call( - "https://ola.prod.code.seat.cloud.vwgroup.com/v1/vehicles/test_vin/mileage", - headers=cupra_instance.headers - ) - mock_session.get.assert_any_call( - "https://ola.prod.code.seat.cloud.vwgroup.com/v2/vehicles/test_vin/status", - headers=cupra_instance.headers - ) - - def test_get_status_mileage_error_sets_odometer_none(self, cupra_instance, mock_session): - # setup - status_response_data = { - "battery": { - "currentSocPercentage": 67, - "estimatedRangeInKm": 305, - "updatedAt": "2026-03-29T11:20:00Z" - } - } - mileage_error_response = { - "error": "service_unavailable" - } - statusv2_response_data = { - "error": "service_unavailable" - } - responses = [ - MockAiohttpResponse(status_response_data, 200), - MockAiohttpResponse(mileage_error_response, 503), - MockAiohttpResponse(statusv2_response_data, 200) - ] - - async def side_effect_func(*args, **kwargs): - return responses.pop(0) - mock_session.get.side_effect = side_effect_func - - # execution - status = asyncio.run(cupra_instance.get_status()) - - # evaluation - assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 67 - assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 305 - assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2026-03-29T11:20:00Z" - assert status['charging']['batteryStatus']['value']['odometer'] is None diff --git a/packages/modules/vehicles/skoda/__init__.py b/packages/modules/vehicles/skoda/__init__.py deleted file mode 100755 index e69de29bb2..0000000000 diff --git a/packages/modules/vehicles/skoda/api.py b/packages/modules/vehicles/skoda/api.py deleted file mode 100755 index 7cb2fee565..0000000000 --- a/packages/modules/vehicles/skoda/api.py +++ /dev/null @@ -1,36 +0,0 @@ -#!/usr/bin/env python3 - -import aiohttp -from asyncio import new_event_loop, set_event_loop -from typing import Optional, Tuple -from modules.vehicles.skoda import libskoda -from modules.vehicles.skoda.config import Skoda -from modules.vehicles.vwgroup.vwgroup import VwGroup - - -SoCResult = Tuple[int, float, str, float, Optional[float]] - - -class api(VwGroup): - - def __init__(self, conf: Skoda, vehicle: int): - super().__init__(conf, vehicle) - - # async method, called from sync fetch_soc, required because libvwid/libskoda expect async environment - async def _fetch_soc(self) -> SoCResult: - async with aiohttp.ClientSession() as self.session: - skoda = libskoda.skoda(self.session) - return await super().request_data(skoda) - - -def fetch_soc(conf: Skoda, vehicle: int) -> SoCResult: - - # prepare and call async method - loop = new_event_loop() - set_event_loop(loop) - - # get soc, range, timestamp, and odometer from server - a = api(conf, vehicle) - soc, range, soc_ts, soc_tsX, odometer = loop.run_until_complete(a._fetch_soc()) - - return soc, range, soc_ts, soc_tsX, odometer diff --git a/packages/modules/vehicles/skoda/config.py b/packages/modules/vehicles/skoda/config.py deleted file mode 100755 index d91a690596..0000000000 --- a/packages/modules/vehicles/skoda/config.py +++ /dev/null @@ -1,28 +0,0 @@ -from typing import Optional - - -class SkodaConfiguration: - 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 Skoda: - def __init__(self, - name: str = "Skoda", - type: str = "skoda", - official: bool = False, - configuration: SkodaConfiguration = None) -> None: - self.name = name - self.type = type - self.official = official - self.configuration = configuration or SkodaConfiguration() diff --git a/packages/modules/vehicles/skoda/libskoda.py b/packages/modules/vehicles/skoda/libskoda.py deleted file mode 100755 index 0144b6ed4e..0000000000 --- a/packages/modules/vehicles/skoda/libskoda.py +++ /dev/null @@ -1,325 +0,0 @@ -# A Python class to communicate with the "Skoda Connect" API. -# Adapted the libvwid.py module to skoda 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://mysmob.api.connect.skoda-auto.cz/api" -CLIENT_ID = "7f045eee-7003-4379-9968-9355ed2adb06@apps_vw-dilab_com" - - -class skoda: - 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 Skoda 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) - - 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() - self.log.debug("Starting Skoda reconnect/auth flow") - - # Get authorize page - _scope = 'address badge birthdate cars driversLicense dealers email mileage mbb nationalIdentifier' - _scope = _scope + ' openid phone profession profile vin' - payload = { - 'client_id': CLIENT_ID, - 'scope': _scope, - 'response_type': 'code', - 'nonce': secrets.token_urlsafe(12), - 'redirect_uri': 'myskoda://redirect/login/', - 'state': str(uuid.uuid4()), - 'code_challenge': code_challenge, - 'code_challenge_method': 'S256' - } - - response = await self.session.get(LOGIN_BASE + '/authorize', params=payload) - self.log.debug("Authorize request finished with status=%s", response.status) - 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 - self.log.debug("Submitting email form to action=%s", action) - response = await self.session.post(LOGIN_HANDLER_BASE+action, data=form) - self.log.debug("Email form response status=%s", response.status) - 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()) - if not form or not action: - self.log.error("Password form parsing failed") - return False - form['password'] = self.password - url = LOGIN_HANDLER_BASE + action - self.log.debug("Submitting password form to url=%s", url) - response = await self.session.post(url, data=form, allow_redirects=False) - self.log.debug("Password form response status=%s", response.status) - - # Can get a 303 redirect for a "terms and conditions" page - if (response.status == 303): - url = response.headers['Location'] - self.log.debug("Received 303 redirect to %s", url) - if ("terms-and-conditions" in url): - # Get terms and conditions page - url = LOGIN_HANDLER_BASE + url - self.log.debug("Opening terms and conditions page: %s", 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 - self.log.debug("Submitting terms and conditions form to %s", url) - 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 weconnect adapter. - while (True): - if 'Location' not in response.headers: - self.log.error("Redirect handling stopped: missing Location header (status=%s)", - response.status) - return False - - url = response.headers['Location'] - self.log.debug("Redirect: status=%s location=%s", response.status, url) - if (url.split(':')[0] == "myskoda"): - 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("&")]} - self.log.debug("Authorization redirect reached") - break - - if (response.status != 302): - self.log.error("Not redirected, status=%u, last_url=%s", - response.status, url) - return False - - response = await self.session.get(url, data=form, allow_redirects=False) - - # Get final token - params = { - 'tokenType': 'CONNECT' - } - payload = { - 'code': query['code'], - 'redirectUri': "myskoda://redirect/login/", - 'verifier': code_verifier - } - response = await self.session.post(API_BASE + '/v1/authentication/exchange-authorization-code', - params=params, json=payload) - self.log.debug("Authorization code exchange status=%s", response.status) - if response.status >= 400: - self.log.error("Login: Non-2xx response") - # Non 2xx response, failed - return False - self.tokens = await response.json() - - # Update header with final token - self.headers['Authorization'] = 'Bearer %s' % self.tokens["accessToken"] - self.log.debug("Skoda authorization completed successfully") - - # Success - return True - - async def refresh_tokens(self): - if not self.headers: - return False - - params = { - 'tokenType': 'CONNECT' - } - # Use the refresh token - payload = { - 'token': self.tokens["refreshToken"] - } - - response = await self.session.post(API_BASE + '/v1/authentication/refresh-token', params=params, json=payload) - if response.status >= 400: - return False - self.tokens = await response.json() - - # Use the newly received access token - self.headers['Authorization'] = 'Bearer %s' % self.tokens["accessToken"] - - return True - - async def get_status(self): - vehicle_status_url = f"{API_BASE}/v2/vehicle-status/{self.vin}/driving-range" - charging_url = f"{API_BASE}/v1/charging/{self.vin}" - maintenance_report_url = f"{API_BASE}/v3/vehicle-maintenance/vehicles/{self.vin}/report" - response = await self.session.get(vehicle_status_url, headers=self.headers) - - if response.status == 403: - self.log.info("vehicle-status returned 403, trying charging_url") - status_data = {} - else: - # 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(vehicle_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(vehicle_status_url, headers=self.headers) - else: - self.log.error("Reconnect failed") - return {} - - if response.status >= 400: - if response.status == 403: - self.log.info("vehicle-status returned 403, trying charging_url") - status_data = {} - else: - self.log.error("Get status failed") - return {} - else: - status_data = await response.json() - self.log.debug(f"Status data from Skoda API (vehicle-status): {status_data}") - - # check if all values are valid, otherwise use charging_url - electric_engine_range = {} - if 'primaryEngineRange' in status_data and status_data['primaryEngineRange']['engineType'] == "electric": - electric_engine_range = status_data['primaryEngineRange'] - elif 'secondaryEngineRange' in status_data and status_data['secondaryEngineRange']['engineType'] == "electric": - electric_engine_range = status_data['secondaryEngineRange'] - - required_keys = ['currentSoCInPercent', 'remainingRangeInKm'] - if not all(k in electric_engine_range for k in required_keys) or 'carCapturedTimestamp' not in status_data: - self.log.info("vehicle-status did not contain all values, trying charging_url") - response = await self.session.get(charging_url, headers=self.headers) - - if response.status >= 400: - self.log.error("Get status from charging_url failed") - return {} - - status_data = await response.json() - self.log.debug(f"Status data from Skoda API (charging): {status_data}") - - soc = status_data['status']['battery']['stateOfChargeInPercent'] - range_km = ( - status_data['status']['battery'].get('remainingCruisingRangeInMeters', 1000) / 1000 - ) - else: - soc = electric_engine_range['currentSoCInPercent'] - range_km = electric_engine_range['remainingRangeInKm'] - - timestamp = status_data['carCapturedTimestamp'].split('.')[0] - if not timestamp.endswith('Z'): - timestamp += 'Z' - - response = await self.session.get(maintenance_report_url, headers=self.headers) - if response.status >= 400: - self.log.error("Get status from maintenance_report_url failed") - odometer = None - else: - maintenance_data = await response.json() - self.log.debug(f"Status data from Skoda API (maintenance report): {maintenance_data}") - odometer = maintenance_data.get('mileageInKm', None) - - return { - 'charging': { - 'batteryStatus': { - 'value': { - 'currentSOC_pct': soc, - 'cruisingRangeElectric_km': range_km, - 'carCapturedTimestamp': timestamp, - 'odometer': odometer, - } - } - } - } diff --git a/packages/modules/vehicles/skoda/soc.py b/packages/modules/vehicles/skoda/soc.py deleted file mode 100755 index 2032f89c24..0000000000 --- a/packages/modules/vehicles/skoda/soc.py +++ /dev/null @@ -1,43 +0,0 @@ -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.skoda import api -from modules.vehicles.skoda.config import Skoda, SkodaConfiguration - - -log = logging.getLogger(__name__) - - -def fetch(vehicle_update_data: VehicleUpdateData, config: Skoda, vehicle: int) -> CarState: - soc, range, soc_ts, soc_tsX, odometer = api.fetch_soc(config, vehicle) - log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts + ", odometer=" + str(odometer)) - return CarState(soc=soc, range=range, soc_timestamp=soc_tsX, odometer=odometer) - - -def create_vehicle(vehicle_config: Skoda, 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 skoda_update(user_id: str, password: str, vin: str, refreshToken: str, charge_point: int): - log.debug("skoda: user_id="+user_id+"vin="+vin+"charge_point="+str(charge_point)) - store.get_car_value_store(charge_point).store.set( - fetch(None, Skoda(configuration=SkodaConfiguration(user_id, password, vin, refreshToken)), charge_point)) - - -def main(argv: List[str]): - run_using_positional_cli_args(skoda_update, argv) - - -device_descriptor = DeviceDescriptor(configuration_factory=Skoda) diff --git a/packages/modules/vehicles/skoda/soc_test.py b/packages/modules/vehicles/skoda/soc_test.py deleted file mode 100644 index b1a9d56577..0000000000 --- a/packages/modules/vehicles/skoda/soc_test.py +++ /dev/null @@ -1,302 +0,0 @@ -from unittest.mock import Mock, MagicMock -import pytest -import asyncio -from modules.common import store -from modules.common.abstract_vehicle import VehicleUpdateData -from modules.common.component_context import SingleComponentUpdateContext -from modules.vehicles.skoda import api -from modules.vehicles.skoda.soc import create_vehicle -from modules.vehicles.skoda.config import Skoda, SkodaConfiguration -from modules.vehicles.skoda.libskoda import skoda as SkodaApi - - -class TestSkoda: - @pytest.fixture(autouse=True) - def set_up(self, monkeypatch): - self.mock_context_exit = Mock(return_value=True) - self.mock_fetch_soc = Mock( - name="fetch_soc", - return_value=(50, 250, "2025-10-18T10:00:00Z", 1760822400.0, 12345), - ) - self.mock_value_store = Mock(name="value_store") - monkeypatch.setattr(api, "fetch_soc", self.mock_fetch_soc) - monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=self.mock_value_store)) - monkeypatch.setattr(SingleComponentUpdateContext, '__exit__', self.mock_context_exit) - - def test_update_updates_value_store(self): - # setup - config = Skoda(configuration=SkodaConfiguration(user_id="test_user", password="test_password", vin="test_vin")) - - # execution - create_vehicle(config, 1).update(VehicleUpdateData()) - - # evaluation - self.assert_context_manager_called_with(None) - self.mock_fetch_soc.assert_called_once_with(config, 1) - assert self.mock_value_store.set.call_count == 1 - call_args = self.mock_value_store.set.call_args[0][0] - assert call_args.soc == 50 - assert call_args.range == 250 - assert call_args.soc_timestamp == 1760822400.0 - assert call_args.odometer == 12345 - - def test_update_passes_errors_to_context(self): - # setup - dummy_error = Exception("Der SoC kann nicht ausgelesen werden") - self.mock_fetch_soc.side_effect = dummy_error - config = Skoda(configuration=SkodaConfiguration(user_id="test_user", password="test_password", vin="test_vin")) - - # execution - create_vehicle(config, 1).update(VehicleUpdateData()) - - # evaluation - # self.assert_context_manager_called_with(dummy_error) - self.assert_context_manager_called_with_substr(dummy_error) - - def assert_context_manager_called_with(self, error): - assert self.mock_context_exit.call_count == 1 - assert self.mock_context_exit.call_args[0][1] is error - - def assert_context_manager_called_with_substr(self, error): - assert self.mock_context_exit.call_count == 1 - assert str(error) in str(self.mock_context_exit.call_args[0][1]) - - -class MockAiohttpResponse: - def __init__(self, json_data, status_code): - self._json_data = json_data - self.status = status_code - - async def json(self): - return self._json_data - - def release(self): - pass - - async def __aenter__(self): - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - pass - - -class TestSkodaGetStatus: - @pytest.fixture - def mock_session(self): - session = MagicMock() - - async def get_side_effect(*args, **kwargs): - return session.get.return_value - session.get = MagicMock(side_effect=get_side_effect) - return session - - @pytest.fixture - def skoda_instance(self, mock_session): - instance = SkodaApi(mock_session) - instance.set_vin("test_vin") - instance.headers = {"Authorization": "Bearer test_token"} - return instance - - def test_get_status_success_primary_url(self, skoda_instance, mock_session): - # setup - response_data = { - "carType": "electric", - "totalRangeInKm": 291, - "primaryEngineRange": { - "engineType": "electric", - "currentSoCInPercent": 66, - "remainingRangeInKm": 291 - }, - "carCapturedTimestamp": "2025-10-17T15:40:35.679Z" - } - maintenance_response_data = { - "mileageInKm": 54321 - } - responses = [ - MockAiohttpResponse(response_data, 200), - MockAiohttpResponse(maintenance_response_data, 200) - ] - - async def side_effect_func(*args, **kwargs): - return responses.pop(0) - mock_session.get.side_effect = side_effect_func - - # execution - status = asyncio.run(skoda_instance.get_status()) - - # evaluation - assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 66 - assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 291 - assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2025-10-17T15:40:35Z" - assert status['charging']['batteryStatus']['value']['odometer'] == 54321 - assert mock_session.get.call_count == 2 - mock_session.get.assert_any_call( - "https://mysmob.api.connect.skoda-auto.cz/api/v2/vehicle-status/test_vin/driving-range", - headers=skoda_instance.headers - ) - mock_session.get.assert_any_call( - "https://mysmob.api.connect.skoda-auto.cz/api/v3/vehicle-maintenance/vehicles/test_vin/report", - headers=skoda_instance.headers - ) - - def test_get_status_fallback_to_charging_url(self, skoda_instance, mock_session): - # setup - vehicle_status_response_data = { - "carType": "electric", - "primaryEngineRange": { - "engineType": "electric", - "currentSoCInPercent": 42 - }, - "carCapturedTimestamp": "2025-09-26T11:00:25.848Z" - } - charging_url_response_data = { - "isVehicleInSavedLocation": False, - "status": { - "chargingRateInKilometersPerHour": 0.0, - "chargePowerInKw": 0.0, - "remainingTimeToFullyChargedInMinutes": 0, - "battery": { - "remainingCruisingRangeInMeters": 291000, - "stateOfChargeInPercent": 66 - } - }, - "settings": { - "targetStateOfChargeInPercent": 80, - "preferredChargeMode": "MANUAL", - "availableChargeModes": [ - "MANUAL" - ], - "chargingCareMode": "ACTIVATED", - "autoUnlockPlugWhenCharged": "OFF", - "maxChargeCurrentAc": "MAXIMUM" - }, - "carCapturedTimestamp": "2025-10-17T15:38:55Z", - "errors": [ - { - "type": "STATUS_OF_CHARGING_NOT_AVAILABLE", - "description": "Status of charging is not available." - } - ] - } - maintenance_response_data = { - "mileageInKm": 123456 - } - responses = [ - MockAiohttpResponse(vehicle_status_response_data, 200), - MockAiohttpResponse(charging_url_response_data, 200), - MockAiohttpResponse(maintenance_response_data, 200) - ] - - async def side_effect_func(*args, **kwargs): - return responses.pop(0) - mock_session.get.side_effect = side_effect_func - - # execution - status = asyncio.run(skoda_instance.get_status()) - - # evaluation - assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 66 - assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 291.0 - assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2025-10-17T15:38:55Z" - assert status['charging']['batteryStatus']['value']['odometer'] == 123456 - assert mock_session.get.call_count == 3 - mock_session.get.assert_any_call( - "https://mysmob.api.connect.skoda-auto.cz/api/v2/vehicle-status/test_vin/driving-range", - headers=skoda_instance.headers - ) - mock_session.get.assert_any_call( - "https://mysmob.api.connect.skoda-auto.cz/api/v1/charging/test_vin", - headers=skoda_instance.headers - ) - mock_session.get.assert_any_call( - "https://mysmob.api.connect.skoda-auto.cz/api/v3/vehicle-maintenance/vehicles/test_vin/report", - headers=skoda_instance.headers - ) - - def test_get_status_timestamp_without_milliseconds(self, skoda_instance, mock_session): - # setup - response_data = { - "carType": "hybrid", - "totalRangeInKm": 647, - "primaryEngineRange": { - "engineType": "gasoline", - "currentSoCInPercent": 100, - "currentFuelLevelInPercent": 100, - "remainingRangeInKm": 600 - }, - "secondaryEngineRange": { - "engineType": "electric", - "currentSoCInPercent": 42, - "currentFuelLevelInPercent": 88, - "remainingRangeInKm": 47 - }, - "carCapturedTimestamp": "2025-10-06T10:12:44Z" - } - maintenance_response_data = { - "mileageInKm": 98765 - } - responses = [ - MockAiohttpResponse(response_data, 200), - MockAiohttpResponse(maintenance_response_data, 200) - ] - - async def side_effect_func(*args, **kwargs): - return responses.pop(0) - mock_session.get.side_effect = side_effect_func - - # execution - status = asyncio.run(skoda_instance.get_status()) - - # evaluation - assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 42 - assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 47 - assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2025-10-06T10:12:44Z" - assert status['charging']['batteryStatus']['value']['odometer'] == 98765 - - def test_get_status_fallback_to_charging_url_on_vehicle_status_403(self, skoda_instance, mock_session): - # setup - vehicle_status_error = {"error": "forbidden"} - charging_url_response_data = { - "status": { - "battery": { - "remainingCruisingRangeInMeters": 302000, - "stateOfChargeInPercent": 71 - } - }, - "carCapturedTimestamp": "2026-03-29T09:15:00Z" - } - maintenance_response_data = { - "mileageInKm": 65432 - } - - responses = [ - MockAiohttpResponse(vehicle_status_error, 403), - MockAiohttpResponse(charging_url_response_data, 200), - MockAiohttpResponse(maintenance_response_data, 200) - ] - - async def side_effect_func(*args, **kwargs): - return responses.pop(0) - mock_session.get.side_effect = side_effect_func - - # execution - status = asyncio.run(skoda_instance.get_status()) - - # evaluation - assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 71 - assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 302.0 - assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2026-03-29T09:15:00Z" - assert status['charging']['batteryStatus']['value']['odometer'] == 65432 - assert mock_session.get.call_count == 3 - mock_session.get.assert_any_call( - "https://mysmob.api.connect.skoda-auto.cz/api/v2/vehicle-status/test_vin/driving-range", - headers=skoda_instance.headers - ) - mock_session.get.assert_any_call( - "https://mysmob.api.connect.skoda-auto.cz/api/v1/charging/test_vin", - headers=skoda_instance.headers - ) - mock_session.get.assert_any_call( - "https://mysmob.api.connect.skoda-auto.cz/api/v3/vehicle-maintenance/vehicles/test_vin/report", - headers=skoda_instance.headers - ) diff --git a/packages/modules/vehicles/vwid/README.txt b/packages/modules/vehicles/vweuda/README.txt similarity index 100% rename from packages/modules/vehicles/vwid/README.txt rename to packages/modules/vehicles/vweuda/README.txt diff --git a/packages/modules/vehicles/cupra/__init__.py b/packages/modules/vehicles/vweuda/__init__.py similarity index 100% rename from packages/modules/vehicles/cupra/__init__.py rename to packages/modules/vehicles/vweuda/__init__.py diff --git a/packages/modules/vehicles/vwid/config.py b/packages/modules/vehicles/vweuda/config.py similarity index 63% rename from packages/modules/vehicles/vwid/config.py rename to packages/modules/vehicles/vweuda/config.py index f251c7992b..458ddf9efd 100755 --- a/packages/modules/vehicles/vwid/config.py +++ b/packages/modules/vehicles/vweuda/config.py @@ -1,28 +1,26 @@ from typing import Optional -class VWIdConfiguration: +class VWEUDAConfiguration: 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 VWId: +class VWEUDA: def __init__(self, - name: str = "VWId", - type: str = "vwid", + name: str = "VWEUDA", + type: str = "vweuda", official: bool = False, - configuration: VWIdConfiguration = None) -> None: + configuration: VWEUDAConfiguration = None) -> None: self.name = name self.type = type self.official = official - self.configuration = configuration or VWIdConfiguration() + self.configuration = configuration or VWEUDAConfiguration() diff --git a/packages/modules/vehicles/vwid/libeuda.py b/packages/modules/vehicles/vweuda/libeuda.py similarity index 99% rename from packages/modules/vehicles/vwid/libeuda.py rename to packages/modules/vehicles/vweuda/libeuda.py index adfa8fa286..7081042a9d 100755 --- a/packages/modules/vehicles/vwid/libeuda.py +++ b/packages/modules/vehicles/vweuda/libeuda.py @@ -26,7 +26,7 @@ from collections import deque from helpermodules.constants import RAMDISK_PATH from modules.common.abstract_vehicle import VehicleUpdateData -from modules.vehicles.vwid.config import VWId +from modules.vehicles.vweuda.config import VWEUDA """Constants for the VW EU Data Act integration.""" @@ -149,10 +149,10 @@ def get_oidc_state(brand: str = DEFAULT_BRAND) -> str: EUDA_THREADNAME = "soc_bt_ev" UTC = None KEEP_JSON = 5 -DATA_PATH = Path(__file__).resolve().parents[4] / "data" / "modules" / "vwid" +DATA_PATH = Path(__file__).resolve().parents[4] / "data" / "modules" / "vweuda" JSON_PATH = Path(str(RAMDISK_PATH) + '/vweuda') storeFileName = '/data_' -MODULE_TYPE = 'vwid' +MODULE_TYPE = 'vweuda' # VIN-Brand map VIN_BRAND_MAP = { @@ -803,7 +803,7 @@ def update_result_from_payload(self, payload: dict, source: str) -> dict: if result['soc'] is None: _LOGGER.info("thread result skipped, no soc found") _valid = False - if _valid: + if _valid and result['odometer'] is not None: if vin in euda.result and result['odometer'] < euda.result[vin]['odometer']: _LOGGER.info("odometer less than earlier - keep earlier value") result['odometer'] = euda.result[vin]['odometer'] @@ -881,7 +881,7 @@ async def async_eudaThread(self, username: str, password: str, vin: str, vehicle _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 vwid: terminate now") + _LOGGER.info(f"vehicle {vehicle} is not using module vweuda: terminate now") _active = False continue try: @@ -934,7 +934,7 @@ def check_tests(self): _LOGGER.info(f"test_result, vin={_vin}:\n{_ano_j}") async def get_status(self, - conf: VWId, + conf: VWEUDA, vehicle: int, vehicle_update_data: VehicleUpdateData) -> Union[int, float, str, float, float]: @@ -1078,7 +1078,7 @@ async def get_status(self, # sync function -def fetch_soc(conf: VWId, +def fetch_soc(conf: VWEUDA, vehicle: int, vehicle_update_data: VehicleUpdateData) -> Union[float, float, float, str, float]: diff --git a/packages/modules/vehicles/vwid/soc.py b/packages/modules/vehicles/vweuda/soc.py similarity index 78% rename from packages/modules/vehicles/vwid/soc.py rename to packages/modules/vehicles/vweuda/soc.py index 6e35bef41b..c463c1bd16 100755 --- a/packages/modules/vehicles/vwid/soc.py +++ b/packages/modules/vehicles/vweuda/soc.py @@ -7,20 +7,20 @@ from modules.common.abstract_vehicle import VehicleUpdateData from modules.common.component_state import CarState from modules.common.configurable_vehicle import ConfigurableVehicle -from modules.vehicles.vwid.config import VWId -from modules.vehicles.vwid import libeuda +from modules.vehicles.vweuda.config import VWEUDA +from modules.vehicles.vweuda import libeuda # from modules.vehicles.vwgroup.vwgroup import VwGroup log = logging.getLogger(__name__) -def fetch(vehicle_update_data: VehicleUpdateData, config: VWId, vehicle: int) -> CarState: +def fetch(vehicle_update_data: VehicleUpdateData, config: VWEUDA, vehicle: int) -> CarState: soc, range, soc_ts, soc_tsX, odometer = 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) -def create_vehicle(vehicle_config: VWId, vehicle: int): +def create_vehicle(vehicle_config: VWEUDA, vehicle: int): def updater(vehicle_update_data: VehicleUpdateData) -> CarState: return fetch(vehicle_update_data, vehicle_config, vehicle) @@ -30,4 +30,4 @@ def updater(vehicle_update_data: VehicleUpdateData) -> CarState: calc_while_charging=vehicle_config.configuration.calculate_soc) -device_descriptor = DeviceDescriptor(configuration_factory=VWId) +device_descriptor = DeviceDescriptor(configuration_factory=VWEUDA) diff --git a/packages/modules/vehicles/vwgroup/__init__.py b/packages/modules/vehicles/vwgroup/__init__.py deleted file mode 100755 index e69de29bb2..0000000000 diff --git a/packages/modules/vehicles/vwgroup/socutils.py b/packages/modules/vehicles/vwgroup/socutils.py deleted file mode 100755 index b11cee4dda..0000000000 --- a/packages/modules/vehicles/vwgroup/socutils.py +++ /dev/null @@ -1,70 +0,0 @@ -#!/usr/bin/env python3 - -from typing import Union -import logging -import datetime -import jwt -from helpermodules.pub import Pub - -initialToken = '1.2.3' - -log = logging.getLogger(__name__) - - -class socUtils: - - def __init__(self): - pass - - def read_token_file(self, path: str) -> str: - try: - self.tf = open(path, "r") # try to open Token file - token = self.tf.read() # read token - self.tf.close() - except Exception: - token = None # if no old token found set Token_old to dummy value - return token - - def write_token_file(self, path: str, token: str, config={}): - try: - log.debug("store Token in file " + path) - self.tf = open(path, "w") - self.tf.write(token) # write Token file - self.tf.close() - except Exception as e: - log.exception('Token file write exception ' + str(e)) - - def write_token_mqtt(self, topic: str, token: str, config={}): - try: - config['configuration']['refreshToken'] = token - # log.debug("write_token.mqtt: " + json.dumps(config, ensure_ascii=False, indent=4)) - Pub().pub(topic, config) - except Exception as e: - log.exception('Token mqtt write exception ' + str(e)) - - def get_token_expiration(self, token: str, fmt: str) -> Union[int, str]: - try: - self.token_dec = jwt.decode(token, 'utf-8', options={"verify_signature": False}) - self.exp = self.token_dec['exp'] - self.exp_dt = datetime.datetime.fromtimestamp(self.exp).strftime(fmt) - except Exception as e: - log.exception('get_token_expiration error ' + str(e)) - self.exp = None - self.exp_dt = None - - return self.exp, self.exp_dt - - def keys_exist(self, element, *keys): - # Check if *keys (nested) exists in `element` (dict). - if not isinstance(element, dict): - raise AttributeError('keys_exists() expects dict as first argument.') - if len(keys) == 0: - raise AttributeError('keys_exists() expects at least two arguments, one given.') - - _element = element - for key in keys: - try: - _element = _element[key] - except KeyError: - return False - return True diff --git a/packages/modules/vehicles/vwgroup/vwgroup.py b/packages/modules/vehicles/vwgroup/vwgroup.py deleted file mode 100644 index f6b4307363..0000000000 --- a/packages/modules/vehicles/vwgroup/vwgroup.py +++ /dev/null @@ -1,138 +0,0 @@ -#!/usr/bin/env python3 - -from datetime import datetime -from json import dumps -import logging -from time import mktime, time -from typing import Union -from helpermodules.constants import RAMDISK_PATH -from modules.vehicles.vwgroup.socutils import socUtils - -date_fmt = '%Y-%m-%d %H:%M:%S' -ts_fmt = '%Y-%m-%dT%H:%M:%S' - -refreshToken_exp_days = 7 # 7 days before refreshToken expires a new refreshToken shall be stored -initialToken = '1.2.3' - - -class VwGroup(object): - def __init__(self, conf, vehicle): - self.log = logging.getLogger(__name__) - self.su = socUtils() - self.user_id = conf.configuration.user_id - self.password = conf.configuration.password - self.vin = conf.configuration.vin - self.refreshToken = conf.configuration.refreshToken - self.replyFile = 'soc_' + str(conf.type) + '_reply_vh_' + str(vehicle) - self.accessTokenFile = str(RAMDISK_PATH) + '/soc_' + str(conf.type) + '_accessToken_vh_' + str(vehicle) - self.accessToken_old = {} - self.vehicle = vehicle - self.conf = conf - - # convert utc timestamp to local time - def utc2local(self, utc): - epoch = mktime(utc.timetuple()) - offset = datetime.fromtimestamp(epoch) - datetime.utcfromtimestamp(epoch) - return utc + offset - - # async method, called from sync fetch_soc, required because libvwid/libskoda expect async environment - async def request_data(self, library) -> Union[int, float, str]: - library.set_vin(self.vin) - library.set_credentials(self.user_id, self.password) - library.set_jobs(['charging']) - library.tokens = {} - library.headers = {} - - # initialize refreshToken - try: - if self.refreshToken is None: - self.log.debug("set refreshToken to initial value") - library.tokens['refreshToken'] = initialToken - else: - library.tokens['refreshToken'] = self.refreshToken - - except Exception: - self.log.debug("refreshToken initialization exception: set refreshToken_old to initial value") - library.tokens['refreshToken'] = initialToken - - self.refreshTokenOld = library.tokens['refreshToken'] # remember current refreshToken - - # initialize accessToken - self.accessTokenOld = self.su.read_token_file(self.accessTokenFile) - if self.accessTokenOld is None: - self.log.debug('set accessToken to initial value') - self.accessTokenOld = initialToken - library.tokens['accessToken'] = self.accessTokenOld # initialize tokens in vwid - library.headers['Authorization'] = 'Bearer %s' % library.tokens["accessToken"] - - # get status from VW server - self.data = await library.get_status() - if (self.data): - if self.su.keys_exist(self.data, 'userCapabilities', 'capabilitiesStatus', 'error'): - self.log.error("Server Error: \n" + - dumps(self.data['userCapabilities']['capabilitiesStatus']['error'], - ensure_ascii=False, indent=4)) - - if self.su.keys_exist(self.data, 'charging', 'batteryStatus'): - self.log.debug("batteryStatus: \n" + - dumps(self.data['charging']['batteryStatus'], - ensure_ascii=False, indent=4)) - - try: - self.soc = int(self.data['charging']['batteryStatus']['value']['currentSOC_pct']) - self.range = float(self.data['charging']['batteryStatus']['value']['cruisingRangeElectric_km']) - soc_tsZ = self.data['charging']['batteryStatus']['value']['carCapturedTimestamp'].replace('ZZ', 'Z') - soc_tsdtZ = datetime.strptime(soc_tsZ, ts_fmt + "Z") - soc_tsdtL = self.utc2local(soc_tsdtZ) - self.soc_tsX = datetime.timestamp(soc_tsdtL) - self.soc_ts = datetime.strftime(soc_tsdtL, ts_fmt) - if self.su.keys_exist(self.data, 'charging', 'batteryStatus', 'value', 'odometer'): - self.odometer = float(self.data['charging']['batteryStatus']['value']['odometer']) - else: - self.odometer = None - except Exception as e: - raise Exception("soc/range/soc_ts field missing exception: e=" + str(e)) - - # decision logic - shall a new refreshToken be stored? - self.store_refreshToken = False - self.refreshTokenNew = library.tokens['refreshToken'] - - if self.refreshTokenOld != initialToken: - try: - self.expOld, self.expOld_dt = self.su.get_token_expiration(self.refreshTokenOld, date_fmt) - self.now = int(time()) - expirationThreshold = self.expOld - refreshToken_exp_days * 86400 - - if expirationThreshold < self.now: - self.log.debug('RefreshToken: expiration in less than ' + - str(refreshToken_exp_days) + ' days on ' + self.expOld_dt + ', store new token') - self.store_refreshToken = True - except Exception as e: - self.log.debug("refreshToken decode exception: e=" + str(e)) - self.store_refreshToken = True # no old refreshToken, store new refreshToken anyway - - else: - self.log.debug("Old refreshToken expires on " + self.expOld_dt + ", keep it") - elif self.refreshTokenNew != initialToken: - self.store_refreshToken = True # no old refreshToken, store new refreshToken anyway - - if self.store_refreshToken: # refreshToken needs to be stored in config json - try: - self.expNew, self.expNew_dt = self.su.get_token_expiration(self.refreshTokenNew, date_fmt) - self.log.debug("store new refreshToken, expires on " + self.expNew_dt) - except Exception as e: - self.log.debug("new refreshToken decode exception, e=" + str(e)) - self.log.debug("new refreshToken=" + str(self.refreshTokenNew)) - - confDict = self.conf.__dict__ - confDict.pop('name') - confDict['configuration'] = self.conf.configuration.__dict__ - self.su.write_token_mqtt( - "openWB/set/vehicle/" + self.vehicle + "/soc_module/config", - self.refreshTokenNew, - self.conf.__dict__) - - if (library.tokens['accessToken'] != self.accessTokenOld): # modified accessToken? - self.su.write_token_file(self.accessTokenFile, library.tokens['accessToken']) - - return self.soc, self.range, self.soc_ts, self.soc_tsX, self.odometer diff --git a/packages/modules/vehicles/vwid/.gitignore b/packages/modules/vehicles/vwid/.gitignore deleted file mode 100644 index 1668885149..0000000000 --- a/packages/modules/vehicles/vwid/.gitignore +++ /dev/null @@ -1,2 +0,0 @@ -libvwid.org -libvwid.flake8 diff --git a/packages/modules/vehicles/vwid/__init__.py b/packages/modules/vehicles/vwid/__init__.py deleted file mode 100755 index e69de29bb2..0000000000