Refactor and simplify using DataUpdateCoordinator

custom_components/polestar_api/__init__.py

@@ -1,16 +1,23 @@
 """Polestar EV integration."""
 
+from __future__ import annotations
+
 import logging
+from typing import TYPE_CHECKING
 
-import httpx
 from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
-from homeassistant.core import HomeAssistant
+from homeassistant.helpers.httpx_client import create_async_httpx_client
 from homeassistant.loader import async_get_loaded_integration
-from pypolestar.exceptions import PolestarApiException, PolestarAuthException
+from pypolestar import PolestarApi
 
 from .const import CONF_VIN
-from .data import PolestarConfigEntry, PolestarData
-from .polestar import PolestarCar, PolestarCoordinator
+from .coordinator import PolestarCoordinator
+from .data import PolestarData
+
+if TYPE_CHECKING:
+    from homeassistant.core import HomeAssistant
+
+    from .data import PolestarConfigEntry
 
 PLATFORMS = [Platform.IMAGE, Platform.SENSOR, Platform.BINARY_SENSOR]
 
@@ -22,49 +29,53 @@ async def async_setup_entry(hass: HomeAssistant, entry: PolestarConfigEntry) ->
 
     _LOGGER.debug("async_setup_entry: %s", entry)
 
-    coordinator = PolestarCoordinator(
-        hass=hass,
+    vin = entry.data.get(CONF_VIN)
+
+    api_client = PolestarApi(
         username=entry.data[CONF_USERNAME],
         password=entry.data[CONF_PASSWORD],
-        vin=entry.data.get(CONF_VIN),
+        client_session=create_async_httpx_client(hass),
+        vins=[vin] if vin else None,
         unique_id=entry.entry_id,
     )
 
-    try:
-        await coordinator.async_init()
-
-        cars: list[PolestarCar] = []
-        for car in coordinator.get_cars():
-            await car.async_update()
-            cars.append(car)
-            _LOGGER.debug(
-                "Added car with VIN %s for %s",
-                car.vin,
-                entry.entry_id,
-            )
-
-        entry.runtime_data = PolestarData(
-            coordinator=coordinator,
-            cars=cars,
-            integration=async_get_loaded_integration(hass, entry.domain),
+    await api_client.async_init()
+
+    coordinators = []
+
+    for coordinator in [
+        PolestarCoordinator(
+            hass=hass,
+            api=api_client,
+            config_entry=entry,
+            vin=vin,
         )
+        for vin in api_client.get_available_vins()
+    ]:
+        await coordinator.async_config_entry_first_refresh()
+        coordinators.append(coordinator)
+        _LOGGER.debug(
+            "Added car with VIN %s for %s",
+            coordinator.vin,
+            entry.entry_id,
+        )
+
+    entry.runtime_data = PolestarData(
+        api_client=api_client,
+        coordinators=coordinators,
+        integration=async_get_loaded_integration(hass, entry.domain),
+    )
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    entry.async_on_unload(entry.add_update_listener(async_reload_entry))
+
+    return True
+
 
-        await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
-        return True
-    except PolestarApiException as e:
-        _LOGGER.exception("API Exception on update data %s", str(e))
-    except PolestarAuthException as e:
-        _LOGGER.exception("Auth Exception on update data %s", str(e))
-    except httpx.ConnectTimeout as e:
-        _LOGGER.exception("Connection Timeout on update data %s", str(e))
-    except httpx.ConnectError as e:
-        _LOGGER.exception("Connection Error on update data %s", str(e))
-    except httpx.ReadTimeout as e:
-        _LOGGER.exception("Read Timeout on update data %s", str(e))
-    except Exception as e:
-        _LOGGER.exception("Unexpected Error on update data %s", str(e))
-    coordinator.polestar_api.latest_call_code = 500
-    return False
+async def async_reload_entry(hass: HomeAssistant, entry: PolestarConfigEntry) -> None:
+    """Reload config entry."""
+    await async_unload_entry(hass, entry)
+    await async_setup_entry(hass, entry)
 
 
 async def async_unload_entry(hass: HomeAssistant, entry: PolestarConfigEntry) -> bool:

custom_components/polestar_api/binary_sensor.py

@@ -1,21 +1,25 @@
 """Support for Polestar binary sensors."""
 
+from __future__ import annotations
+
 import logging
-from typing import Final
+from typing import TYPE_CHECKING, Final
 
 from homeassistant.components.binary_sensor import (
     BinarySensorDeviceClass,
     BinarySensorEntity,
     BinarySensorEntityDescription,
 )
 from homeassistant.const import EntityCategory
-from homeassistant.core import HomeAssistant
-from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from .const import DOMAIN as POLESTAR_API_DOMAIN
-from .data import PolestarConfigEntry
 from .entity import PolestarEntity
-from .polestar import PolestarCar
+
+if TYPE_CHECKING:
+    from homeassistant.core import HomeAssistant
+    from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+    from .coordinator import PolestarCoordinator
+    from .data import PolestarConfigEntry
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -38,11 +42,11 @@ async def async_setup_entry(
     """Set up the binary_sensor platform."""
     async_add_entities(
         PolestarBinarySensor(
-            car=car,
+            coordinator=coordinator,
             entity_description=entity_description,
         )
         for entity_description in ENTITY_DESCRIPTIONS
-        for car in entry.runtime_data.cars
+        for coordinator in entry.runtime_data.coordinators
     )
 
 
@@ -51,20 +55,13 @@ class PolestarBinarySensor(PolestarEntity, BinarySensorEntity):
 
     def __init__(
         self,
-        car: PolestarCar,
+        coordinator: PolestarCoordinator,
         entity_description: BinarySensorEntityDescription,
     ) -> None:
         """Initialize the binary_sensor class."""
-        super().__init__(car)
-        self.car = car
-        self.entity_description = entity_description
-        self.entity_id = f"{POLESTAR_API_DOMAIN}.'polestar_'.{car.get_short_id()}_{entity_description.key}"
-        self._attr_unique_id = (
-            f"polestar_{car.get_unique_id()}_{entity_description.key}"
-        )
-        self._attr_translation_key = f"polestar_{entity_description.key}"
+        super().__init__(coordinator, entity_description)
 
     @property
     def is_on(self) -> bool | None:
         """Return true if the binary_sensor is on."""
-        return self.car.data.get(self.entity_description.key)
+        return self.coordinator.data.get(self.entity_description.key)

custom_components/polestar_api/const.py

@@ -4,6 +4,8 @@
 from typing import Final
 
 DOMAIN = "polestar_api"
+ATTRIBUTION = "Data provided by https://polestar.com/"
+
 TIMEOUT = 90
 
 DEFAULT_SCAN_INTERVAL = timedelta(seconds=60)

custom_components/polestar_api/coordinator.py

@@ -0,0 +1,200 @@
+"""Polestar API for Polestar integration."""
+
+from __future__ import annotations
+
+import logging
+import re
+from datetime import datetime, timedelta
+from typing import TYPE_CHECKING, Any
+
+import homeassistant.util.dt as dt_util
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+from pypolestar.exceptions import PolestarApiException, PolestarAuthException
+
+from .const import DEFAULT_SCAN_INTERVAL
+
+if TYPE_CHECKING:
+    from homeassistant.core import HomeAssistant
+    from pypolestar import PolestarApi
+
+    from .data import PolestarConfigEntry
+
+_LOGGER = logging.getLogger(__name__)
+
+
+# https://developers.home-assistant.io/docs/integration_fetching_data#coordinated-single-api-poll-for-data-for-all-entities
+class PolestarCoordinator(DataUpdateCoordinator):
+    """Polestar EV integration."""
+
+    config_entry: PolestarConfigEntry
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        api: PolestarApi,
+        config_entry: PolestarConfigEntry,
+        vin: str,
+    ) -> None:
+        """Initialize the Polestar Car."""
+        self.config_entry = config_entry
+        self.vin = vin.upper()
+        super().__init__(
+            hass,
+            logger=_LOGGER,
+            name=f"Polestar {self.get_short_id()}",
+            update_interval=DEFAULT_SCAN_INTERVAL,
+        )
+        self.unique_id = f"{self.config_entry.entry_id}_{self.vin.lower()}"
+        self.polestar_api = api
+        self.car_information = self.get_car_information()
+        self.model = (
+            self.car_information["model_name"] if self.car_information else "Unknown"
+        )
+
+    def get_short_id(self) -> str:
+        """Last 4 characters of the VIN"""
+        return self.vin[-4:]
+
+    def get_car_information(self) -> dict[str, Any]:
+        """Get current car information"""
+
+        if data := self.polestar_api.get_car_information(self.vin):
+            if data.battery and (match := re.search(r"(\d+) kWh", data.battery)):
+                battery_capacity = match.group(1)
+            else:
+                battery_capacity = None
+
+            if data.torque and (match := re.search(r"(\d+) Nm", data.torque)):
+                torque = match.group(1)
+            else:
+                torque = None
+
+            return {
+                "vin": self.vin,
+                "internal_vehicle_id": data.internal_vehicle_identifier,
+                "car_image": data.image_url,
+                "registration_number": data.registration_no,
+                "registration_date": data.registration_date,
+                "factory_complete_date": data.factory_complete_date,
+                "model_name": data.model_name,
+                "software_version": data.software_version,
+                "software_version_release": data.software_version_timestamp,
+                "battery_capacity": battery_capacity,
+                "torque": torque,
+            }
+        else:
+            _LOGGER.warning("No car information for VIN %s", self.vin)
+            return {}
+
+    def get_car_battery(self) -> dict[str, Any]:
+        """Get current car battery readings"""
+
+        if data := self.polestar_api.get_car_battery(self.vin):
+            if (
+                data.battery_charge_level_percentage is not None
+                and data.battery_charge_level_percentage != 0
+                and data.estimated_distance_to_empty_km is not None
+            ):
+                estimate_full_charge_range = round(
+                    data.estimated_distance_to_empty_km
+                    / data.battery_charge_level_percentage
+                    * 100,
+                    2,
+                )
+            else:
+                estimate_full_charge_range = None
+
+            if data.estimated_charging_time_to_full_minutes:
+                timestamp = datetime.now().replace(second=0, microsecond=0) + timedelta(
+                    minutes=data.estimated_charging_time_to_full_minutes
+                )
+                estimated_fully_charged_time = dt_util.as_local(timestamp).strftime(
+                    "%Y-%m-%d %H:%M:%S"
+                )
+            else:
+                estimated_fully_charged_time = None
+
+            return {
+                "battery_charge_level": data.battery_charge_level_percentage,
+                "charging_status": data.charging_status,
+                "charger_connection_status": data.charger_connection_status,
+                "charging_power": data.charging_power_watts,
+                "charging_current": data.charging_current_amps,
+                "average_energy_consumption_kwh_per_100": data.average_energy_consumption_kwh_per_100km,
+                "estimate_range": data.estimated_distance_to_empty_km,
+                "estimate_full_charge_range": estimate_full_charge_range,
+                "estimated_charging_time_minutes_to_target_distance": data.estimated_charging_time_minutes_to_target_distance,
+                "estimated_charging_time_to_full": data.estimated_charging_time_to_full_minutes,
+                "estimated_fully_charged_time": estimated_fully_charged_time,
+                "last_updated_battery_data": data.event_updated_timestamp,
+            }
+        else:
+            _LOGGER.warning("No battery information for VIN %s", self.vin)
+            return {}
+
+    def get_car_odometer(self) -> dict[str, Any]:
+        """Get current car odometer readings"""
+
+        if data := self.polestar_api.get_car_odometer(self.vin):
+            return {
+                "current_odometer": data.odometer_meters,
+                "average_speed": data.average_speed_km_per_hour,
+                "current_trip_meter_automatic": data.trip_meter_automatic_km,
+                "current_trip_meter_manual": data.trip_meter_manual_km,
+                "last_updated_odometer_data": data.event_updated_timestamp,
+            }
+        else:
+            _LOGGER.warning("No odometer information for VIN %s", self.vin)
+            return {}
+
+    async def _async_update_data(self) -> Any:
+        """Update data via library."""
+
+        res = self.car_information.copy()
+        try:
+            await self.polestar_api.update_latest_data(self.vin)
+            res.update(self.get_car_odometer())
+            res.update(self.get_car_battery())
+        except PolestarAuthException as exc:
+            _LOGGER.error("Authentication failed for VIN %s: %s", self.vin, str(exc))
+            res["api_connected"] = False
+            raise ConfigEntryAuthFailed(exc) from exc
+        except PolestarApiException as exc:
+            _LOGGER.error("Update failed for VIN %s: %s", self.vin, str(exc))
+            res["api_connected"] = False
+            raise UpdateFailed(exc) from exc
+        except Exception as exc:
+            _LOGGER.error(
+                "Unexpected error updating data for VIN %s: %s", self.vin, str(exc)
+            )
+            res["api_connected"] = False
+            raise UpdateFailed(exc) from exc
+        else:
+            res["api_connected"] = (
+                self.get_latest_call_code_data() == 200
+                and self.get_latest_call_code_auth() == 200
+                and self.polestar_api.auth.is_token_valid()
+            )
+        finally:
+            if token_expire := self.get_token_expiry():
+                res["api_token_expires_at"] = dt_util.as_local(token_expire).strftime(
+                    "%Y-%m-%d %H:%M:%S"
+                )
+            else:
+                res["api_token_expires_at"] = None
+            res["api_status_code_data"] = self.get_latest_call_code_data() or "Error"
+            res["api_status_code_auth"] = self.get_latest_call_code_auth() or "Error"
+        return res
+
+    def get_token_expiry(self) -> datetime | None:
+        """Get the token expiry time."""
+        return self.polestar_api.auth.token_expiry
+
+    def get_latest_call_code_data(self) -> int | None:
+        """Get the latest call code data API."""
+        return self.polestar_api.get_status_code()
+
+    def get_latest_call_code_auth(self) -> int | None:
+        """Get the latest call code auth API."""
+        return self.polestar_api.auth.get_status_code()