Merge pull request #249 from jschlyter/use_models

Use new models for all sensors

custom_components/polestar_api/binary_sensor.py

@@ -1,6 +1,7 @@
 """Support for Polestar binary sensors."""
 
 import logging
+from typing import Final
 
 from homeassistant.components.binary_sensor import (
     BinarySensorDeviceClass,
@@ -19,7 +20,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-ENTITY_DESCRIPTIONS = (
+ENTITY_DESCRIPTIONS: Final[tuple[BinarySensorEntityDescription, ...]] = (
     BinarySensorEntityDescription(
         key="api_connected",
         name="API Connected",

custom_components/polestar_api/image.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass
 from typing import Final
 
 import homeassistant.util.dt as dt_util
@@ -19,25 +18,10 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-@dataclass
-class PolestarImageDescriptionMixin:
-    """A mixin class for image entities."""
-
-    query: str
-    field_name: str
-
-
-@dataclass
-class PolestarImageDescription(ImageEntityDescription, PolestarImageDescriptionMixin):
-    """A class that describes image entities."""
-
-
-POLESTAR_IMAGE_TYPES: Final[tuple[PolestarImageDescription, ...]] = (
-    PolestarImageDescription(
+ENTITY_DESCRIPTIONS: Final[tuple[ImageEntityDescription, ...]] = (
+    ImageEntityDescription(
         key="car_image",
         name="Car Image",
-        query="getConsumerCarsV2",
-        field_name="content/images/studio/url",
         entity_registry_enabled_default=False,
     ),
 )
@@ -53,7 +37,7 @@ async def async_setup_entry(
     async_add_entities(
         [
             PolestarImage(car, entity_description, hass)
-            for entity_description in POLESTAR_IMAGE_TYPES
+            for entity_description in ENTITY_DESCRIPTIONS
             for car in entry.runtime_data.cars
         ]
     )
@@ -62,13 +46,13 @@ async def async_setup_entry(
 class PolestarImage(PolestarEntity, ImageEntity):
     """Representation of a Polestar image."""
 
-    entity_description: PolestarImageDescription
+    entity_description: ImageEntityDescription
     _attr_has_entity_name = True
 
     def __init__(
         self,
         car: PolestarCar,
-        entity_description: PolestarImageDescription,
+        entity_description: ImageEntityDescription,
         hass: HomeAssistant,
     ) -> None:
         """Initialize the Polestar image."""
@@ -83,10 +67,7 @@ def __init__(
         self._attr_translation_key = f"polestar_{entity_description.key}"
 
     async def async_update_image_url(self) -> None:
-        value = self.car.get_value(
-            query=self.entity_description.query,
-            field_name=self.entity_description.field_name,
-        )
+        value = self.car.data.get(self.entity_description.key)
         if value is None:
             _LOGGER.debug("No image URL found")
         elif isinstance(value, str) and value != self._attr_image_url:

custom_components/polestar_api/polestar.py

@@ -1,8 +1,10 @@
 """Polestar API for Polestar integration."""
 
 import logging
+import re
 from datetime import datetime, timedelta
 
+import homeassistant.util.dt as dt_util
 import httpx
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity import DeviceInfo
@@ -29,12 +31,10 @@ def __init__(
             f"{unique_id}_{self.vin.lower()}" if unique_id else self.vin.lower()
         )
         self.name = "Polestar " + self.get_short_id()
-        self.model = str(
-            self.get_value("getConsumerCarsV2", "content/model/name") or "Unknown model"
-        )
         self.scan_interval = DEFAULT_SCAN_INTERVAL
         self.async_update = Throttle(min_time=self.scan_interval)(self.async_update)
         self.data = {}
+        self.update_car_information()
 
     def get_unique_id(self) -> str:
         """Return unique identifier"""
@@ -49,20 +49,134 @@ def get_device_info(self) -> DeviceInfo:
         return DeviceInfo(
             identifiers={(POLESTAR_API_DOMAIN, self.get_unique_id())},
             manufacturer="Polestar",
-            model=self.model,
+            model=self.data.get("model_name", "Unknown model"),
             name=self.name,
             serial_number=self.vin,
         )
 
+    def update_car_information(self) -> None:
+        """Update data with 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
+
+            self.data.update(
+                {
+                    "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,
+                }
+            )
+
+    def update_battery(self) -> None:
+        """Update data with 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 = "Not charging"
+
+            self.data.update(
+                {
+                    "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,
+                }
+            )
+
+    def update_odometer(self) -> None:
+        """Update data with current car odometer readings"""
+
+        if data := self.polestar_api.get_car_odometer(self.vin):
+            average_speed_km_per_hour = (
+                round(data.average_speed_km_per_hour)
+                if data.average_speed_km_per_hour
+                else None
+            )
+            self.data.update(
+                {
+                    "current_odometer": data.odometer_meters,
+                    "average_speed": 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,
+                }
+            )
+
     async def async_update(self) -> None:
         """Update data from Polestar."""
+
         try:
             await self.polestar_api.get_ev_data(self.vin)
+
+            self.update_odometer()
+            self.update_battery()
+
+            if token_expire := self.get_token_expiry():
+                self.data["api_token_expires_at"] = dt_util.as_local(
+                    token_expire
+                ).strftime("%Y-%m-%d %H:%M:%S")
+            else:
+                self.data["api_token_expires_at"] = None
+
             self.data["api_connected"] = (
-                self.polestar_api.latest_call_code == 200
-                and self.polestar_api.auth.latest_call_code == 200
+                self.get_latest_call_code_data() == 200
+                and self.get_latest_call_code_auth() == 200
                 and self.polestar_api.auth.is_token_valid()
             )
+
+            self.data["api_status_code_data"] = (
+                self.get_latest_call_code_data() or "Error"
+            )
+            self.data["api_status_code_auth"] = (
+                self.get_latest_call_code_auth() or "Error"
+            )
+
             return
         except PolestarApiException as e:
             _LOGGER.warning("API Exception on update data %s", str(e))
@@ -85,20 +199,6 @@ async def async_update(self) -> None:
             self.polestar_api.next_update = datetime.now() + timedelta(seconds=60)
         self.polestar_api.latest_call_code = 500
 
-    def get_value(self, query: str, field_name: str):
-        """Get the latest value from the Polestar API."""
-        if query is None or field_name is None:
-            return None
-        data = self.polestar_api.get_latest_data(
-            vin=self.vin, query=query, field_name=field_name
-        )
-        if data is None:
-            # if amp and voltage can be null, so we will return 0
-            if field_name in ("chargingCurrentAmps", "chargingPowerWatts"):
-                return 0
-            return
-        return data
-
     def get_token_expiry(self) -> datetime | None:
         """Get the token expiry time."""
         return self.polestar_api.auth.token_expiry

custom_components/polestar_api/pypolestar/models.py

@@ -49,6 +49,7 @@ class CarInformationData(CarBaseInformation):
     battery: str | None
     torque: str | None
     software_version: str | None
+    software_version_timestamp: datetime | None
 
     @classmethod
     def from_dict(cls, data: GqlDict) -> Self:
@@ -68,13 +69,16 @@ def from_dict(cls, data: GqlDict) -> Self:
             battery=get_field_name_str("content/specification/battery", data),
             torque=get_field_name_str("content/specification/torque", data),
             software_version=get_field_name_str("software/version", data),
+            software_version_timestamp=get_field_name_datetime(
+                "software/versionTimestamp", data
+            ),
             _received_timestamp=datetime.now(tz=timezone.utc),
         )
 
 
 @dataclass(frozen=True)
 class CarOdometerData(CarBaseInformation):
-    average_speed_km_per_hour: float | None
+    average_speed_km_per_hour: int | None
     odometer_meters: int | None
     trip_meter_automatic_km: float | None
     trip_meter_manual_km: float | None
@@ -86,9 +90,7 @@ def from_dict(cls, data: GqlDict) -> Self:
             raise TypeError
 
         return cls(
-            average_speed_km_per_hour=get_field_name_float(
-                "averageSpeedKmPerHour", data
-            ),
+            average_speed_km_per_hour=get_field_name_int("averageSpeedKmPerHour", data),
             odometer_meters=get_field_name_int("odometerMeters", data),
             trip_meter_automatic_km=get_field_name_float("tripMeterAutomaticKm", data),
             trip_meter_manual_km=get_field_name_float("tripMeterManualKm", data),