Merge branch 'main' into feature/car-health-data

.github/workflows/test.yml

@@ -10,3 +10,27 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - uses: chartboost/ruff-action@v1
+  pytest:
+    needs: ruff
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Get full python version
+        id: full-python-version
+        run: |
+          echo version=$(python -c "import sys, platform; print('.'.join(str(v) for v in sys.version_info[:3]) + '_' + platform.machine())") >> $GITHUB_OUTPUT
+      - name: Set up cache
+        uses: actions/cache@v4
+        with:
+          path: .venv
+          key: ${{ runner.os }}-venv-${{ steps.full-python-version.outputs.version }}-${{ hashFiles('requirements.txt') }}
+      - name: Create virtual environment
+        run: python3 -m venv .venv
+      - name: Install dependencies
+        run: .venv/bin/pip3 install -r requirements.txt
+      - name: Execute tests
+        run: make PYTEST=.venv/bin/pytest test

Makefile

@@ -1,11 +1,14 @@
 SOURCE=		custom_components
-
+PYTEST=		pytest
 
 all:
 
 lint:
 	ruff check $(SOURCE)
 
 reformat:
-	ruff check --select I --fix $(SOURCE)
-	ruff format $(SOURCE)
+	ruff check --select I --fix $(SOURCE) tests
+	ruff format $(SOURCE) tests
+
+test:
+	PYTHONPATH=. $(PYTEST) -vv tests

config/configuration.yaml

@@ -8,6 +8,8 @@ logbook:
 
 history:
 
+system_health:
+
 # https://www.home-assistant.io/integrations/logger/
 logger:
   default: info

custom_components/polestar_api/__init__.py

@@ -12,7 +12,7 @@
 from .polestar import PolestarCar, PolestarCoordinator
 from .pypolestar.exception import PolestarApiException, PolestarAuthException
 
-PLATFORMS = [Platform.IMAGE, Platform.SENSOR]
+PLATFORMS = [Platform.IMAGE, Platform.SENSOR, Platform.BINARY_SENSOR]
 
 _LOGGER = logging.getLogger(__name__)
 

custom_components/polestar_api/binary_sensor.py

@@ -0,0 +1,69 @@
+"""Support for Polestar binary sensors."""
+
+import logging
+
+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
+
+_LOGGER = logging.getLogger(__name__)
+
+
+ENTITY_DESCRIPTIONS = (
+    BinarySensorEntityDescription(
+        key="api_connected",
+        name="API Connected",
+        device_class=BinarySensorDeviceClass.CONNECTIVITY,
+        entity_category=EntityCategory.DIAGNOSTIC,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,  # noqa: ARG001 Unused function argument: `hass`
+    entry: PolestarConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the binary_sensor platform."""
+    async_add_entities(
+        PolestarBinarySensor(
+            car=car,
+            entity_description=entity_description,
+        )
+        for entity_description in ENTITY_DESCRIPTIONS
+        for car in entry.runtime_data.cars
+    )
+
+
+class PolestarBinarySensor(PolestarEntity, BinarySensorEntity):
+    """integration_blueprint binary_sensor class."""
+
+    def __init__(
+        self,
+        car: PolestarCar,
+        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}"
+
+    @property
+    def is_on(self) -> bool | None:
+        """Return true if the binary_sensor is on."""
+        return self.car.data.get(self.entity_description.key)

custom_components/polestar_api/diagnostics.py

@@ -0,0 +1,35 @@
+"""Provides diagnostics for Polestar API."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.diagnostics import async_redact_data
+from homeassistant.const import CONF_PASSWORD
+from homeassistant.core import HomeAssistant
+
+from .data import PolestarConfigEntry
+
+TO_REDACT = {CONF_PASSWORD}
+
+
+async def async_get_config_entry_diagnostics(
+    hass: HomeAssistant, entry: PolestarConfigEntry
+) -> dict[str, Any]:
+    """Return diagnostics for a config entry."""
+
+    coordinator = entry.runtime_data.coordinator
+    cars = entry.runtime_data.cars
+    api = coordinator.polestar_api
+
+    return {
+        "config_entry_data": async_redact_data(dict(entry.data), TO_REDACT),
+        "cars": [{"vin": car.vin, "model": car.model} for car in cars],
+        "auth_api": {
+            "oidc_provider": api.auth.oidc_provider,
+            "access_token_valid": api.auth.is_token_valid(),
+            "endpoint": api.auth.api_url,
+            "status": api.auth.latest_call_code,
+        },
+        "data_api": {"endpoint": api.api_url, "status": api.latest_call_code},
+    }

custom_components/polestar_api/polestar.py

@@ -34,6 +34,7 @@ def __init__(
         )
         self.scan_interval = DEFAULT_SCAN_INTERVAL
         self.async_update = Throttle(min_time=self.scan_interval)(self.async_update)
+        self.data = {}
 
     def get_unique_id(self) -> str:
         """Return unique identifier"""
@@ -57,6 +58,11 @@ async def async_update(self) -> None:
         """Update data from Polestar."""
         try:
             await self.polestar_api.get_ev_data(self.vin)
+            self.data["api_connected"] = (
+                self.polestar_api.latest_call_code == 200
+                and self.polestar_api.auth.latest_call_code == 200
+                and self.polestar_api.auth.is_token_valid()
+            )
             return
         except PolestarApiException as e:
             _LOGGER.warning("API Exception on update data %s", str(e))
@@ -77,7 +83,7 @@ async def async_update(self) -> None:
         except Exception as e:
             _LOGGER.error("Unexpected Error on update data %s", str(e))
             self.polestar_api.next_update = datetime.now() + timedelta(seconds=60)
-        self.polestar_api.latest_call_code_v2 = 500
+        self.polestar_api.latest_call_code = 500
 
     def get_value(self, query: str, field_name: str):
         """Get the latest value from the Polestar API."""
@@ -97,28 +103,14 @@ def get_token_expiry(self) -> datetime | None:
         """Get the token expiry time."""
         return self.polestar_api.auth.token_expiry
 
-    def get_latest_call_code_v1(self) -> int | None:
-        """Get the latest call code mystar API."""
+    def get_latest_call_code_data(self) -> int | None:
+        """Get the latest call code data API."""
         return self.polestar_api.latest_call_code
 
-    def get_latest_call_code_v2(self) -> int | None:
-        """Get the latest call code mystar-v2 API."""
-        return self.polestar_api.latest_call_code_2
-
     def get_latest_call_code_auth(self) -> int | None:
-        """Get the latest call code mystar API."""
+        """Get the latest call code auth API."""
         return self.polestar_api.auth.latest_call_code
 
-    def get_latest_call_code(self) -> int | None:
-        """Get the latest call code."""
-        # if AUTH code last code is not 200 then we return that error code,
-        # otherwise just give the call_code in API from v1 and then v2
-        if self.polestar_api.auth.latest_call_code != 200:
-            return self.polestar_api.auth.latest_call_code
-        if self.polestar_api.latest_call_code != 200:
-            return self.polestar_api.latest_call_code
-        return self.polestar_api.latest_call_code_2
-
 
 class PolestarCoordinator:
     """Polestar EV integration."""
@@ -137,6 +129,7 @@ def __init__(
         else:
             _LOGGER.debug("Configure Polestar API client for all cars")
         self.unique_id = unique_id
+        self.username = username
         self.polestar_api = PolestarApi(
             username=username,
             password=password,

custom_components/polestar_api/pypolestar/auth.py

@@ -40,6 +40,8 @@ def __init__(
         self.oidc_configuration = {}
         self.latest_call_code = None
         self.logger = _LOGGER.getChild(unique_id) if unique_id else _LOGGER
+        self.oidc_provider = OIDC_PROVIDER_BASE_URL
+        self.api_url = API_AUTH_URL
         self.gql_client = get_gql_client(url=API_AUTH_URL, client=self.client_session)
 
     async def async_init(self) -> None:
@@ -65,6 +67,13 @@ def need_token_refresh(self) -> bool:
             return True
         return False
 
+    def is_token_valid(self) -> bool:
+        return (
+            self.access_token is not None
+            and self.token_expiry is not None
+            and self.token_expiry > datetime.now(tz=timezone.utc)
+        )
+
     async def get_token(self, refresh=False) -> None:
         """Get the token from Polestar."""
         # can't use refresh if the token is expired or not set even if refresh is True

custom_components/polestar_api/pypolestar/const.py

@@ -13,6 +13,4 @@
 OIDC_CLIENT_ID = "l3oopkc_10"
 
 API_AUTH_URL = "https://pc-api.polestar.com/eu-north-1/auth/"
-
-BASE_URL = "https://pc-api.polestar.com/eu-north-1/my-star/"
-BASE_URL_V2 = "https://pc-api.polestar.com/eu-north-1/mystar-v2/"
+API_MYSTAR_V2_URL = "https://pc-api.polestar.com/eu-north-1/mystar-v2/"