Add sensors for device battery voltage and RSSI

custom_components/trmnl/__init__.py

@@ -1 +1,27 @@
 """TRMNL integration for Home Assistant."""
+
+from __future__ import annotations
+
+from homeassistant.const import CONF_API_KEY, CONF_URL
+from homeassistant.core import HomeAssistant
+
+from .trmnl_api import TrmnlApi
+from .trmnl_context import TrmnlConfigEntry, TrmnlContext
+
+_PLATFORMS = ["sensor"]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: TrmnlConfigEntry) -> bool:
+    """Set up the TRMNL integration."""
+    api = TrmnlApi(entry.data[CONF_URL], entry.data[CONF_API_KEY])
+
+    if not await api.validate_api_key():
+        return False
+
+    context = TrmnlContext(api=api)
+    entry.runtime_data = context
+
+    # forward to platform init
+    await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
+
+    return True

custom_components/trmnl/battery_sensor.py

@@ -0,0 +1,37 @@
+"""Represents a battery sensor entity."""
+
+from __future__ import annotations
+
+from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
+from homeassistant.const import UnitOfElectricPotential
+from homeassistant.core import callback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .coordinator import TrmnlCoordinator
+
+
+class BatterySensor(CoordinatorEntity, SensorEntity):
+    """Represents the TRMNL battery voltage."""
+
+    _attr_native_unit_of_measurement = UnitOfElectricPotential.VOLT
+    _attr_device_class = SensorDeviceClass.VOLTAGE
+    _attr_icon = "mdi:battery"
+
+    def __init__(self, coordinator: TrmnlCoordinator, device: dict[str, any]) -> None:
+        """Initialize the entity."""
+        super().__init__(coordinator)
+        self._friendly_id = device["friendly_id"]
+        self._attr_name = f"{device['name']} Battery"
+        self._attr_native_value = device["battery_voltage"]
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        device = next(
+            (d for d in self.coordinator.data if d["friendly_id"] == self._friendly_id),
+            None,
+        )
+        if device is None:
+            return
+        self._attr_native_value = device["battery_voltage"]
+        self.async_write_ha_state()

custom_components/trmnl/config_flow.py

@@ -0,0 +1,53 @@
+"""Config flow for TRMNL integration."""
+
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_API_KEY, CONF_URL
+
+from .const import DOMAIN
+from .trmnl_api import TrmnlApi
+
+CONFIG_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_API_KEY): str,
+        vol.Required(CONF_URL, default="https://trmnl.app/api"): str,
+    },
+)
+
+
+class TrmnlConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """TRMNL config flow."""
+
+    # The schema version of the entries that it creates
+    # Home Assistant will call your migrate method if the version changes
+    VERSION = 1
+    MINOR_VERSION = 1
+
+    async def async_step_user(self, user_input=None):
+        """Handle the initial step."""
+        errors = {}
+        if user_input is not None:
+            # Validate the user input
+            if not await _validate_api_key(
+                user_input[CONF_URL], user_input[CONF_API_KEY]
+            ):
+                errors[CONF_API_KEY] = "Invalid API key"
+
+        if errors or user_input is None:
+            return self.async_show_form(
+                step_id="user",
+                data_schema=CONFIG_SCHEMA,
+                errors=errors,
+            )
+
+        title = user_input[CONF_API_KEY][:11] + "..."
+        return self.async_create_entry(title=title, data=user_input)
+
+
+async def _validate_api_key(url: str, api_key: str) -> bool:
+    """Validate the API key."""
+    if not api_key.startswith("user_"):
+        return False
+
+    return await TrmnlApi(url, api_key).validate_api_key()

custom_components/trmnl/const.py

@@ -0,0 +1,3 @@
+"""Constants for the TRMNL integration."""
+
+DOMAIN = "trmnl"

custom_components/trmnl/coordinator.py

@@ -0,0 +1,67 @@
+"""Coordinator for fetching TRMNL device stats."""
+
+import asyncio
+from datetime import timedelta
+import logging
+
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .trmnl_api import TrmnlApiAuthError, TrmnlApiError
+from .trmnl_context import TrmnlConfigEntry
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class TrmnlCoordinator(DataUpdateCoordinator):
+    """TRMNL API fetch coordinator."""
+
+    def __init__(self, hass: HomeAssistant, config_entry: TrmnlConfigEntry) -> None:
+        """Initialize my coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            # Name of the data. For logging purposes.
+            name="My sensor",
+            config_entry=config_entry,
+            # Polling interval. Will only be polled if there are subscribers.
+            update_interval=timedelta(seconds=10),
+            # Set always_update to `False` if the data returned from the
+            # api can be compared via `__eq__` to avoid duplicate updates
+            # being dispatched to listeners
+            always_update=True,
+        )
+        self._api = config_entry.runtime_data.api
+
+    async def _async_setup(self):
+        """Set up the coordinator.
+
+        This is the place to set up your coordinator,
+        or to load data, that only needs to be loaded once.
+
+        This method will be called automatically during
+        coordinator.async_config_entry_first_refresh.
+        """
+
+    async def _async_update_data(self):
+        """Fetch data from API endpoint.
+
+        This is the place to pre-process the data to lookup tables
+        so entities can quickly look up their data.
+        """
+
+        try:
+            # Note: asyncio.TimeoutError and aiohttp.ClientError are already
+            # handled by the data update coordinator.
+            async with asyncio.timeout(10):
+                # Grab active context variables to limit data required to be fetched from API
+                # Note: using context is not required if there is no need or ability to limit
+                # data retrieved from API.
+                return await self._api.get_devices()
+        except TrmnlApiAuthError as err:
+            # Raising ConfigEntryAuthFailed will cancel future updates
+            # and start a config flow with SOURCE_REAUTH (async_step_reauth)
+            raise ConfigEntryAuthFailed from err
+        except TrmnlApiError as err:
+            raise UpdateFailed(f"Error communicating with API: {err}") from err

custom_components/trmnl/rssi_sensor.py

@@ -0,0 +1,37 @@
+"""Represents a battery sensor entity."""
+
+from __future__ import annotations
+
+from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
+from homeassistant.const import SIGNAL_STRENGTH_DECIBELS_MILLIWATT
+from homeassistant.core import callback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .coordinator import TrmnlCoordinator
+
+
+class RssiSensor(CoordinatorEntity, SensorEntity):
+    """Represents the TRMNL Wi-Fi signal strength."""
+
+    _attr_native_unit_of_measurement = SIGNAL_STRENGTH_DECIBELS_MILLIWATT
+    _attr_device_class = SensorDeviceClass.SIGNAL_STRENGTH
+    _attr_icon = "mdi:wifi"
+
+    def __init__(self, coordinator: TrmnlCoordinator, device: dict[str, any]) -> None:
+        """Initialize the entity."""
+        super().__init__(coordinator)
+        self._friendly_id = device["friendly_id"]
+        self._attr_name = f"{device['name']} RSSI"
+        self._attr_native_value = device["rssi"]
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        device = next(
+            (d for d in self.coordinator.data if d["friendly_id"] == self._friendly_id),
+            None,
+        )
+        if device is None:
+            return
+        self._attr_native_value = device["rssi"]
+        self.async_write_ha_state()

custom_components/trmnl/sensor.py

@@ -0,0 +1,28 @@
+"""TRMNL sensor platform for Home Assistant."""
+
+import logging
+
+from homeassistant.core import HomeAssistant
+
+from .battery_sensor import BatterySensor
+from .coordinator import TrmnlCoordinator
+from .rssi_sensor import RssiSensor
+from .trmnl_context import TrmnlConfigEntry
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: TrmnlConfigEntry, async_add_entities
+) -> bool:
+    """Set up a TRMNL sensor."""
+
+    coordinator = TrmnlCoordinator(hass, entry)
+    await coordinator.async_config_entry_first_refresh()
+
+    async_add_entities(
+        BatterySensor(coordinator, device) for device in coordinator.data
+    )
+    async_add_entities(RssiSensor(coordinator, device) for device in coordinator.data)
+
+    return True

custom_components/trmnl/trmnl_api.py

@@ -0,0 +1,58 @@
+"""TRMNL API client."""
+
+from aiohttp import ClientResponse, ClientSession
+
+type JsonDict = dict[str, str | int | float | bool | None]
+
+
+class TrmnlApiError(Exception):
+    """Exception raised for general API errors."""
+
+
+class TrmnlApiAuthError(TrmnlApiError):
+    """Exception raised for API authentication errors."""
+
+
+class TrmnlApi:
+    """Class to make authenticated requests."""
+
+    def __init__(self, base_url: str, api_key: str) -> None:
+        """Initialize the client."""
+        self.base_url = base_url
+        self.api_key = api_key
+
+    async def get_devices(self) -> list[dict]:
+        """Get the devices from the API."""
+        return await self._get("/devices")
+
+    async def validate_api_key(self) -> bool:
+        """Check if the API key is valid."""
+        try:
+            await self._get("/devices")
+        except TrmnlApiError:
+            return False
+        return True
+
+    @property
+    def _headers(self) -> dict[str, str]:
+        """Return the headers for the request."""
+        return {
+            "accept": "application/json",
+            "content-type": "application/json",
+            "authorization": f"bearer {self.api_key}",
+        }
+
+    async def _get(self, route: str) -> JsonDict:
+        """Make a GET request to the API."""
+        async with ClientSession() as session:
+            url: str = f"{self.base_url}{route}"
+            response: ClientResponse = await session.get(url, headers=self._headers)
+            json: JsonDict = await response.json()
+
+            if response.status == 200:
+                return json["data"]
+
+            if response.status == 401:
+                raise TrmnlApiAuthError("Invalid API key")
+
+            raise TrmnlApiError(f"Error {response.status}: {json['error']}")

custom_components/trmnl/trmnl_context.py

@@ -0,0 +1,16 @@
+"""Runtime context for the TRMNL config entry."""
+
+import dataclasses
+
+from homeassistant.config_entries import ConfigEntry
+
+from .trmnl_api import TrmnlApi
+
+type TrmnlConfigEntry = ConfigEntry[TrmnlContext]
+
+
+@dataclasses.dataclass
+class TrmnlContext:
+    """Runtime context for the TRMNL config entry."""
+
+    api: TrmnlApi