Add character usage sensors to ElevenLabs

homeassistant/components/elevenlabs/__init__.py

@@ -15,8 +15,9 @@
 from homeassistant.helpers.httpx_client import get_async_client
 
 from .const import CONF_MODEL
+from .coordinator import ElevenLabsDataUpdateCoordinator
 
-PLATFORMS: list[Platform] = [Platform.TTS]
+PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.TTS]
 
 
 async def get_model_by_id(client: AsyncElevenLabs, model_id: str) -> Model | None:
@@ -33,13 +34,14 @@ class ElevenLabsData:
     """ElevenLabs data type."""
 
     client: AsyncElevenLabs
+    coordinator: ElevenLabsDataUpdateCoordinator
     model: Model
 
 
-type EleventLabsConfigEntry = ConfigEntry[ElevenLabsData]
+type ElevenLabsConfigEntry = ConfigEntry[ElevenLabsData]
 
 
-async def async_setup_entry(hass: HomeAssistant, entry: EleventLabsConfigEntry) -> bool:
+async def async_setup_entry(hass: HomeAssistant, entry: ElevenLabsConfigEntry) -> bool:
     """Set up ElevenLabs text-to-speech from a config entry."""
     entry.add_update_listener(update_listener)
     httpx_client = get_async_client(hass)
@@ -55,21 +57,24 @@ async def async_setup_entry(hass: HomeAssistant, entry: EleventLabsConfigEntry)
     if model is None or (not model.languages):
         raise ConfigEntryError("Model could not be resolved")
 
-    entry.runtime_data = ElevenLabsData(client=client, model=model)
+    coordinator = ElevenLabsDataUpdateCoordinator(hass, client)
+    await coordinator.async_config_entry_first_refresh()
+
+    entry.runtime_data = ElevenLabsData(
+        client=client, model=model, coordinator=coordinator
+    )
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
     return True
 
 
-async def async_unload_entry(
-    hass: HomeAssistant, entry: EleventLabsConfigEntry
-) -> bool:
+async def async_unload_entry(hass: HomeAssistant, entry: ElevenLabsConfigEntry) -> bool:
     """Unload a config entry."""
     return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
 
 
 async def update_listener(
-    hass: HomeAssistant, config_entry: EleventLabsConfigEntry
+    hass: HomeAssistant, config_entry: ElevenLabsConfigEntry
 ) -> None:
     """Handle options update."""
     await hass.config_entries.async_reload(config_entry.entry_id)

homeassistant/components/elevenlabs/coordinator.py

@@ -0,0 +1,96 @@
+"""The coordinator for the ElevenLabs integration."""
+
+from datetime import UTC, datetime, timedelta
+from logging import getLogger
+
+from elevenlabs.client import AsyncElevenLabs
+from elevenlabs.core import ApiError
+from elevenlabs.types import UsageCharactersResponseModel
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN
+
+_LOGGER = getLogger(__name__)
+SCAN_INTERVAL = timedelta(seconds=120)
+BREAKDOWN_KEY = "All"
+
+
+class ElevenLabsDataUpdateCoordinator(
+    DataUpdateCoordinator[UsageCharactersResponseModel]
+):
+    """Class to manage fetching ElevenLabs data."""
+
+    def __init__(self, hass: HomeAssistant, client: AsyncElevenLabs) -> None:
+        """Initialize the ElevenLabs data updater."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=SCAN_INTERVAL,
+            always_update=False,
+        )
+        self.client = client
+        self.usage_data: list[int] = []
+        self.times: list[int] = []
+        self.last_unix_update = 1
+
+    def merge_usage_data(
+        self, usage_data: UsageCharactersResponseModel
+    ) -> UsageCharactersResponseModel:
+        """Merge new usage data to the coordinator. Returns the merged data."""
+        if BREAKDOWN_KEY not in usage_data.usage:
+            _LOGGER.warning(
+                "No data found for breakdown key %s in ElevenLabs usage response!",
+                BREAKDOWN_KEY,
+            )
+            return UsageCharactersResponseModel(
+                time=self.times, usage={BREAKDOWN_KEY: self.usage_data}
+            )
+        for idx, time in enumerate(usage_data.time):
+            if time not in self.times:
+                self.times.append(time)
+                self.usage_data.append(usage_data.usage[BREAKDOWN_KEY][idx])
+            else:
+                self.usage_data[self.times.index(time)] = usage_data.usage[
+                    BREAKDOWN_KEY
+                ][idx]
+        return UsageCharactersResponseModel(
+            time=self.times, usage={BREAKDOWN_KEY: self.usage_data}
+        )
+
+    async def _async_update_data(self) -> UsageCharactersResponseModel:
+        """Fetch data from ElevenLabs."""
+        now = datetime.now(UTC)
+        # Convert to milliseconds
+        current_time_unix = int(now.timestamp() * 1_000)
+        _LOGGER.debug(
+            "Updating ElevenLabs usage data, start: %s, end: %s",
+            self.last_unix_update,
+            current_time_unix,
+        )
+        try:
+            response = await self.client.usage.get_characters_usage_metrics(
+                start_unix=self.last_unix_update, end_unix=current_time_unix
+            )
+            merged_response = self.merge_usage_data(response)
+            current_date_unix = int(
+                datetime(now.year, now.month, now.day, tzinfo=UTC).timestamp() * 1_000
+            )
+            self.last_unix_update = current_date_unix
+        except ApiError as e:
+            _LOGGER.exception(
+                "API Error when fetching usage data in ElevenLabs integration!"
+            )
+            raise UpdateFailed(
+                "Error fetching usage data for ElevenLabs integration"
+            ) from e
+        except Exception as e:
+            _LOGGER.exception(
+                "Unknown error when fetching usage data in ElevenLabs integration!"
+            )
+            raise UpdateFailed(
+                "Unknown error fetching usage data for ElevenLabs integration"
+            ) from e
+        return merged_response

homeassistant/components/elevenlabs/sensor.py

@@ -0,0 +1,143 @@
+"""The sensor entities for the ElevenLabs integration."""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from logging import getLogger
+
+from elevenlabs.types import UsageCharactersResponseModel
+from propcache import cached_property
+
+from homeassistant.components.sensor import (
+    SensorEntity,
+    SensorEntityDescription,
+    SensorStateClass,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import ElevenLabsConfigEntry
+from .coordinator import BREAKDOWN_KEY, ElevenLabsDataUpdateCoordinator
+
+_LOGGER = getLogger(__name__)
+
+
+@dataclass(frozen=True, kw_only=True)
+class ElevenLabsSensorEntityDescription(SensorEntityDescription):
+    """Describes ElevenLabs sensor entity."""
+
+    value_fn: Callable[[UsageCharactersResponseModel], int]
+
+
+def characters_sum_from_timestamp(
+    data: UsageCharactersResponseModel, timestamp: datetime
+) -> int:
+    """Return the sum of characters used since the given timestamp."""
+    # ElevenLabs timestamps are 0:00 UTC, so we need to adjust the time to match
+    timestamp = datetime(timestamp.year, timestamp.month, timestamp.day, tzinfo=UTC)
+    now = datetime.now(UTC)
+    curr = timestamp
+    characters = 0
+    while curr < now:
+        curr_unix = int(curr.timestamp() * 1_000)
+        curr_idx = data.time.index(curr_unix)
+        _LOGGER.debug(
+            "Day %s: %s characters",
+            curr.strftime("%Y-%m-%d"),
+            data.usage[BREAKDOWN_KEY][curr_idx],
+        )
+        characters += data.usage[BREAKDOWN_KEY][curr_idx]
+        curr += timedelta(days=1)
+    return characters
+
+
+def characters_this_month(data: UsageCharactersResponseModel) -> int:
+    """Return the number of characters used this month."""
+    now = datetime.now(UTC)
+    first_day = datetime(now.year, now.month, 1, tzinfo=UTC)
+    return characters_sum_from_timestamp(data, first_day)
+
+
+def characters_7_days(data: UsageCharactersResponseModel) -> int:
+    """Return the number of characters used in the last 7 days."""
+    now = datetime.now(UTC)
+    last_seven = now - timedelta(days=6)  # Last 7 days, including today, so 6 days ago
+    return characters_sum_from_timestamp(data, last_seven)
+
+
+def characters_today(data: UsageCharactersResponseModel) -> int:
+    """Return the number of characters used today."""
+    now = datetime.now(UTC)
+    # ElevenLabs timestamps are 0:00 UTC, so we need to adjust the current time to match
+    today = datetime(now.year, now.month, now.day, tzinfo=UTC)
+    today_unix = int(today.timestamp() * 1_000)
+    today_idx = data.time.index(today_unix)
+    return data.usage[BREAKDOWN_KEY][today_idx]
+
+
+SENSORS: tuple[ElevenLabsSensorEntityDescription, ...] = (
+    ElevenLabsSensorEntityDescription(
+        key="total_characters",
+        translation_key="total_characters",
+        value_fn=lambda data: sum(data.usage[BREAKDOWN_KEY]),
+    ),
+    ElevenLabsSensorEntityDescription(
+        key="characters_today",
+        translation_key="characters_today",
+        value_fn=characters_today,
+    ),
+    ElevenLabsSensorEntityDescription(
+        key="characters_7_days",
+        translation_key="characters_7_days",
+        value_fn=characters_7_days,
+    ),
+    ElevenLabsSensorEntityDescription(
+        key="characters_this_month",
+        translation_key="characters_this_month",
+        value_fn=characters_this_month,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ElevenLabsConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up ElevenLabs sensors based on a config entry."""
+    coordinator = entry.runtime_data.coordinator
+    async_add_entities(
+        ElevenLabsUsageEntity(
+            coordinator=coordinator,
+            description=description,
+            entry_id=f"{entry.entry_id}",
+        )
+        for description in SENSORS
+    )
+
+
+class ElevenLabsUsageEntity(
+    CoordinatorEntity[ElevenLabsDataUpdateCoordinator], SensorEntity
+):
+    """ElevenLabs usage entity for getting usage metrics."""
+
+    _attr_state_class = SensorStateClass.TOTAL_INCREASING
+
+    def __init__(
+        self,
+        *,
+        coordinator: ElevenLabsDataUpdateCoordinator,
+        description: ElevenLabsSensorEntityDescription,
+        entry_id: str,
+    ) -> None:
+        """Initialize the ElevenLabs usage entity."""
+        super().__init__(coordinator)
+        self.entity_description: ElevenLabsSensorEntityDescription = description
+        self._attr_unique_id = f"{entry_id}_{description.key}"
+
+    @cached_property
+    def native_value(self) -> float | None:
+        """Return the state of the entity."""
+        # Report the value from the description's value_fn
+        return self.entity_description.value_fn(self.coordinator.data)

homeassistant/components/elevenlabs/strings.json

@@ -45,5 +45,21 @@
         }
       }
     }
+  },
+  "entity": {
+    "sensor": {
+      "total_characters": {
+        "name": "Total characters"
+      },
+      "characters_today": {
+        "name": "Characters today"
+      },
+      "characters_7_days": {
+        "name": "Characters last 7 days"
+      },
+      "characters_this_month": {
+        "name": "Characters this month"
+      }
+    }
   }
 }

homeassistant/components/elevenlabs/tts.py

@@ -21,7 +21,7 @@
 from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import EleventLabsConfigEntry
+from . import ElevenLabsConfigEntry
 from .const import (
     CONF_OPTIMIZE_LATENCY,
     CONF_SIMILARITY,
@@ -54,7 +54,7 @@ def to_voice_settings(options: MappingProxyType[str, Any]) -> VoiceSettings:
 
 async def async_setup_entry(
     hass: HomeAssistant,
-    config_entry: EleventLabsConfigEntry,
+    config_entry: ElevenLabsConfigEntry,
     async_add_entities: AddEntitiesCallback,
 ) -> None:
     """Set up ElevenLabs tts platform via config entry."""