diff --git a/.github/workflows/build_and_publish.yml b/.github/workflows/build_and_publish.yml index af4ea3d..fc853ba 100644 --- a/.github/workflows/build_and_publish.yml +++ b/.github/workflows/build_and_publish.yml @@ -10,7 +10,7 @@ jobs: runs-on: ubuntu-latest environment: name: pypi - url: https://pypi.org/p/carconnectivity-connector-cupra + url: https://pypi.org/p/carconnectivity-connector-seatcupra permissions: id-token: write diff --git a/CHANGELOG.md b/CHANGELOG.md index e541576..b01c1aa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,5 +8,5 @@ All notable changes to this project will be documented in this file. ## [0.1] - XXXX-XX-XX Initial release, let's go and give this to the public to try out... -[unreleased]: https://github.com/tillsteinbach/CarConnectivity-connector-cupra/compare/v0.1...HEAD -[0.1]: https://github.com/tillsteinbach/CarConnectivity-connector-cupra/releases/tag/v0.1 +[unreleased]: https://github.com/tillsteinbach/CarConnectivity-connector-seatcupra/compare/v0.1...HEAD +[0.1]: https://github.com/tillsteinbach/CarConnectivity-connector-seatcupra/releases/tag/v0.1 diff --git a/README.md b/README.md index a4fd230..978fa49 100644 --- a/README.md +++ b/README.md @@ -1,28 +1,28 @@ -# CarConnectivity Connector for Cupra Vehicles -[![GitHub sourcecode](https://img.shields.io/badge/Source-GitHub-green)](https://github.com/tillsteinbach/CarConnectivity-connector-cupra/) -[![GitHub release (latest by date)](https://img.shields.io/github/v/release/tillsteinbach/CarConnectivity-connector-cupra)](https://github.com/tillsteinbach/CarConnectivity-connector-cupra/releases/latest) -[![GitHub](https://img.shields.io/github/license/tillsteinbach/CarConnectivity-connector-cupra)](https://github.com/tillsteinbach/CarConnectivity-connector-cupra/blob/master/LICENSE) -[![GitHub issues](https://img.shields.io/github/issues/tillsteinbach/CarConnectivity-connector-cupra)](https://github.com/tillsteinbach/CarConnectivity-connector-cupra/issues) -[![PyPI - Downloads](https://img.shields.io/pypi/dm/carconnectivity-connector-cupra?label=PyPI%20Downloads)](https://pypi.org/project/carconnectivity-connector-cupra/) -[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/carconnectivity-connector-cupra)](https://pypi.org/project/carconnectivity-connector-cupra/) +# CarConnectivity Connector for Seat and Cupra Vehicles +[![GitHub sourcecode](https://img.shields.io/badge/Source-GitHub-green)](https://github.com/tillsteinbach/CarConnectivity-connector-seatcupra/) +[![GitHub release (latest by date)](https://img.shields.io/github/v/release/tillsteinbach/CarConnectivity-connector-seatcupra)](https://github.com/tillsteinbach/CarConnectivity-connector-seatcupra/releases/latest) +[![GitHub](https://img.shields.io/github/license/tillsteinbach/CarConnectivity-connector-seatcupra)](https://github.com/tillsteinbach/CarConnectivity-connector-seatcupra/blob/master/LICENSE) +[![GitHub issues](https://img.shields.io/github/issues/tillsteinbach/CarConnectivity-connector-seatcupra)](https://github.com/tillsteinbach/CarConnectivity-connector-seatcupra/issues) +[![PyPI - Downloads](https://img.shields.io/pypi/dm/carconnectivity-connector-seatcupra?label=PyPI%20Downloads)](https://pypi.org/project/carconnectivity-connector-seatcupra/) +[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/carconnectivity-connector-seatcupra)](https://pypi.org/project/carconnectivity-connector-seatcupra/) [![Donate at PayPal](https://img.shields.io/badge/Donate-PayPal-2997d8)](https://www.paypal.com/donate?hosted_button_id=2BVFF5GJ9SXAJ) [![Sponsor at Github](https://img.shields.io/badge/Sponsor-GitHub-28a745)](https://github.com/sponsors/tillsteinbach) ## Due to lack of access to a Cupra car the development of this conenctor is currently stuck. If you want to help me with access to your account, please contact me! -[CarConnectivity](https://github.com/tillsteinbach/CarConnectivity) is a python API to connect to various car services. This connector enables the integration of cupra vehicles through the MyCupra API. Look at [CarConnectivity](https://github.com/tillsteinbach/CarConnectivity) for other supported brands. +[CarConnectivity](https://github.com/tillsteinbach/CarConnectivity) is a python API to connect to various car services. This connector enables the integration of seat and cupra vehicles through the MyCupra API. Look at [CarConnectivity](https://github.com/tillsteinbach/CarConnectivity) for other supported brands. ## Configuration -In your carconnectivity.json configuration add a section for the cupra connector like this: +In your carconnectivity.json configuration add a section for the seatcupra connector like this: ``` { "carConnectivity": { "connectors": [ { - "type": "cupra", + "type": "seatcupra", "config": { "username": "test@test.de", "password": "testpassword123" @@ -36,7 +36,7 @@ In your carconnectivity.json configuration add a section for the cupra connector If you do not want to provide your username or password inside the configuration you have to create a ".netrc" file at the appropriate location (usually this is your home folder): ``` # For MyCupra -machine cupra +machine seatcupra login test@test.de password testpassword123 ``` @@ -46,7 +46,7 @@ In this case the configuration needs to look like this: "carConnectivity": { "connectors": [ { - "type": "cupra", + "type": "seatcupra", "config": { } } @@ -61,7 +61,7 @@ You can also provide the location of the netrc file in the configuration. "carConnectivity": { "connectors": [ { - "type": "cupra", + "type": "seatcupra", "config": { "netrc": "/some/path/on/your/filesystem" } @@ -73,7 +73,7 @@ You can also provide the location of the netrc file in the configuration. The optional S-PIN needed for some commands can be provided in the account section of the netrc: ``` # For MyCupra -machine cupra +machine seatcupra login test@test.de password testpassword123 account 1234 diff --git a/doc/Config.md b/doc/Config.md index 765dfe4..070107e 100644 --- a/doc/Config.md +++ b/doc/Config.md @@ -1,20 +1,20 @@ -# CarConnectivity Connector for Cupra Config Options +# CarConnectivity Connector for Seat Cupra Config Options The configuration for CarConnectivity is a .json file. -## Cupra Connector Options -These are the valid options for the Cupra Connector +## Seat Cupra Connector Options +These are the valid options for the Seat Cupra Connector ```json { "carConnectivity": { "connectors": [ { - "type": "cupra", // Definition for the Cupra Connector + "type": "seatcupra", // Definition for the Seat Cupra Connector "config": { "log_level": "error", // set the connectos log level "interval": 300, // Interval in which the server is checked in seconds - "username": "test@test.de", // Username of your Cupra Account - "password": "testpassword123", // Username of your Cupra Account + "username": "test@test.de", // Username of your Seat/Cupra Account + "password": "testpassword123", // Username of your Seat/Cupra Account "spin": 1234, //S-Pin used for some special commands like locking/unlocking "netrc": "~/.netr", // netrc file if to be used for passwords "api_log_level": "debug", // Show debug information regarding the API diff --git a/pyproject.toml b/pyproject.toml index 92f52cc..4c09d5e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,8 +6,8 @@ requires = [ build-backend = "setuptools.build_meta" [project] -name = "carconnectivity-connector-cupra" -description = "CarConnectivity connector for Cupra services" +name = "carconnectivity-connector-seatcupra" +description = "CarConnectivity connector for Seat and Cupra services" dynamic = ["version"] requires-python = ">=3.9" authors = [ @@ -36,7 +36,7 @@ classifiers = [ [project.urls] [tool.setuptools_scm] -write_to = "src/carconnectivity_connectors/cupra/_version.py" +write_to = "src/carconnectivity_connectors/seatcupra/_version.py" [tool.pylint.format] max-line-length=160 diff --git a/src/carconnectivity_connectors/cupra/connector.py b/src/carconnectivity_connectors/cupra/connector.py deleted file mode 100644 index 7408140..0000000 --- a/src/carconnectivity_connectors/cupra/connector.py +++ /dev/null @@ -1,430 +0,0 @@ -"""Module implements the connector to interact with the Cupra API.""" -from __future__ import annotations -from typing import TYPE_CHECKING - -import threading - -import json -import os -import logging -import netrc -from datetime import datetime, timezone, timedelta -import requests - -from carconnectivity.garage import Garage -from carconnectivity.errors import AuthenticationError, TooManyRequestsError, RetrievalError, APIError, APICompatibilityError, \ - TemporaryAuthenticationError, SetterError, CommandError -from carconnectivity.util import robust_time_parse, log_extra_keys, config_remove_credentials -from carconnectivity.units import Length, Power, Speed -from carconnectivity.vehicle import GenericVehicle -from carconnectivity.doors import Doors -from carconnectivity.windows import Windows -from carconnectivity.lights import Lights -from carconnectivity.drive import GenericDrive, ElectricDrive, CombustionDrive -from carconnectivity.attributes import BooleanAttribute, DurationAttribute, GenericAttribute, TemperatureAttribute -from carconnectivity.units import Temperature -from carconnectivity.command_impl import ClimatizationStartStopCommand, WakeSleepCommand, HonkAndFlashCommand, LockUnlockCommand, ChargingStartStopCommand -from carconnectivity.climatization import Climatization -from carconnectivity.commands import Commands -from carconnectivity.charging import Charging - -from carconnectivity_connectors.base.connector import BaseConnector -from carconnectivity_connectors.cupra.auth.session_manager import SessionManager, SessionUser, Service -from carconnectivity_connectors.cupra.auth.my_cupra_session import MyCupraSession -from carconnectivity_connectors.cupra._version import __version__ - -SUPPORT_IMAGES = False -try: - from PIL import Image - import base64 - import io - SUPPORT_IMAGES = True - from carconnectivity.attributes import ImageAttribute -except ImportError: - pass - -if TYPE_CHECKING: - from typing import Dict, List, Optional, Any, Union - - from carconnectivity.carconnectivity import CarConnectivity - -LOG: logging.Logger = logging.getLogger("carconnectivity.connectors.cupra") -LOG_API: logging.Logger = logging.getLogger("carconnectivity.connectors.cupra-api-debug") - - -# pylint: disable=too-many-lines -class Connector(BaseConnector): - """ - Connector class for Cupra API connectivity. - Args: - car_connectivity (CarConnectivity): An instance of CarConnectivity. - config (Dict): Configuration dictionary containing connection details. - Attributes: - max_age (Optional[int]): Maximum age for cached data in seconds. - """ - def __init__(self, connector_id: str, car_connectivity: CarConnectivity, config: Dict) -> None: - BaseConnector.__init__(self, connector_id=connector_id, car_connectivity=car_connectivity, config=config, log=LOG, api_log=LOG_API) - - self._background_thread: Optional[threading.Thread] = None - self._stop_event = threading.Event() - - self.connected: BooleanAttribute = BooleanAttribute(name="connected", parent=self, tags={'connector_custom'}) - self.interval: DurationAttribute = DurationAttribute(name="interval", parent=self, tags={'connector_custom'}) - self.commands: Commands = Commands(parent=self) - - LOG.info("Loading cupra connector with config %s", config_remove_credentials(config)) - - if 'spin' in config and config['spin'] is not None: - self.active_config['spin'] = config['spin'] - else: - self.active_config['spin'] = None - - self.active_config['username'] = None - self.active_config['password'] = None - if 'username' in config and 'password' in config: - self.active_config['username'] = config['username'] - self.active_config['password'] = config['password'] - else: - if 'netrc' in config: - self.active_config['netrc'] = config['netrc'] - else: - self.active_config['netrc'] = os.path.join(os.path.expanduser("~"), ".netrc") - try: - secrets = netrc.netrc(file=self.active_config['netrc']) - secret: tuple[str, str, str] | None = secrets.authenticators("cupra") - if secret is None: - raise AuthenticationError(f'Authentication using {self.active_config["netrc"]} failed: cupra not found in netrc') - self.active_config['username'], account, self.active_config['password'] = secret - - if self.active_config['spin'] is None and account is not None: - try: - self.active_config['spin'] = account - except ValueError as err: - LOG.error('Could not parse spin from netrc: %s', err) - except netrc.NetrcParseError as err: - LOG.error('Authentification using %s failed: %s', self.active_config['netrc'], err) - raise AuthenticationError(f'Authentication using {self.active_config["netrc"]} failed: {err}') from err - except TypeError as err: - if 'username' not in config: - raise AuthenticationError(f'"cupra" entry was not found in {self.active_config["netrc"]} netrc-file.' - ' Create it or provide username and password in config') from err - except FileNotFoundError as err: - raise AuthenticationError(f'{self.active_config["netrc"]} netrc-file was not found. Create it or provide username and password in config') \ - from err - - self.active_config['interval'] = 300 - if 'interval' in config: - self.active_config['interval'] = config['interval'] - if self.active_config['interval'] < 180: - raise ValueError('Intervall must be at least 180 seconds') - self.active_config['max_age'] = self.active_config['interval'] - 1 - if 'max_age' in config: - self.active_config['max_age'] = config['max_age'] - self.interval._set_value(timedelta(seconds=self.active_config['interval'])) # pylint: disable=protected-access - - if self.active_config['username'] is None or self.active_config['password'] is None: - raise AuthenticationError('Username or password not provided') - - self._manager: SessionManager = SessionManager(tokenstore=car_connectivity.get_tokenstore(), cache=car_connectivity.get_cache()) - session: requests.Session = self._manager.get_session(Service.MY_CUPRA, SessionUser(username=self.active_config['username'], - password=self.active_config['password'])) - if not isinstance(session, MyCupraSession): - raise AuthenticationError('Could not create session') - self.session: MyCupraSession = session - self.session.retries = 3 - self.session.timeout = 180 - self.session.refresh() - - self._elapsed: List[timedelta] = [] - - def startup(self) -> None: - self._background_thread = threading.Thread(target=self._background_loop, daemon=False) - self._background_thread.start() - - def _background_loop(self) -> None: - self._stop_event.clear() - while not self._stop_event.is_set(): - interval = 300 - try: - try: - self.fetch_all() - self.last_update._set_value(value=datetime.now(tz=timezone.utc)) # pylint: disable=protected-access - if self.interval.value is not None: - interval: float = self.interval.value.total_seconds() - except Exception: - self.connected._set_value(value=False) # pylint: disable=protected-access - if self.interval.value is not None: - interval: float = self.interval.value.total_seconds() - raise - except TooManyRequestsError as err: - LOG.error('Retrieval error during update. Too many requests from your account (%s). Will try again after 15 minutes', str(err)) - self._stop_event.wait(900) - except RetrievalError as err: - LOG.error('Retrieval error during update (%s). Will try again after configured interval of %ss', str(err), interval) - self._stop_event.wait(interval) - except APICompatibilityError as err: - LOG.error('API compatability error during update (%s). Will try again after configured interval of %ss', str(err), interval) - self._stop_event.wait(interval) - except TemporaryAuthenticationError as err: - LOG.error('Temporary authentification error during update (%s). Will try again after configured interval of %ss', str(err), interval) - self._stop_event.wait(interval) - else: - self.connected._set_value(value=True) # pylint: disable=protected-access - self._stop_event.wait(interval) - - def persist(self) -> None: - """ - Persists the current state using the manager's persist method. - - This method calls the `persist` method of the `_manager` attribute to save the current state. - """ - self._manager.persist() - - def shutdown(self) -> None: - """ - Shuts down the connector by persisting current state, closing the session, - and cleaning up resources. - - This method performs the following actions: - 1. Persists the current state. - 2. Closes the session. - 3. Sets the session and manager to None. - 4. Calls the shutdown method of the base connector. - - Returns: - None - """ - # Disable and remove all vehicles managed soley by this connector - for vehicle in self.car_connectivity.garage.list_vehicles(): - if len(vehicle.managing_connectors) == 1 and self in vehicle.managing_connectors: - self.car_connectivity.garage.remove_vehicle(vehicle.id) - vehicle.enabled = False - self._stop_event.set() - if self._background_thread is not None: - self._background_thread.join() - self.persist() - self.session.close() - BaseConnector.shutdown(self) - - def fetch_all(self) -> None: - """ - Fetches all necessary data for the connector. - - This method calls the `fetch_vehicles` method to retrieve vehicle data. - """ - self.fetch_vehicles() - self.car_connectivity.transaction_end() - - def fetch_vehicles(self) -> None: - """ - Fetches the list of vehicles from the Skoda Connect API and updates the garage with new vehicles. - This method sends a request to the Skoda Connect API to retrieve the list of vehicles associated with the user's account. - If new vehicles are found in the response, they are added to the garage. - - Returns: - None - """ - garage: Garage = self.car_connectivity.garage - url = f'https://ola.prod.code.seat.cloud.vwgroup.com/v2/users/{self.session.user_id}/garage/vehicles' - data: Dict[str, Any] | None = self._fetch_data(url, session=self.session) - print(data) - - seen_vehicle_vins: set[str] = set() - if data is not None: - if 'data' in data and data['data'] is not None: - for vehicle_dict in data['data']: - if 'vin' in vehicle_dict and vehicle_dict['vin'] is not None: - seen_vehicle_vins.add(vehicle_dict['vin']) - vehicle: Optional[GenericVehicle] = garage.get_vehicle(vehicle_dict['vin']) # pyright: ignore[reportAssignmentType] - if vehicle is None: - vehicle = GenericVehicle(vin=vehicle_dict['vin'], garage=garage, managing_connector=self) - garage.add_vehicle(vehicle_dict['vin'], vehicle) - - if 'nickname' in vehicle_dict and vehicle_dict['nickname'] is not None: - vehicle.name._set_value(vehicle_dict['nickname']) # pylint: disable=protected-access - else: - vehicle.name._set_value(None) # pylint: disable=protected-access - - if 'model' in vehicle_dict and vehicle_dict['model'] is not None: - vehicle.model._set_value(vehicle_dict['model']) # pylint: disable=protected-access - else: - vehicle.model._set_value(None) # pylint: disable=protected-access - - if 'capabilities' in vehicle_dict and vehicle_dict['capabilities'] is not None: - found_capabilities = set() - for capability_dict in vehicle_dict['capabilities']: - if 'id' in capability_dict and capability_dict['id'] is not None: - capability_id = capability_dict['id'] - found_capabilities.add(capability_id) - if vehicle.capabilities.has_capability(capability_id): - capability: Capability = vehicle.capabilities.get_capability(capability_id) # pyright: ignore[reportAssignmentType] - else: - capability = Capability(capability_id=capability_id, capabilities=vehicle.capabilities) - vehicle.capabilities.add_capability(capability_id, capability) - if 'expirationDate' in capability_dict and capability_dict['expirationDate'] is not None: - expiration_date: datetime = robust_time_parse(capability_dict['expirationDate']) - capability.expiration_date._set_value(expiration_date) # pylint: disable=protected-access - else: - capability.expiration_date._set_value(None) # pylint: disable=protected-access - if 'userDisablingAllowed' in capability_dict and capability_dict['userDisablingAllowed'] is not None: - # pylint: disable-next=protected-access - capability.user_disabling_allowed._set_value(capability_dict['userDisablingAllowed']) - else: - capability.user_disabling_allowed._set_value(None) # pylint: disable=protected-access - else: - raise APIError('Could not fetch capabilities, capability ID missing') - for capability_id in vehicle.capabilities.capabilities.keys() - found_capabilities: - vehicle.capabilities.remove_capability(capability_id) - else: - vehicle.capabilities.clear_capabilities() - - if vehicle.capabilities.has_capability('vehicleWakeUpTrigger'): - if vehicle.commands is not None and vehicle.commands.commands is not None \ - and not vehicle.commands.contains_command('wake-sleep'): - wake_sleep_command = WakeSleepCommand(parent=vehicle.commands) - wake_sleep_command._add_on_set_hook(self.__on_wake_sleep) # pylint: disable=protected-access - wake_sleep_command.enabled = True - vehicle.commands.add_command(wake_sleep_command) - - # Add honkAndFlash command if necessary capabilities are available - if vehicle.capabilities.has_capability('honkAndFlash'): - if vehicle.commands is not None and vehicle.commands.commands is not None \ - and not vehicle.commands.contains_command('honk-flash'): - honk_flash_command = HonkAndFlashCommand(parent=vehicle.commands, with_duration=True) - honk_flash_command._add_on_set_hook(self.__on_honk_flash) # pylint: disable=protected-access - honk_flash_command.enabled = True - vehicle.commands.add_command(honk_flash_command) - - # Add lock and unlock command - if vehicle.capabilities.has_capability('access'): - if vehicle.doors is not None and vehicle.doors.commands is not None and vehicle.doors.commands.commands is not None \ - and not vehicle.doors.commands.contains_command('lock-unlock'): - lock_unlock_command = LockUnlockCommand(parent=vehicle.doors.commands) - lock_unlock_command._add_on_set_hook(self.__on_lock_unlock) # pylint: disable=protected-access - lock_unlock_command.enabled = True - vehicle.doors.commands.add_command(lock_unlock_command) - - self.fetch_vehicle_status(vehicle) - if vehicle.capabilities.has_capability('parkingPosition'): - self.fetch_parking_position(vehicle) - - if SUPPORT_IMAGES: - # fetch vehcile images - url: str = f'https://emea.bff.cariad.digital/media/v2/vehicle-images/{vehicle_dict["vin"]}?resolution=2x' - data = self._fetch_data(url, session=self.session, allow_http_error=True) - if data is not None and 'data' in data: # pylint: disable=too-many-nested-blocks - for image in data['data']: - img = None - cache_date = None - imageurl: str = image['url'] - if self.active_config['max_age'] is not None and self.session.cache is not None and imageurl in self.session.cache: - img, cache_date_string = self.session.cache[imageurl] - img = base64.b64decode(img) # pyright: ignore[reportPossiblyUnboundVariable] - img = Image.open(io.BytesIO(img)) # pyright: ignore[reportPossiblyUnboundVariable] - cache_date = datetime.fromisoformat(cache_date_string) - if img is None or self.active_config['max_age'] is None \ - or (cache_date is not None and cache_date < (datetime.utcnow() - timedelta(seconds=self.active_config['max_age']))): - try: - image_download_response = self.session.get(imageurl, stream=True) - if image_download_response.status_code == requests.codes['ok']: - img = Image.open(image_download_response.raw) # pyright: ignore[reportPossiblyUnboundVariable] - if self.session.cache is not None: - buffered = io.BytesIO() # pyright: ignore[reportPossiblyUnboundVariable] - img.save(buffered, format="PNG") - img_str = base64.b64encode(buffered.getvalue()).decode("utf-8") # pyright: ignore[reportPossiblyUnboundVariable] - self.session.cache[imageurl] = (img_str, str(datetime.utcnow())) - elif image_download_response.status_code == requests.codes['unauthorized']: - LOG.info('Server asks for new authorization') - self.session.login() - image_download_response = self.session.get(imageurl, stream=True) - if image_download_response.status_code == requests.codes['ok']: - img = Image.open(image_download_response.raw) # pyright: ignore[reportPossiblyUnboundVariable] - if self.session.cache is not None: - buffered = io.BytesIO() # pyright: ignore[reportPossiblyUnboundVariable] - img.save(buffered, format="PNG") - img_str = base64.b64encode(buffered.getvalue()).decode("utf-8") # pyright: ignore[reportPossiblyUnboundVariable] - self.session.cache[imageurl] = (img_str, str(datetime.utcnow())) - except requests.exceptions.ConnectionError as connection_error: - raise RetrievalError(f'Connection error: {connection_error}') from connection_error - except requests.exceptions.ChunkedEncodingError as chunked_encoding_error: - raise RetrievalError(f'Error: {chunked_encoding_error}') from chunked_encoding_error - except requests.exceptions.ReadTimeout as timeout_error: - raise RetrievalError(f'Timeout during read: {timeout_error}') from timeout_error - except requests.exceptions.RetryError as retry_error: - raise RetrievalError(f'Retrying failed: {retry_error}') from retry_error - if img is not None: - vehicle._car_images[image['id']] = img # pylint: disable=protected-access - if image['id'] == 'car_34view': - if 'car_picture' in vehicle.images.images: - vehicle.images.images['car_picture']._set_value(img) # pylint: disable=protected-access - else: - vehicle.images.images['car_picture'] = ImageAttribute(name="car_picture", parent=vehicle.images, - value=img, tags={'carconnectivity'}) - else: - raise APIError('Could not fetch vehicle data, VIN missing') - for vin in set(garage.list_vehicle_vins()) - seen_vehicle_vins: - vehicle_to_remove = garage.get_vehicle(vin) - if vehicle_to_remove is not None and vehicle_to_remove.is_managed_by_connector(self): - garage.remove_vehicle(vin) - - def _record_elapsed(self, elapsed: timedelta) -> None: - """ - Records the elapsed time. - - Args: - elapsed (timedelta): The elapsed time to record. - """ - self._elapsed.append(elapsed) - - def _fetch_data(self, url, session, force=False, allow_empty=False, allow_http_error=False, allowed_errors=None) -> Optional[Dict[str, Any]]: # noqa: C901 - data: Optional[Dict[str, Any]] = None - cache_date: Optional[datetime] = None - if not force and (self.active_config['max_age'] is not None and session.cache is not None and url in session.cache): - data, cache_date_string = session.cache[url] - cache_date = datetime.fromisoformat(cache_date_string) - if data is None or self.active_config['max_age'] is None \ - or (cache_date is not None and cache_date < (datetime.utcnow() - timedelta(seconds=self.active_config['max_age']))): - try: - status_response: requests.Response = session.get(url, allow_redirects=False) - self._record_elapsed(status_response.elapsed) - if status_response.status_code in (requests.codes['ok'], requests.codes['multiple_status']): - data = status_response.json() - if session.cache is not None: - session.cache[url] = (data, str(datetime.utcnow())) - elif status_response.status_code == requests.codes['too_many_requests']: - raise TooManyRequestsError('Could not fetch data due to too many requests from your account. ' - f'Status Code was: {status_response.status_code}') - elif status_response.status_code == requests.codes['unauthorized']: - LOG.info('Server asks for new authorization') - session.login() - status_response = session.get(url, allow_redirects=False) - - if status_response.status_code in (requests.codes['ok'], requests.codes['multiple_status']): - data = status_response.json() - if session.cache is not None: - session.cache[url] = (data, str(datetime.utcnow())) - elif not allow_http_error or (allowed_errors is not None and status_response.status_code not in allowed_errors): - raise RetrievalError(f'Could not fetch data even after re-authorization. Status Code was: {status_response.status_code}') - elif not allow_http_error or (allowed_errors is not None and status_response.status_code not in allowed_errors): - print(status_response.request.headers) - raise RetrievalError(f'Could not fetch data. Status Code was: {status_response.status_code}') - except requests.exceptions.ConnectionError as connection_error: - raise RetrievalError(f'Connection error: {connection_error}') from connection_error - except requests.exceptions.ChunkedEncodingError as chunked_encoding_error: - raise RetrievalError(f'Error: {chunked_encoding_error}') from chunked_encoding_error - except requests.exceptions.ReadTimeout as timeout_error: - raise RetrievalError(f'Timeout during read: {timeout_error}') from timeout_error - except requests.exceptions.RetryError as retry_error: - raise RetrievalError(f'Retrying failed: {retry_error}') from retry_error - except requests.exceptions.JSONDecodeError as json_error: - if allow_empty: - data = None - else: - raise RetrievalError(f'JSON decode error: {json_error}') from json_error - return data - - def get_version(self) -> str: - return __version__ - - def get_type(self) -> str: - return "carconnectivity-connector-cupra" \ No newline at end of file diff --git a/src/carconnectivity_connectors/cupra/__init__.py b/src/carconnectivity_connectors/seatcupra/__init__.py similarity index 100% rename from src/carconnectivity_connectors/cupra/__init__.py rename to src/carconnectivity_connectors/seatcupra/__init__.py diff --git a/src/carconnectivity_connectors/cupra/auth/__init__.py b/src/carconnectivity_connectors/seatcupra/auth/__init__.py similarity index 100% rename from src/carconnectivity_connectors/cupra/auth/__init__.py rename to src/carconnectivity_connectors/seatcupra/auth/__init__.py diff --git a/src/carconnectivity_connectors/cupra/auth/auth_util.py b/src/carconnectivity_connectors/seatcupra/auth/auth_util.py similarity index 98% rename from src/carconnectivity_connectors/cupra/auth/auth_util.py rename to src/carconnectivity_connectors/seatcupra/auth/auth_util.py index 718ae7e..405529b 100644 --- a/src/carconnectivity_connectors/cupra/auth/auth_util.py +++ b/src/carconnectivity_connectors/seatcupra/auth/auth_util.py @@ -1,7 +1,7 @@ """ This module provides utility functions and classes for handling authentication and parsing HTML forms -and scripts for the cupra car connectivity connector. +and scripts for the seatcupra car connectivity connector. """ from __future__ import annotations from typing import TYPE_CHECKING diff --git a/src/carconnectivity_connectors/cupra/auth/helpers/blacklist_retry.py b/src/carconnectivity_connectors/seatcupra/auth/helpers/blacklist_retry.py similarity index 100% rename from src/carconnectivity_connectors/cupra/auth/helpers/blacklist_retry.py rename to src/carconnectivity_connectors/seatcupra/auth/helpers/blacklist_retry.py diff --git a/src/carconnectivity_connectors/cupra/auth/my_cupra_session.py b/src/carconnectivity_connectors/seatcupra/auth/my_cupra_session.py similarity index 94% rename from src/carconnectivity_connectors/cupra/auth/my_cupra_session.py rename to src/carconnectivity_connectors/seatcupra/auth/my_cupra_session.py index 4702d49..3ba1530 100644 --- a/src/carconnectivity_connectors/cupra/auth/my_cupra_session.py +++ b/src/carconnectivity_connectors/seatcupra/auth/my_cupra_session.py @@ -19,14 +19,14 @@ from carconnectivity.errors import AuthenticationError, RetrievalError, TemporaryAuthenticationError -from carconnectivity_connectors.cupra.auth.openid_session import AccessType -from carconnectivity_connectors.cupra.auth.vw_web_session import VWWebSession +from carconnectivity_connectors.seatcupra.auth.openid_session import AccessType +from carconnectivity_connectors.seatcupra.auth.vw_web_session import VWWebSession if TYPE_CHECKING: from typing import Tuple, Dict -LOG: logging.Logger = logging.getLogger("carconnectivity.connectors.cupra.auth") +LOG: logging.Logger = logging.getLogger("carconnectivity.connectors.seatcupra.auth") class MyCupraSession(VWWebSession): @@ -93,13 +93,13 @@ def fetch_tokens( if self.token is not None and all(key in self.token for key in ('state', 'id_token', 'access_token', 'code')): # Generate json body for token request body: Dict[str, str] = { - 'state': self.token['state'], - 'id_token': self.token['id_token'], - 'redirect_uri': self.redirect_uri, - 'client_id': self.client_id, - 'client_secret': 'eb8814e641c81a2640ad62eeccec11c98effc9bccd4269ab7af338b50a94b3a2', - 'code': self.token['code'], - 'grant_type': 'authorization_code', + 'state': self.token['state'], + 'id_token': self.token['id_token'], + 'redirect_uri': self.redirect_uri, + 'client_id': self.client_id, + 'client_secret': 'eb8814e641c81a2640ad62eeccec11c98effc9bccd4269ab7af338b50a94b3a2', + 'code': self.token['code'], + 'grant_type': 'authorization_code' } request_headers: CaseInsensitiveDict = dict(self.headers) # pyright: ignore reportAssignmentType diff --git a/src/carconnectivity_connectors/cupra/auth/openid_session.py b/src/carconnectivity_connectors/seatcupra/auth/openid_session.py similarity index 98% rename from src/carconnectivity_connectors/cupra/auth/openid_session.py rename to src/carconnectivity_connectors/seatcupra/auth/openid_session.py index bae9c04..74a8774 100644 --- a/src/carconnectivity_connectors/cupra/auth/openid_session.py +++ b/src/carconnectivity_connectors/seatcupra/auth/openid_session.py @@ -18,13 +18,13 @@ from carconnectivity.errors import AuthenticationError, RetrievalError -from carconnectivity_connectors.cupra.auth.auth_util import add_bearer_auth_header -from carconnectivity_connectors.cupra.auth.helpers.blacklist_retry import BlacklistRetry +from carconnectivity_connectors.seatcupra.auth.auth_util import add_bearer_auth_header +from carconnectivity_connectors.seatcupra.auth.helpers.blacklist_retry import BlacklistRetry if TYPE_CHECKING: from typing import Dict -LOG = logging.getLogger("carconnectivity.connectors.cupra.auth") +LOG = logging.getLogger("carconnectivity.connectors.seatcupra.auth") class AccessType(Enum): diff --git a/src/carconnectivity_connectors/cupra/auth/session_manager.py b/src/carconnectivity_connectors/seatcupra/auth/session_manager.py similarity index 95% rename from src/carconnectivity_connectors/cupra/auth/session_manager.py rename to src/carconnectivity_connectors/seatcupra/auth/session_manager.py index d7f84ec..ffd9c1e 100644 --- a/src/carconnectivity_connectors/cupra/auth/session_manager.py +++ b/src/carconnectivity_connectors/seatcupra/auth/session_manager.py @@ -8,13 +8,13 @@ import logging -from carconnectivity_connectors.cupra.auth.my_cupra_session import MyCupraSession +from carconnectivity_connectors.seatcupra.auth.my_cupra_session import MyCupraSession if TYPE_CHECKING: from typing import Dict, Any - from carconnectivity_connectors.cupra.auth.vw_web_session import VWWebSession + from carconnectivity_connectors.seatcupra.auth.vw_web_session import VWWebSession -LOG = logging.getLogger("carconnectivity.connectors.cupra.auth") +LOG = logging.getLogger("carconnectivity.connectors.seatcupra.auth") class SessionUser(): diff --git a/src/carconnectivity_connectors/cupra/auth/vw_web_session.py b/src/carconnectivity_connectors/seatcupra/auth/vw_web_session.py similarity index 97% rename from src/carconnectivity_connectors/cupra/auth/vw_web_session.py rename to src/carconnectivity_connectors/seatcupra/auth/vw_web_session.py index 49ef3b3..b66b864 100644 --- a/src/carconnectivity_connectors/cupra/auth/vw_web_session.py +++ b/src/carconnectivity_connectors/seatcupra/auth/vw_web_session.py @@ -15,8 +15,8 @@ from carconnectivity.errors import APICompatibilityError, AuthenticationError, RetrievalError -from carconnectivity_connectors.cupra.auth.auth_util import CredentialsFormParser, HTMLFormParser, TermsAndConditionsFormParser -from carconnectivity_connectors.cupra.auth.openid_session import OpenIDSession +from carconnectivity_connectors.seatcupra.auth.auth_util import CredentialsFormParser, HTMLFormParser, TermsAndConditionsFormParser +from carconnectivity_connectors.seatcupra.auth.openid_session import OpenIDSession if TYPE_CHECKING: from typing import Any, Dict diff --git a/src/carconnectivity_connectors/seatcupra/charging.py b/src/carconnectivity_connectors/seatcupra/charging.py new file mode 100644 index 0000000..6378ea9 --- /dev/null +++ b/src/carconnectivity_connectors/seatcupra/charging.py @@ -0,0 +1,74 @@ +""" +Module for charging for Seat/Cupra vehicles. +""" +from __future__ import annotations +from typing import TYPE_CHECKING + +from enum import Enum + +from carconnectivity.charging import Charging +from carconnectivity.vehicle import ElectricVehicle + +if TYPE_CHECKING: + from typing import Optional, Dict + + +class SeatCupraCharging(Charging): # pylint: disable=too-many-instance-attributes + """ + SeatCupraCharging class for handling SeatCupra vehicle charging information. + + This class extends the Charging class and includes an enumeration of various + charging states specific to SeatCupra vehicles. + """ + def __init__(self, vehicle: ElectricVehicle | None = None, origin: Optional[Charging] = None) -> None: + if origin is not None: + super().__init__(origin=origin) + else: + super().__init__(vehicle=vehicle) + + class SeatCupraChargingState(Enum,): + """ + Enum representing the various charging states for a SeatCupra vehicle. + """ + OFF = 'off' + READY_FOR_CHARGING = 'readyForCharging' + NOT_READY_FOR_CHARGING = 'NotReadyForCharging' + CONSERVATION = 'conservation' + CHARGE_PURPOSE_REACHED_NOT_CONSERVATION_CHARGING = 'chargePurposeReachedAndNotConservationCharging' + CHARGE_PURPOSE_REACHED_CONSERVATION = 'chargePurposeReachedAndConservation' + CHARGING = 'charging' + ERROR = 'error' + UNSUPPORTED = 'unsupported' + DISCHARGING = 'discharging' + UNKNOWN = 'unknown charging state' + + class SeatCupraChargeMode(Enum,): + """ + Enum class representing different SeatCupra charge modes. + """ + MANUAL = 'manual' + INVALID = 'invalid' + OFF = 'off' + TIMER = 'timer' + ONLY_OWN_CURRENT = 'onlyOwnCurrent' + PREFERRED_CHARGING_TIMES = 'preferredChargingTimes' + TIMER_CHARGING_WITH_CLIMATISATION = 'timerChargingWithClimatisation' + HOME_STORAGE_CHARGING = 'homeStorageCharging' + IMMEDIATE_DISCHARGING = 'immediateDischarging' + UNKNOWN = 'unknown charge mode' + + +# Mapping of Cupra charging states to generic charging states +mapping_seatcupra_charging_state: Dict[SeatCupraCharging.SeatCupraChargingState, Charging.ChargingState] = { + SeatCupraCharging.SeatCupraChargingState.OFF: Charging.ChargingState.OFF, + SeatCupraCharging.SeatCupraChargingState.NOT_READY_FOR_CHARGING: Charging.ChargingState.OFF, + SeatCupraCharging.SeatCupraChargingState.READY_FOR_CHARGING: Charging.ChargingState.READY_FOR_CHARGING, + SeatCupraCharging.SeatCupraChargingState.CONSERVATION: Charging.ChargingState.CONSERVATION, + SeatCupraCharging.SeatCupraChargingState.CHARGE_PURPOSE_REACHED_NOT_CONSERVATION_CHARGING: Charging.ChargingState.READY_FOR_CHARGING, + SeatCupraCharging.SeatCupraChargingState.CHARGE_PURPOSE_REACHED_CONSERVATION: Charging.ChargingState.CONSERVATION, + SeatCupraCharging.SeatCupraChargingState.CHARGING: Charging.ChargingState.CHARGING, + SeatCupraCharging.SeatCupraChargingState.ERROR: Charging.ChargingState.ERROR, + SeatCupraCharging.SeatCupraChargingState.UNSUPPORTED: Charging.ChargingState.UNSUPPORTED, + SeatCupraCharging.SeatCupraChargingState.DISCHARGING: Charging.ChargingState.DISCHARGING, + SeatCupraCharging.SeatCupraChargingState.UNKNOWN: Charging.ChargingState.UNKNOWN +} diff --git a/src/carconnectivity_connectors/seatcupra/connector.py b/src/carconnectivity_connectors/seatcupra/connector.py new file mode 100644 index 0000000..45c8b4a --- /dev/null +++ b/src/carconnectivity_connectors/seatcupra/connector.py @@ -0,0 +1,686 @@ +"""Module implements the connector to interact with the Seat/Cupra API.""" +from __future__ import annotations +from typing import TYPE_CHECKING + +import threading + +import json +import os +import logging +import netrc +from datetime import datetime, timezone, timedelta +import requests + +from carconnectivity.garage import Garage +from carconnectivity.errors import AuthenticationError, TooManyRequestsError, RetrievalError, APIError, APICompatibilityError, \ + TemporaryAuthenticationError, SetterError, CommandError +from carconnectivity.util import robust_time_parse, log_extra_keys, config_remove_credentials +from carconnectivity.units import Length, Power, Speed +from carconnectivity.vehicle import GenericVehicle, ElectricVehicle, CombustionVehicle, HybridVehicle +from carconnectivity.doors import Doors +from carconnectivity.windows import Windows +from carconnectivity.lights import Lights +from carconnectivity.drive import GenericDrive, ElectricDrive, CombustionDrive +from carconnectivity.attributes import BooleanAttribute, DurationAttribute, GenericAttribute, TemperatureAttribute +from carconnectivity.units import Temperature +from carconnectivity.command_impl import ClimatizationStartStopCommand, WakeSleepCommand, HonkAndFlashCommand, LockUnlockCommand, ChargingStartStopCommand +from carconnectivity.climatization import Climatization +from carconnectivity.commands import Commands +from carconnectivity.charging import Charging +from carconnectivity.position import Position + +from carconnectivity_connectors.base.connector import BaseConnector +from carconnectivity_connectors.seatcupra.auth.session_manager import SessionManager, SessionUser, Service +from carconnectivity_connectors.seatcupra.auth.my_cupra_session import MyCupraSession +from carconnectivity_connectors.seatcupra._version import __version__ +from carconnectivity_connectors.seatcupra.charging import SeatCupraCharging, mapping_seatcupra_charging_state + +SUPPORT_IMAGES = False +try: + from PIL import Image + import base64 + import io + SUPPORT_IMAGES = True + from carconnectivity.attributes import ImageAttribute +except ImportError: + pass + +if TYPE_CHECKING: + from typing import Dict, List, Optional, Any, Union + + from carconnectivity.carconnectivity import CarConnectivity + +LOG: logging.Logger = logging.getLogger("carconnectivity.connectors.seatcupra") +LOG_API: logging.Logger = logging.getLogger("carconnectivity.connectors.seatcupra-api-debug") + + +# pylint: disable=too-many-lines +class Connector(BaseConnector): + """ + Connector class for Seat/Cupra API connectivity. + Args: + car_connectivity (CarConnectivity): An instance of CarConnectivity. + config (Dict): Configuration dictionary containing connection details. + Attributes: + max_age (Optional[int]): Maximum age for cached data in seconds. + """ + def __init__(self, connector_id: str, car_connectivity: CarConnectivity, config: Dict) -> None: + BaseConnector.__init__(self, connector_id=connector_id, car_connectivity=car_connectivity, config=config, log=LOG, api_log=LOG_API) + + self._background_thread: Optional[threading.Thread] = None + self._stop_event = threading.Event() + + self.connected: BooleanAttribute = BooleanAttribute(name="connected", parent=self, tags={'connector_custom'}) + self.interval: DurationAttribute = DurationAttribute(name="interval", parent=self, tags={'connector_custom'}) + self.commands: Commands = Commands(parent=self) + + LOG.info("Loading seatcupra connector with config %s", config_remove_credentials(config)) + + if 'spin' in config and config['spin'] is not None: + self.active_config['spin'] = config['spin'] + else: + self.active_config['spin'] = None + + self.active_config['username'] = None + self.active_config['password'] = None + if 'username' in config and 'password' in config: + self.active_config['username'] = config['username'] + self.active_config['password'] = config['password'] + else: + if 'netrc' in config: + self.active_config['netrc'] = config['netrc'] + else: + self.active_config['netrc'] = os.path.join(os.path.expanduser("~"), ".netrc") + try: + secrets = netrc.netrc(file=self.active_config['netrc']) + secret: tuple[str, str, str] | None = secrets.authenticators("seatcupra") + if secret is None: + raise AuthenticationError(f'Authentication using {self.active_config["netrc"]} failed: seatcupra not found in netrc') + self.active_config['username'], account, self.active_config['password'] = secret + + if self.active_config['spin'] is None and account is not None: + try: + self.active_config['spin'] = account + except ValueError as err: + LOG.error('Could not parse spin from netrc: %s', err) + except netrc.NetrcParseError as err: + LOG.error('Authentification using %s failed: %s', self.active_config['netrc'], err) + raise AuthenticationError(f'Authentication using {self.active_config["netrc"]} failed: {err}') from err + except TypeError as err: + if 'username' not in config: + raise AuthenticationError(f'"seatcupra" entry was not found in {self.active_config["netrc"]} netrc-file.' + ' Create it or provide username and password in config') from err + except FileNotFoundError as err: + raise AuthenticationError(f'{self.active_config["netrc"]} netrc-file was not found. Create it or provide username and password in config') \ + from err + + self.active_config['interval'] = 300 + if 'interval' in config: + self.active_config['interval'] = config['interval'] + if self.active_config['interval'] < 180: + raise ValueError('Intervall must be at least 180 seconds') + self.active_config['max_age'] = self.active_config['interval'] - 1 + if 'max_age' in config: + self.active_config['max_age'] = config['max_age'] + self.interval._set_value(timedelta(seconds=self.active_config['interval'])) # pylint: disable=protected-access + + if self.active_config['username'] is None or self.active_config['password'] is None: + raise AuthenticationError('Username or password not provided') + + self._manager: SessionManager = SessionManager(tokenstore=car_connectivity.get_tokenstore(), cache=car_connectivity.get_cache()) + session: requests.Session = self._manager.get_session(Service.MY_CUPRA, SessionUser(username=self.active_config['username'], + password=self.active_config['password'])) + if not isinstance(session, MyCupraSession): + raise AuthenticationError('Could not create session') + self.session: MyCupraSession = session + self.session.retries = 3 + self.session.timeout = 180 + self.session.refresh() + + self._elapsed: List[timedelta] = [] + + def startup(self) -> None: + self._background_thread = threading.Thread(target=self._background_loop, daemon=False) + self._background_thread.start() + + def _background_loop(self) -> None: + self._stop_event.clear() + fetch: bool = True + while not self._stop_event.is_set(): + interval = 300 + try: + try: + if fetch: + self.fetch_all() + fetch = False + else: + self.update_vehicles() + self.last_update._set_value(value=datetime.now(tz=timezone.utc)) # pylint: disable=protected-access + if self.interval.value is not None: + interval: float = self.interval.value.total_seconds() + except Exception: + if self.interval.value is not None: + interval: float = self.interval.value.total_seconds() + raise + except TooManyRequestsError as err: + LOG.error('Retrieval error during update. Too many requests from your account (%s). Will try again after 15 minutes', str(err)) + self._stop_event.wait(900) + except RetrievalError as err: + LOG.error('Retrieval error during update (%s). Will try again after configured interval of %ss', str(err), interval) + self._stop_event.wait(interval) + except APIError as err: + LOG.error('API error during update (%s). Will try again after configured interval of %ss', str(err), interval) + self._stop_event.wait(interval) + except APICompatibilityError as err: + LOG.error('API compatability error during update (%s). Will try again after configured interval of %ss', str(err), interval) + self._stop_event.wait(interval) + except TemporaryAuthenticationError as err: + LOG.error('Temporary authentification error during update (%s). Will try again after configured interval of %ss', str(err), interval) + self._stop_event.wait(interval) + else: + self._stop_event.wait(interval) + + def persist(self) -> None: + """ + Persists the current state using the manager's persist method. + + This method calls the `persist` method of the `_manager` attribute to save the current state. + """ + self._manager.persist() + + def shutdown(self) -> None: + """ + Shuts down the connector by persisting current state, closing the session, + and cleaning up resources. + + This method performs the following actions: + 1. Persists the current state. + 2. Closes the session. + 3. Sets the session and manager to None. + 4. Calls the shutdown method of the base connector. + + Returns: + None + """ + # Disable and remove all vehicles managed soley by this connector + for vehicle in self.car_connectivity.garage.list_vehicles(): + if len(vehicle.managing_connectors) == 1 and self in vehicle.managing_connectors: + self.car_connectivity.garage.remove_vehicle(vehicle.id) + vehicle.enabled = False + self._stop_event.set() + if self._background_thread is not None: + self._background_thread.join() + self.persist() + self.session.close() + BaseConnector.shutdown(self) + + def fetch_all(self) -> None: + """ + Fetches all necessary data for the connector. + + This method calls the `fetch_vehicles` method to retrieve vehicle data. + """ + self.fetch_vehicles() + self.car_connectivity.transaction_end() + + def update_vehicles(self) -> None: + """ + Updates the status of all vehicles in the garage managed by this connector. + + This method iterates through all vehicle VINs in the garage, and for each vehicle that is + managed by this connector and is an instance of SkodaVehicle, it updates the vehicle's status + by fetching data from various APIs. If the vehicle is an instance of SkodaElectricVehicle, + it also fetches charging information. + + Returns: + None + """ + garage: Garage = self.car_connectivity.garage + for vin in set(garage.list_vehicle_vins()): + vehicle_to_update: Optional[GenericVehicle] = garage.get_vehicle(vin) + if vehicle_to_update is not None and vehicle_to_update.is_managed_by_connector(self): + vehicle_to_update = self.fetch_vehicle_status(vehicle_to_update) + vehicle_to_update = self.fetch_vehicle_mycar_status(vehicle_to_update) + # TODO check for parking capability + vehicle_to_update = self.fetch_parking_position(vehicle_to_update) + + def fetch_vehicles(self) -> None: + """ + Fetches the list of vehicles from the Skoda Connect API and updates the garage with new vehicles. + This method sends a request to the Skoda Connect API to retrieve the list of vehicles associated with the user's account. + If new vehicles are found in the response, they are added to the garage. + + Returns: + None + """ + garage: Garage = self.car_connectivity.garage + url = f'https://ola.prod.code.seat.cloud.vwgroup.com/v2/users/{self.session.user_id}/garage/vehicles' + data: Dict[str, Any] | None = self._fetch_data(url, session=self.session) + + seen_vehicle_vins: set[str] = set() + if data is not None: + if 'vehicles' in data and data['vehicles'] is not None: + for vehicle_dict in data['vehicles']: + if 'vin' in vehicle_dict and vehicle_dict['vin'] is not None: + seen_vehicle_vins.add(vehicle_dict['vin']) + vehicle: Optional[GenericVehicle] = garage.get_vehicle(vehicle_dict['vin']) # pyright: ignore[reportAssignmentType] + if vehicle is None: + vehicle = GenericVehicle(vin=vehicle_dict['vin'], garage=garage, managing_connector=self) + garage.add_vehicle(vehicle_dict['vin'], vehicle) + + if 'vehicleNickname' in vehicle_dict and vehicle_dict['vehicleNickname'] is not None: + vehicle.name._set_value(vehicle_dict['vehicleNickname']) # pylint: disable=protected-access + else: + vehicle.name._set_value(None) # pylint: disable=protected-access + + if 'specifications' in vehicle_dict and vehicle_dict['specifications'] is not None: + if 'steeringRight' in vehicle_dict['specifications'] and vehicle_dict['specifications']['steeringRight'] is not None: + if vehicle_dict['specifications']['steeringRight']: + # pylint: disable-next=protected-access + vehicle.specification.steering_wheel_position._set_value(GenericVehicle.VehicleSpecification.SteeringPosition.RIGHT) + else: + # pylint: disable-next=protected-access + vehicle.specification.steering_wheel_position._set_value(GenericVehicle.VehicleSpecification.SteeringPosition.LEFT) + else: + vehicle.specification.steering_wheel_position._set_value(None) # pylint: disable=protected-access + if 'factoryModel' in vehicle_dict['specifications'] and vehicle_dict['specifications']['factoryModel'] is not None: + factory_model: Dict = vehicle_dict['specifications']['factoryModel'] + if 'vehicleBrand' in factory_model and factory_model['vehicleBrand'] is not None: + vehicle.manufacturer._set_value(factory_model['vehicleBrand']) # pylint: disable=protected-access + else: + vehicle.manufacturer._set_value(None) # pylint: disable=protected-access + if 'vehicleModel' in factory_model and factory_model['vehicleModel'] is not None: + vehicle.model._set_value(factory_model['vehicleModel']) # pylint: disable=protected-access + else: + vehicle.model._set_value(None) # pylint: disable=protected-access + if 'modYear' in factory_model and factory_model['modYear'] is not None: + vehicle.model_year._set_value(factory_model['modYear']) # pylint: disable=protected-access + else: + vehicle.model_year._set_value(None) # pylint: disable=protected-access + log_extra_keys(LOG_API, 'factoryModel', factory_model, {'vehicleBrand', 'vehicleModel', 'modYear'}) + log_extra_keys(LOG_API, 'specifications', vehicle_dict['specifications'], {'steeringRight', 'factoryModel'}) + + + #TODO: https://ola.prod.code.seat.cloud.vwgroup.com/vehicles/{{VIN}}/connection + + #TODO: https://ola.prod.code.seat.cloud.vwgroup.com/v2/vehicles/{{VIN}}/capabilities + else: + raise APIError('Could not fetch vehicle data, VIN missing') + for vin in set(garage.list_vehicle_vins()) - seen_vehicle_vins: + vehicle_to_remove = garage.get_vehicle(vin) + if vehicle_to_remove is not None and vehicle_to_remove.is_managed_by_connector(self): + garage.remove_vehicle(vin) + self.update_vehicles() + + def fetch_vehicle_status(self, vehicle: GenericVehicle, no_cache: bool = False) -> GenericVehicle: + """ + Fetches the status of a vehicle from seat/cupra API. + + Args: + vehicle (GenericVehicle): The vehicle object containing the VIN. + + Returns: + None + """ + vin = vehicle.vin.value + if vin is None: + raise APIError('VIN is missing') + url = f'https://ola.prod.code.seat.cloud.vwgroup.com/v2/vehicles/{vin}/status' + vehicle_status_data: Dict[str, Any] | None = self._fetch_data(url=url, session=self.session, no_cache=no_cache) + if vehicle_status_data: + if 'updatedAt' in vehicle_status_data and vehicle_status_data['updatedAt'] is not None: + captured_at: Optional[datetime] = robust_time_parse(vehicle_status_data['updatedAt']) + else: + captured_at: Optional[datetime] = None + if 'locked' in vehicle_status_data and vehicle_status_data['locked'] is not None: + if vehicle_status_data['locked']: + vehicle.doors.lock_state._set_value(Doors.LockState.LOCKED, measured=captured_at) # pylint: disable=protected-access + else: + vehicle.doors.lock_state._set_value(Doors.LockState.UNLOCKED, measured=captured_at) # pylint: disable=protected-access + if 'lights' in vehicle_status_data and vehicle_status_data['lights'] is not None: + if vehicle_status_data['lights'] == 'on': + vehicle.lights.light_state._set_value(Lights.LightState.ON, measured=captured_at) # pylint: disable=protected-access + elif vehicle_status_data['lights'] == 'off': + vehicle.lights.light_state._set_value(Lights.LightState.OFF, measured=captured_at) # pylint: disable=protected-access + else: + vehicle.lights.light_state._set_value(Lights.LightState.UNKNOWN, measured=captured_at) # pylint: disable=protected-access + LOG_API.info('Unknown lights state %s', vehicle_status_data['lights']) + else: + vehicle.lights.light_state._set_value(None) # pylint: disable=protected-access + + if 'hood' in vehicle_status_data and vehicle_status_data['hood'] is not None: + vehicle_status_data['doors']['hood'] = vehicle_status_data['hood'] + if 'trunk' in vehicle_status_data and vehicle_status_data['trunk'] is not None: + vehicle_status_data['doors']['trunk'] = vehicle_status_data['trunk'] + + if 'doors' in vehicle_status_data and vehicle_status_data['doors'] is not None: + all_doors_closed = True + seen_door_ids: set[str] = set() + for door_id, door_status in vehicle_status_data['doors'].items(): + seen_door_ids.add(door_id) + if door_id in vehicle.doors.doors: + door: Doors.Door = vehicle.doors.doors[door_id] + else: + door = Doors.Door(door_id=door_id, doors=vehicle.doors) + vehicle.doors.doors[door_id] = door + if 'open' in door_status and door_status['open'] is not None: + if door_status['open'] == 'true': + door.open_state._set_value(Doors.OpenState.OPEN, measured=captured_at) # pylint: disable=protected-access + all_doors_closed = False + elif door_status['open'] == 'false': + door.open_state._set_value(Doors.OpenState.CLOSED, measured=captured_at) # pylint: disable=protected-access + else: + door.open_state._set_value(Doors.OpenState.UNKNOWN, measured=captured_at) # pylint: disable=protected-access + LOG_API.info('Unknown door open state %s', door_status['open']) + else: + door.open_state._set_value(None) # pylint: disable=protected-access + if 'locked' in door_status and door_status['locked'] is not None: + if door_status['locked'] == 'true': + door.lock_state._set_value(Doors.LockState.LOCKED, measured=captured_at) # pylint: disable=protected-access + elif door_status['locked'] == 'false': + door.lock_state._set_value(Doors.LockState.UNLOCKED, measured=captured_at) # pylint: disable=protected-access + else: + door.lock_state._set_value(Doors.LockState.UNKNOWN, measured=captured_at) # pylint: disable=protected-access + LOG_API.info('Unknown door lock state %s', door_status['locked']) + else: + door.lock_state._set_value(None) # pylint: disable=protected-access + log_extra_keys(LOG_API, 'door', door_status, {'open', 'locked'}) + for door_id in vehicle.doors.doors.keys() - seen_door_ids: + vehicle.doors.doors[door_id].enabled = False + if all_doors_closed: + vehicle.doors.open_state._set_value(Doors.OpenState.CLOSED, measured=captured_at) # pylint: disable=protected-access + else: + vehicle.doors.open_state._set_value(Doors.OpenState.OPEN, measured=captured_at) # pylint: disable=protected-access + seen_window_ids: set[str] = set() + if 'windows' in vehicle_status_data and vehicle_status_data['windows'] is not None: + all_windows_closed = True + for window_id, window_status in vehicle_status_data['windows'].items(): + seen_window_ids.add(window_id) + if window_id in vehicle.windows.windows: + window: Windows.Window = vehicle.windows.windows[window_id] + else: + window = Windows.Window(window_id=window_id, windows=vehicle.windows) + vehicle.windows.windows[window_id] = window + if window_status in Windows.OpenState: + open_state: Windows.OpenState = Windows.OpenState(window_status) + if open_state == Windows.OpenState.OPEN: + all_windows_closed = False + window.open_state._set_value(open_state, measured=captured_at) # pylint: disable=protected-access + else: + LOG_API.info('Unknown window status %s', window_status) + window.open_state._set_value(Windows.OpenState.UNKNOWN, measured=captured_at) # pylint: disable=protected-access + if all_windows_closed: + vehicle.windows.open_state._set_value(Windows.OpenState.CLOSED, measured=captured_at) # pylint: disable=protected-access + else: + vehicle.windows.open_state._set_value(Windows.OpenState.OPEN, measured=captured_at) # pylint: disable=protected-access + else: + vehicle.windows.open_state._set_value(None) # pylint: disable=protected-access + for window_id in vehicle.windows.windows.keys() - seen_window_ids: + vehicle.windows.windows[window_id].enabled = False + log_extra_keys(LOG_API, f'/api/v2/vehicle-status/{vin}', vehicle_status_data, {'updatedAt', 'locked', 'lights', 'hood', 'trunk', 'doors', + 'windows'}) + return vehicle + + def fetch_vehicle_mycar_status(self, vehicle: GenericVehicle, no_cache: bool = False) -> GenericVehicle: + """ + Fetches the status of a vehicle from seat/cupra API. + + Args: + vehicle (GenericVehicle): The vehicle object containing the VIN. + + Returns: + None + """ + vin = vehicle.vin.value + if vin is None: + raise APIError('VIN is missing') + url = f'https://ola.prod.code.seat.cloud.vwgroup.com/v5/users/{self.session.user_id}/vehicles/{vin}/mycar' + vehicle_status_data: Dict[str, Any] | None = self._fetch_data(url=url, session=self.session, no_cache=no_cache) + if vehicle_status_data: + if 'engines' in vehicle_status_data and vehicle_status_data['engines'] is not None: + drive_ids: set[str] = {'primary', 'secondary'} + total_range: float = 0.0 + for drive_id in drive_ids: + if drive_id in vehicle_status_data['engines'] and vehicle_status_data['engines'][drive_id] is not None \ + and 'fuelType' in vehicle_status_data['engines'][drive_id] and vehicle_status_data['engines'][drive_id]['fuelType'] is not None: + try: + engine_type: GenericDrive.Type = GenericDrive.Type(vehicle_status_data['engines'][drive_id]['fuelType']) + except ValueError: + LOG_API.warning('Unknown fuelType type %s', vehicle_status_data['engines'][drive_id]['fuelType']) + engine_type: GenericDrive.Type = GenericDrive.Type.UNKNOWN + + if drive_id in vehicle.drives.drives: + drive: GenericDrive = vehicle.drives.drives[drive_id] + else: + if engine_type == GenericDrive.Type.ELECTRIC: + drive = ElectricDrive(drive_id=drive_id, drives=vehicle.drives) + elif engine_type in [GenericDrive.Type.FUEL, + GenericDrive.Type.GASOLINE, + GenericDrive.Type.PETROL, + GenericDrive.Type.DIESEL, + GenericDrive.Type.CNG, + GenericDrive.Type.LPG]: + drive = CombustionDrive(drive_id=drive_id, drives=vehicle.drives) + else: + drive = GenericDrive(drive_id=drive_id, drives=vehicle.drives) + drive.type._set_value(engine_type) # pylint: disable=protected-access + vehicle.drives.add_drive(drive) + if 'levelPct' in vehicle_status_data['engines'][drive_id] and vehicle_status_data['engines'][drive_id]['levelPct'] is not None: + # pylint: disable-next=protected-access + drive.level._set_value(value=vehicle_status_data['engines'][drive_id]['levelPct']) + else: + drive.level._set_value(None) # pylint: disable=protected-access + if 'rangeKm' in vehicle_status_data['engines'][drive_id] and vehicle_status_data['engines'][drive_id]['rangeKm'] is not None: + # pylint: disable-next=protected-access + drive.range._set_value(value=vehicle_status_data['engines'][drive_id]['rangeKm'], unit=Length.KM) + total_range += vehicle_status_data['engines'][drive_id]['rangeKm'] + else: + drive.range._set_value(None, unit=Length.KM) # pylint: disable=protected-access + log_extra_keys(LOG_API, drive_id, vehicle_status_data['engines'][drive_id], {'fuelType', + 'levelPct', + 'rangeKm'}) + vehicle.drives.total_range._set_value(total_range, unit=Length.KM) # pylint: disable=protected-access + else: + vehicle.drives.enabled = False + if len(vehicle.drives.drives) > 0: + has_electric = False + has_combustion = False + for drive in vehicle.drives.drives.values(): + if isinstance(drive, ElectricDrive): + has_electric = True + elif isinstance(drive, CombustionDrive): + has_combustion = True + if has_electric and not has_combustion and not isinstance(vehicle, ElectricVehicle): + LOG.debug('Promoting %s to ElectricVehicle object for %s', vehicle.__class__.__name__, vin) + vehicle = ElectricVehicle(origin=vehicle) + self.car_connectivity.garage.replace_vehicle(vin, vehicle) + elif has_combustion and not has_electric and not isinstance(vehicle, CombustionVehicle): + LOG.debug('Promoting %s to CombustionVehicle object for %s', vehicle.__class__.__name__, vin) + vehicle = CombustionVehicle(origin=vehicle) + self.car_connectivity.garage.replace_vehicle(vin, vehicle) + elif has_combustion and has_electric and not isinstance(vehicle, HybridVehicle): + LOG.debug('Promoting %s to HybridVehicle object for %s', vehicle.__class__.__name__, vin) + vehicle = HybridVehicle(origin=vehicle) + self.car_connectivity.garage.replace_vehicle(vin, vehicle) + if 'services' in vehicle_status_data and vehicle_status_data['services'] is not None: + if 'charging' in vehicle_status_data['services'] and vehicle_status_data['services']['charging'] is not None: + charging_status: Dict = vehicle_status_data['services']['charging'] + if 'status' in charging_status and charging_status['status'] is not None: + if charging_status['status'] in SeatCupraCharging.SeatCupraChargingState: + volkswagen_charging_state = SeatCupraCharging.SeatCupraChargingState(charging_status['status']) + charging_state: Charging.ChargingState = mapping_seatcupra_charging_state[volkswagen_charging_state] + else: + LOG_API.info('Unkown charging state %s not in %s', charging_status['status'], + str(SeatCupraCharging.SeatCupraChargingState)) + charging_state = Charging.ChargingState.UNKNOWN + if isinstance(vehicle, ElectricVehicle): + vehicle.charging.state._set_value(value=charging_state) # pylint: disable=protected-access + else: + LOG_API.warning('Vehicle is not an electric or hybrid vehicle, but charging state was fetched') + else: + if isinstance(vehicle, ElectricVehicle): + vehicle.charging.state._set_value(None) # pylint: disable=protected-access + else: + LOG_API.warning('Vehicle is not an electric or hybrid vehicle, but charging state was fetched') + if 'targetPct' in charging_status and charging_status['targetPct'] is not None: + if isinstance(vehicle, ElectricVehicle): + vehicle.charging.settings.target_level._set_value(charging_status['targetPct']) # pylint: disable=protected-access + if 'chargeMode' in charging_status and charging_status['chargeMode'] is not None: + if charging_status['chargeMode'] in Charging.ChargingType: + if isinstance(vehicle, ElectricVehicle): + vehicle.charging.type._set_value(value=Charging.ChargingType(charging_status['chargeMode'])) # pylint: disable=protected-access + else: + LOG_API.info('Unknown charge type %s', charging_status['chargeMode']) + if isinstance(vehicle, ElectricVehicle): + vehicle.charging.type._set_value(Charging.ChargingType.UNKNOWN) # pylint: disable=protected-access + else: + if isinstance(vehicle, ElectricVehicle): + vehicle.charging.type._set_value(None) # pylint: disable=protected-access + if 'remainingTime' in charging_status and charging_status['remainingTime'] is not None: + remaining_duration: timedelta = timedelta(minutes=charging_status['remainingTime']) + estimated_date_reached: datetime = datetime.now(tz=timezone.utc) + remaining_duration + estimated_date_reached = estimated_date_reached.replace(second=0, microsecond=0) + if isinstance(vehicle, ElectricVehicle): + vehicle.charging.estimated_date_reached._set_value(value=estimated_date_reached) # pylint: disable=protected-access + else: + if isinstance(vehicle, ElectricVehicle): + vehicle.charging.estimated_date_reached._set_value(None) # pylint: disable=protected-access + log_extra_keys(LOG_API, 'charging', charging_status, {'status', 'targetPct', 'currentPct', 'chargeMode', 'remainingTime'}) + else: + if isinstance(vehicle, ElectricVehicle): + vehicle.charging.enabled = False + if 'climatisation' in vehicle_status_data['services'] and vehicle_status_data['services']['climatisation'] is not None: + climatisation_status: Dict = vehicle_status_data['services']['climatisation'] + if 'status' in climatisation_status and climatisation_status['status'] is not None: + if climatisation_status['status'].lower() in Climatization.ClimatizationState: + climatization_state: Climatization.ClimatizationState = Climatization.ClimatizationState(climatisation_status['status'].lower()) + else: + LOG_API.info('Unknown climatization state %s not in %s', climatisation_status['status'], + str(Climatization.ClimatizationState)) + climatization_state = Climatization.ClimatizationState.UNKNOWN + vehicle.climatization.state._set_value(value=climatization_state) # pylint: disable=protected-access + else: + vehicle.climatization.state._set_value(None) # pylint: disable=protected-access + if 'targetTemperatureCelsius' in climatisation_status and climatisation_status['targetTemperatureCelsius'] is not None: + target_temperature: Optional[float] = climatisation_status['targetTemperatureCelsius'] + vehicle.climatization.settings.target_temperature._set_value(value=target_temperature, # pylint: disable=protected-access + unit=Temperature.C) + elif 'targetTemperatureFahrenheit' in climatisation_status and climatisation_status['targetTemperatureFahrenheit'] is not None: + target_temperature = climatisation_status['targetTemperatureFahrenheit'] + vehicle.climatization.settings.target_temperature._set_value(value=target_temperature, # pylint: disable=protected-access + unit=Temperature.F) + else: + vehicle.climatization.settings.target_temperature._set_value(None) # pylint: disable=protected-access + if 'remainingTime' in climatisation_status and climatisation_status['remainingTime'] is not None: + remaining_duration: timedelta = timedelta(minutes=climatisation_status['remainingTime']) + estimated_date_reached: datetime = datetime.now(tz=timezone.utc) + remaining_duration + estimated_date_reached = estimated_date_reached.replace(second=0, microsecond=0) + vehicle.charging.estimated_date_reached._set_value(value=estimated_date_reached) # pylint: disable=protected-access + else: + vehicle.charging.estimated_date_reached._set_value(None) # pylint: disable=protected-access + log_extra_keys(LOG_API, 'climatisation', climatisation_status, {'status', 'targetTemperatureCelsius', 'targetTemperatureFahrenheit', + 'remainingTime'}) + return vehicle + + def fetch_parking_position(self, vehicle: GenericVehicle, no_cache: bool = False) -> GenericVehicle: + """ + Fetches the position of the given vehicle and updates its position attributes. + + Args: + vehicle (SkodaVehicle): The vehicle object containing the VIN and position attributes. + + Returns: + SkodaVehicle: The updated vehicle object with the fetched position data. + + Raises: + APIError: If the VIN is missing. + ValueError: If the vehicle has no position object. + """ + vin = vehicle.vin.value + if vin is None: + raise APIError('VIN is missing') + if vehicle.position is None: + raise ValueError('Vehicle has no charging object') + url = f'https://ola.prod.code.seat.cloud.vwgroup.com/v1/vehicles/{vin}/parkingposition' + data: Dict[str, Any] | None = self._fetch_data(url=url, session=self.session, no_cache=no_cache) + if data is not None: + if 'lat' in data and data['lat'] is not None: + latitude: Optional[float] = data['lat'] + else: + latitude = None + if 'lon' in data and data['lon'] is not None: + longitude: Optional[float] = data['lon'] + else: + longitude = None + vehicle.position.latitude._set_value(latitude) # pylint: disable=protected-access + vehicle.position.longitude._set_value(longitude) # pylint: disable=protected-access + vehicle.position.position_type._set_value(Position.PositionType.PARKING) # pylint: disable=protected-access + log_extra_keys(LOG_API, 'parkingposition', data, {'lat', 'lon'}) + else: + vehicle.position.latitude._set_value(None) # pylint: disable=protected-access + vehicle.position.longitude._set_value(None) # pylint: disable=protected-access + vehicle.position.position_type._set_value(None) # pylint: disable=protected-access + return vehicle + + def _record_elapsed(self, elapsed: timedelta) -> None: + """ + Records the elapsed time. + + Args: + elapsed (timedelta): The elapsed time to record. + """ + self._elapsed.append(elapsed) + + def _fetch_data(self, url, session, no_cache=False, allow_empty=False, allow_http_error=False, + allowed_errors=None) -> Optional[Dict[str, Any]]: # noqa: C901 + data: Optional[Dict[str, Any]] = None + cache_date: Optional[datetime] = None + if not no_cache and (self.active_config['max_age'] is not None and session.cache is not None and url in session.cache): + data, cache_date_string = session.cache[url] + cache_date = datetime.fromisoformat(cache_date_string) + if data is None or self.active_config['max_age'] is None \ + or (cache_date is not None and cache_date < (datetime.utcnow() - timedelta(seconds=self.active_config['max_age']))): + try: + status_response: requests.Response = session.get(url, allow_redirects=False) + self._record_elapsed(status_response.elapsed) + if status_response.status_code in (requests.codes['ok'], requests.codes['multiple_status']): + data = status_response.json() + if session.cache is not None: + session.cache[url] = (data, str(datetime.utcnow())) + elif status_response.status_code == requests.codes['too_many_requests']: + raise TooManyRequestsError('Could not fetch data due to too many requests from your account. ' + f'Status Code was: {status_response.status_code}') + elif status_response.status_code == requests.codes['unauthorized']: + LOG.info('Server asks for new authorization') + session.login() + status_response = session.get(url, allow_redirects=False) + + if status_response.status_code in (requests.codes['ok'], requests.codes['multiple_status']): + data = status_response.json() + if session.cache is not None: + session.cache[url] = (data, str(datetime.utcnow())) + elif not allow_http_error or (allowed_errors is not None and status_response.status_code not in allowed_errors): + raise RetrievalError(f'Could not fetch data even after re-authorization. Status Code was: {status_response.status_code}') + elif not allow_http_error or (allowed_errors is not None and status_response.status_code not in allowed_errors): + raise RetrievalError(f'Could not fetch data. Status Code was: {status_response.status_code}') + except requests.exceptions.ConnectionError as connection_error: + raise RetrievalError(f'Connection error: {connection_error}.' + ' If this happens frequently, please check if other applications communicate with the Skoda server.') from connection_error + except requests.exceptions.ChunkedEncodingError as chunked_encoding_error: + raise RetrievalError(f'Error: {chunked_encoding_error}') from chunked_encoding_error + except requests.exceptions.ReadTimeout as timeout_error: + raise RetrievalError(f'Timeout during read: {timeout_error}') from timeout_error + except requests.exceptions.RetryError as retry_error: + raise RetrievalError(f'Retrying failed: {retry_error}') from retry_error + except requests.exceptions.JSONDecodeError as json_error: + if allow_empty: + data = None + else: + raise RetrievalError(f'JSON decode error: {json_error}') from json_error + return data + + def get_version(self) -> str: + return __version__ + + def get_type(self) -> str: + return "carconnectivity-connector-seatcupra" diff --git a/src/carconnectivity_connectors/cupra/ui/connector_ui.py b/src/carconnectivity_connectors/seatcupra/ui/connector_ui.py similarity index 68% rename from src/carconnectivity_connectors/cupra/ui/connector_ui.py rename to src/carconnectivity_connectors/seatcupra/ui/connector_ui.py index 9a29178..4f02ee4 100644 --- a/src/carconnectivity_connectors/cupra/ui/connector_ui.py +++ b/src/carconnectivity_connectors/seatcupra/ui/connector_ui.py @@ -1,4 +1,4 @@ -""" User interface for the cupra connector in the Car Connectivity application. """ +""" User interface for the seatcupra connector in the Car Connectivity application. """ from __future__ import annotations from typing import TYPE_CHECKING @@ -16,16 +16,16 @@ class ConnectorUI(BaseConnectorUI): """ - A user interface class for the Cupra connector in the Car Connectivity application. + A user interface class for the Seat/Cupra connector in the Car Connectivity application. """ def __init__(self, connector: BaseConnector): - blueprint: Optional[flask.Blueprint] = flask.Blueprint(name='cupra', import_name='carconnectivity-connector-cupra', url_prefix='/cupra', + blueprint: Optional[flask.Blueprint] = flask.Blueprint(name='seatcupra', import_name='carconnectivity-connector-seatcupra', url_prefix='/seatcupra', template_folder=os.path.dirname(__file__) + '/templates') super().__init__(connector, blueprint=blueprint) def get_nav_items(self) -> List[Dict[Literal['text', 'url', 'sublinks', 'divider'], Union[str, List]]]: """ - Generates a list of navigation items for the Cupra connector UI. + Generates a list of navigation items for the Seat/Cupra connector UI. """ return super().get_nav_items() @@ -34,6 +34,6 @@ def get_title(self) -> str: Returns the title of the connector. Returns: - str: The title of the connector, which is "Cupra". + str: The title of the connector, which is "Seat/Cupra". """ - return "Cupra" + return "Seat/Cupra" diff --git a/test/integration_test/carConnectivity.json b/test/integration_test/carConnectivity.json index ccf5dd7..0cf8fb6 100644 --- a/test/integration_test/carConnectivity.json +++ b/test/integration_test/carConnectivity.json @@ -3,7 +3,7 @@ "log_level": "debug", "connectors": [ { - "type": "cupra", + "type": "seatcupra", "disabled": false, "config": { "log_level": "debug",