Merge pull request #98 from wbyoung/add-last-webhook-received-sensor

Add a sensor for the when webhooks are received

Author: wbyoung

README.md

@@ -222,6 +222,10 @@ All entities have the following attributes:
 
 Links to relevant API documentation are provided for each entity described below as well as the [permissions each entity requires](https://smartcar.com/docs/api-reference/permissions). When the required permissions are [not requested during setup](#authorization-data-entry), those entities will not be created.
 
+In addition to the above, there is a sensor to aid in setting up the integration:
+
+- [`sensor.<make_model>_last_webhook_received`](#sensormake_model_last_webhook_received)
+
 ### `device_tracker.<make_model>_location`
 
 The GPS [location](https://smartcar.com/docs/api-reference/get-location) of the vehicle.
@@ -624,6 +628,17 @@ Requires permissions: `read_security`, `control_security`
 
 _Note: some models, e.g., VW ID.4 2023+ do not have this functionality._
 
+### `sensor.<make_model>_last_webhook_received`
+
+The time the last webhook was received. This is set even when a received webhook is not valid (i.e. an unauthenticated request or for an unknown vehicle).
+
+Enabled by default: :white_check_mark:
+
+#### Attributes
+
+- `response_status`: The status code used to respond to the webhook.
+- `response_data`: The data sent in the response (when available).
+
 ## Actions
 
 Smartcar provides the following actions:

custom_components/smartcar/__init__.py

@@ -17,6 +17,7 @@
     async_get_config_entry_implementation,
 )
 from homeassistant.helpers.typing import ConfigType
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
 from . import util
 from .auth import AbstractAuth
@@ -59,7 +60,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     oauth_session = OAuth2Session(hass, entry, implementation)
     auth = AsyncConfigEntryAuth(websession, oauth_session, API_HOST)
     coordinators: dict[str, SmartcarVehicleCoordinator] = {}
-    entry.runtime_data = SmartcarData(auth=auth, coordinators=coordinators)
+    meta_coordinator = DataUpdateCoordinator(
+        hass, _LOGGER, name=f"{DOMAIN}_meta", config_entry=entry
+    )
+    meta_coordinator.async_set_updated_data({})
+    entry.runtime_data = SmartcarData(
+        auth=auth,
+        coordinators=coordinators,
+        meta_coordinator=meta_coordinator,
+    )
     device_registry = dr.async_get(hass)
     other_vins = vehicle_vins_in_use(hass, entry)
 

custom_components/smartcar/const.py

@@ -113,6 +113,7 @@ class EntityDescriptionKey(StrEnum):
     CHARGE_AMPERAGE_MAX = auto()
     CHARGE_FAST_CHARGER_PRESENT = auto()
     FIRMWARE_VERSION = auto()
+    LAST_WEBHOOK_RECEIVED = auto()
 
 
 DEFAULT_ENABLED_ENTITY_DESCRIPTION_KEYS = {

custom_components/smartcar/coordinator.py

@@ -583,6 +583,8 @@ def _batch_add_defaults(self) -> None:
 
         for entity in entities:
             _, key = entity.unique_id.split("_", 1)
+            if key not in DATAPOINT_ENTITY_KEY_MAP:
+                continue
             config = DATAPOINT_ENTITY_KEY_MAP[key]
 
             # currently polling is only supported via the v2 api

custom_components/smartcar/entity.py

@@ -226,6 +226,13 @@ class SmartcarEntityDescription(EntityDescription):
     )
 
 
+class SmartcarMetaEntityDescription(EntityDescription):
+    """Class describing Smartcar meta sensor entities."""
+
+    value_fn: Callable[[dict[str, Any]], str | int | float | dt.datetime | None]
+    attr_fn: Callable[[dict[str, Any]], dict[str, Any]] = lambda _: {}
+
+
 def inject_raw_value[RawValueT](
     coordinator: SmartcarVehicleCoordinator,
     description: EntityDescription,

custom_components/smartcar/sensor.py

@@ -1,7 +1,9 @@
 from dataclasses import dataclass
+import datetime as dt
 from datetime import date, datetime
 from decimal import Decimal
 import logging
+from typing import Any
 
 from homeassistant.components.sensor import (
     SensorDeviceClass,
@@ -12,6 +14,7 @@
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
     PERCENTAGE,
+    EntityCategory,
     UnitOfElectricCurrent,
     UnitOfElectricPotential,
     UnitOfEnergy,
@@ -23,19 +26,28 @@
     UnitOfVolume,
 )
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import StateType
+from homeassistant.helpers.update_coordinator import (
+    CoordinatorEntity,
+    DataUpdateCoordinator,
+)
 from homeassistant.util.unit_conversion import DistanceConverter, PressureConverter
 
-from .const import EntityDescriptionKey
+from .const import DOMAIN, EntityDescriptionKey
 from .coordinator import (
     VEHICLE_BACK_ROW,
     VEHICLE_FRONT_ROW,
     VEHICLE_LEFT_COLUMN,
     VEHICLE_RIGHT_COLUMN,
     SmartcarVehicleCoordinator,
 )
-from .entity import SmartcarEntity, SmartcarEntityDescription
+from .entity import (
+    SmartcarEntity,
+    SmartcarEntityDescription,
+    SmartcarMetaEntityDescription,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -45,6 +57,13 @@ class SmartcarSensorDescription(SensorEntityDescription, SmartcarEntityDescripti
     """Class describing Smartcar sensor entities."""
 
 
+@dataclass(frozen=True, kw_only=True)
+class SmartcarMetaSensorDescription(
+    SensorEntityDescription, SmartcarMetaEntityDescription
+):
+    """Class describing Smartcar meta sensor entities."""
+
+
 SENSOR_TYPES: tuple[SmartcarSensorDescription, ...] = (
     SmartcarSensorDescription(
         key=EntityDescriptionKey.BATTERY_CAPACITY,
@@ -310,6 +329,23 @@ class SmartcarSensorDescription(SensorEntityDescription, SmartcarEntityDescripti
     ),
 )
 
+META_SENSOR_TYPES: tuple[SmartcarMetaSensorDescription, ...] = (
+    SmartcarMetaSensorDescription(
+        key=EntityDescriptionKey.LAST_WEBHOOK_RECEIVED,
+        name="Last Webhook Received",
+        device_class=SensorDeviceClass.TIMESTAMP,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        value_fn=lambda data: data.get("last_webhook_received_at"),
+        attr_fn=lambda data: (
+            {
+                f"response_{key}": value
+                for key, value in data.get("last_webhook_response", {}).items()
+            }
+        ),
+        icon="mdi:clock",
+    ),
+)
+
 
 async def async_setup_entry(  # noqa: RUF029
     hass: HomeAssistant,  # noqa: ARG001
@@ -320,12 +356,21 @@ async def async_setup_entry(  # noqa: RUF029
     coordinators: dict[str, SmartcarVehicleCoordinator] = (
         entry.runtime_data.coordinators
     )
+    meta_coordinator = entry.runtime_data.meta_coordinator
     _LOGGER.debug("Setting up sensors for VINs: %s", list(coordinators.keys()))
     entities = [
         SmartcarSensor(coordinator, description)
         for coordinator in coordinators.values()
         for description in SENSOR_TYPES
         if coordinator.is_scope_enabled(description.key, verbose=True)
+    ] + [
+        SmartcarMetaSensor(
+            meta_coordinator,
+            description,
+            {"identifiers": {(DOMAIN, vehicle_coordinator.vin)}},
+        )
+        for vehicle_coordinator in coordinators.values()
+        for description in META_SENSOR_TYPES
     ]
     _LOGGER.info("Adding %s Smartcar sensor entities", len(entities))
     async_add_entities(entities)
@@ -341,3 +386,35 @@ class SmartcarSensor[ValueT, RawValueT](
     @property
     def native_value(self) -> StateType | date | datetime | Decimal:
         return self._extract_value()
+
+
+class SmartcarMetaSensor(CoordinatorEntity[DataUpdateCoordinator], SensorEntity):
+    """Meta sensor entity."""
+
+    _attr_has_entity_name = True
+    entity_description: SmartcarMetaEntityDescription
+
+    def __init__(
+        self,
+        coordinator: DataUpdateCoordinator,
+        description: SmartcarMetaEntityDescription,
+        device_info: DeviceInfo,
+    ) -> None:
+        """Initialize."""
+        super().__init__(coordinator)
+
+        (_, vin) = next(iter(device_info["identifiers"]))
+
+        self.entity_description = description
+        self._attr_unique_id = f"{vin}_{description.key}"
+        self._attr_device_info = device_info
+
+    @property
+    def native_value(self) -> str | int | float | dt.datetime | None:
+        """Return the state."""
+        return self.entity_description.value_fn(self.coordinator.data)
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Return the state attributes."""
+        return self.entity_description.attr_fn(self.coordinator.data)

custom_components/smartcar/types.py

@@ -5,6 +5,8 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
 from .auth import AbstractAuth
 
 if TYPE_CHECKING:
@@ -17,6 +19,7 @@ class SmartcarData:
 
     auth: AbstractAuth
     coordinators: dict[str, SmartcarVehicleCoordinator]
+    meta_coordinator: DataUpdateCoordinator
 
 
 @dataclass

custom_components/smartcar/webhooks.py

@@ -1,9 +1,11 @@
+from collections.abc import Callable
 import copy
+from functools import wraps
 import hmac
 from http import HTTPStatus
 import json
 import logging
-from typing import Any, Literal
+from typing import Any, Literal, cast
 
 from aiohttp import web
 from homeassistant.components import cloud, webhook
@@ -39,6 +41,30 @@ async def webhook_url_from_id(hass: HomeAssistant, webhook_id: str) -> tuple[str
     return webhook_url, cloudhook
 
 
+def update_meta_coordinator_data[F: Callable[..., Any], ReturnT](fn: F) -> F:
+    @wraps(fn)
+    async def wrapper(*args, **kwargs) -> ReturnT:  # noqa: ANN002, ANN003
+        response = await fn(*args, **kwargs)
+        config_entry = kwargs["config_entry"]
+        status = response.status
+        data = response.text or (response.body and response.body.decode("utf-8"))
+        meta_coordinator = config_entry.runtime_data.meta_coordinator
+        meta_coordinator.async_set_updated_data(
+            {
+                **meta_coordinator.data,
+                "last_webhook_received_at": dt_util.utcnow(),
+                "last_webhook_response": {
+                    "status": status,
+                    **({"data": data} if data else {}),
+                },
+            }
+        )
+        return cast("ReturnT", response)
+
+    return cast("F", wrapper)
+
+
+@update_meta_coordinator_data
 async def handle_webhook(
     hass: HomeAssistant,  # noqa: ARG001
     webhook_id: str,  # noqa: ARG001

tests/__init__.py

@@ -2,6 +2,7 @@
 
 import importlib
 
+import datetime as dt
 from homeassistant.core import HomeAssistant
 from pytest_homeassistant_custom_component.common import (
     MockConfigEntry,
@@ -12,6 +13,7 @@
 from custom_components.smartcar.const import DOMAIN
 
 MOCK_API_ENDPOINT = "http://test.local"
+MOCK_UTC_NOW = dt.datetime(2026, 2, 17, 16, 21, 32, 3842, tzinfo=dt.UTC)
 
 
 async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:

tests/conftest.py

@@ -2,6 +2,8 @@
 
 from . import bootstrap as bootstrap  # noqa: I001, PLC0414
 
+from freezegun.api import FrozenDateTimeFactory
+
 from collections.abc import Awaitable, Callable, Generator
 import json
 import logging
@@ -31,6 +33,7 @@
 from pytest_homeassistant_custom_component.typing import ClientSessionGenerator
 from syrupy.assertion import SnapshotAssertion
 from .syrupy import SmartcarSnapshotExtension
+from . import MOCK_UTC_NOW
 
 from custom_components.smartcar.auth import AbstractAuth
 from custom_components.smartcar.const import (
@@ -94,6 +97,15 @@ def snapshot(snapshot: SnapshotAssertion) -> SnapshotAssertion:
     return snapshot.use_extension(SmartcarSnapshotExtension)
 
 
+@pytest.fixture
+def mock_now(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+):
+    """Return a mock now & utcnow datetime."""
+    freezer.move_to(MOCK_UTC_NOW)
+
+
 @pytest.fixture
 def mock_hmac_sha256_hexdigest_value() -> str:
     return "1234"
@@ -346,6 +358,7 @@ def webhook_scenario(
     webhook_body: str,
     webhook_headers: dict[str, Any],
     expected: dict[str, Any],
+    mock_now: Any,
     device_registry: dr.DeviceRegistry,
     entity_registry: er.EntityRegistry,
     caplog: pytest.LogCaptureFixture,