Add binary_sensor to Starlink

.coveragerc

@@ -1225,6 +1225,7 @@ omit =
     homeassistant/components/squeezebox/__init__.py
     homeassistant/components/squeezebox/browse_media.py
     homeassistant/components/squeezebox/media_player.py
+    homeassistant/components/starlink/binary_sensor.py
     homeassistant/components/starlink/coordinator.py
     homeassistant/components/starlink/entity.py
     homeassistant/components/starlink/sensor.py

homeassistant/components/starlink/__init__.py

@@ -8,7 +8,7 @@
 from .const import DOMAIN
 from .coordinator import StarlinkUpdateCoordinator
 
-PLATFORMS: list[Platform] = [Platform.SENSOR]
+PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:

homeassistant/components/starlink/binary_sensor.py

@@ -0,0 +1,129 @@
+"""Contains binary sensors exposed by the Starlink integration."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorDeviceClass,
+    BinarySensorEntity,
+    BinarySensorEntityDescription,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import EntityCategory
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN
+from .coordinator import StarlinkData, StarlinkUpdateCoordinator
+from .entity import StarlinkEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up all binary sensors for this entry."""
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+
+    async_add_entities(
+        StarlinkBinarySensorEntity(coordinator, description)
+        for description in BINARY_SENSORS
+    )
+
+
+@dataclass
+class StarlinkBinarySensorEntityDescriptionMixin:
+    """Mixin for required keys."""
+
+    value_fn: Callable[[StarlinkData], bool | None]
+
+
+@dataclass
+class StarlinkBinarySensorEntityDescription(
+    BinarySensorEntityDescription, StarlinkBinarySensorEntityDescriptionMixin
+):
+    """Describes a Starlink binary sensor entity."""
+
+
+class StarlinkBinarySensorEntity(StarlinkEntity, BinarySensorEntity):
+    """A BinarySensorEntity for Starlink devices. Handles creating unique IDs."""
+
+    entity_description: StarlinkBinarySensorEntityDescription
+
+    def __init__(
+        self,
+        coordinator: StarlinkUpdateCoordinator,
+        description: StarlinkBinarySensorEntityDescription,
+    ) -> None:
+        """Initialize the binary sensor."""
+        super().__init__(coordinator)
+        self.entity_description = description
+        self._attr_unique_id = f"{self.coordinator.data.status['id']}_{description.key}"
+
+    @property
+    def is_on(self) -> bool | None:
+        """Calculate the binary sensor value from the entity description."""
+        return self.entity_description.value_fn(self.coordinator.data)
+
+
+BINARY_SENSORS = [
+    StarlinkBinarySensorEntityDescription(
+        key="roaming",
+        name="Roaming mode",
+        value_fn=lambda data: data.alert["alert_roaming"],
+    ),
+    StarlinkBinarySensorEntityDescription(
+        key="currently_obstructed",
+        name="Obstructed",
+        device_class=BinarySensorDeviceClass.PROBLEM,
+        value_fn=lambda data: data.status["currently_obstructed"],
+    ),
+    StarlinkBinarySensorEntityDescription(
+        key="heating",
+        name="Heating",
+        entity_category=EntityCategory.DIAGNOSTIC,
+        value_fn=lambda data: data.alert["alert_is_heating"],
+    ),
+    StarlinkBinarySensorEntityDescription(
+        key="power_save_idle",
+        name="Idle",
+        entity_category=EntityCategory.DIAGNOSTIC,
+        value_fn=lambda data: data.alert["alert_is_power_save_idle"],
+    ),
+    StarlinkBinarySensorEntityDescription(
+        key="mast_near_vertical",
+        name="Mast near vertical",
+        device_class=BinarySensorDeviceClass.PROBLEM,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        value_fn=lambda data: data.alert["alert_mast_not_near_vertical"],
+    ),
+    StarlinkBinarySensorEntityDescription(
+        key="motors_stuck",
+        name="Motors stuck",
+        device_class=BinarySensorDeviceClass.PROBLEM,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        value_fn=lambda data: data.alert["alert_motors_stuck"],
+    ),
+    StarlinkBinarySensorEntityDescription(
+        key="slow_ethernet",
+        name="Ethernet speeds",
+        device_class=BinarySensorDeviceClass.PROBLEM,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        value_fn=lambda data: data.alert["alert_slow_ethernet_speeds"],
+    ),
+    StarlinkBinarySensorEntityDescription(
+        key="thermal_throttle",
+        name="Thermal throttle",
+        device_class=BinarySensorDeviceClass.PROBLEM,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        value_fn=lambda data: data.alert["alert_thermal_throttle"],
+    ),
+    StarlinkBinarySensorEntityDescription(
+        key="unexpected_location",
+        name="Unexpected location",
+        device_class=BinarySensorDeviceClass.PROBLEM,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        value_fn=lambda data: data.alert["alert_unexpected_location"],
+    ),
+]

homeassistant/components/starlink/coordinator.py

@@ -1,19 +1,36 @@
 """Contains the shared Coordinator for Starlink systems."""
 from __future__ import annotations
 
+from dataclasses import dataclass
 from datetime import timedelta
 import logging
 
 import async_timeout
-from starlink_grpc import ChannelContext, GrpcError, StatusDict, status_data
+from starlink_grpc import (
+    AlertDict,
+    ChannelContext,
+    GrpcError,
+    ObstructionDict,
+    StatusDict,
+    status_data,
+)
 
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class StarlinkUpdateCoordinator(DataUpdateCoordinator[StatusDict]):
+@dataclass
+class StarlinkData:
+    """Contains data pulled from the Starlink system."""
+
+    status: StatusDict
+    obstruction: ObstructionDict
+    alert: AlertDict
+
+
+class StarlinkUpdateCoordinator(DataUpdateCoordinator[StarlinkData]):
     """Coordinates updates between all Starlink sensors defined in this file."""
 
     def __init__(self, hass: HomeAssistant, name: str, url: str) -> None:
@@ -27,12 +44,12 @@ def __init__(self, hass: HomeAssistant, name: str, url: str) -> None:
             update_interval=timedelta(seconds=5),
         )
 
-    async def _async_update_data(self) -> StatusDict:
+    async def _async_update_data(self) -> StarlinkData:
         async with async_timeout.timeout(4):
             try:
                 status = await self.hass.async_add_executor_job(
                     status_data, self.channel_context
                 )
-                return status[0]
+                return StarlinkData(*status)
             except GrpcError as exc:
                 raise UpdateFailed from exc

homeassistant/components/starlink/entity.py

@@ -1,64 +1,32 @@
 """Contains base entity classes for Starlink entities."""
 from __future__ import annotations
 
-from collections.abc import Callable
-from dataclasses import dataclass
-from datetime import datetime
-
-from starlink_grpc import StatusDict
-
-from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
-from homeassistant.helpers.entity import DeviceInfo
-from homeassistant.helpers.typing import StateType
+from homeassistant.helpers.entity import DeviceInfo, Entity
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import DOMAIN
 from .coordinator import StarlinkUpdateCoordinator
 
 
-@dataclass
-class StarlinkSensorEntityDescriptionMixin:
-    """Mixin for required keys."""
-
-    value_fn: Callable[[StatusDict], datetime | StateType]
-
-
-@dataclass
-class StarlinkSensorEntityDescription(
-    SensorEntityDescription, StarlinkSensorEntityDescriptionMixin
-):
-    """Describes a Starlink sensor entity."""
-
-
-class StarlinkSensorEntity(CoordinatorEntity[StarlinkUpdateCoordinator], SensorEntity):
-    """A SensorEntity that is registered under the Starlink device, and handles creating unique IDs."""
-
-    entity_description: StarlinkSensorEntityDescription
+class StarlinkEntity(CoordinatorEntity[StarlinkUpdateCoordinator], Entity):
+    """A base Entity that is registered under a Starlink device."""
 
     _attr_has_entity_name = True
 
     def __init__(
         self,
         coordinator: StarlinkUpdateCoordinator,
-        description: StarlinkSensorEntityDescription,
     ) -> None:
-        """Initialize the sensor and set the update coordinator."""
+        """Initialize the device info and set the update coordinator."""
         super().__init__(coordinator)
-        self.entity_description = description
-        self._attr_unique_id = f"{self.coordinator.data['id']}_{description.key}"
         self._attr_device_info = DeviceInfo(
             identifiers={
-                (DOMAIN, self.coordinator.data["id"]),
+                (DOMAIN, self.coordinator.data.status["id"]),
             },
-            sw_version=self.coordinator.data["software_version"],
-            hw_version=self.coordinator.data["hardware_version"],
+            sw_version=self.coordinator.data.status["software_version"],
+            hw_version=self.coordinator.data.status["hardware_version"],
             name="Starlink",
             configuration_url=f"http://{self.coordinator.channel_context.target.split(':')[0]}",
             manufacturer="SpaceX",
             model="Starlink",
         )
-
-    @property
-    def native_value(self) -> StateType | datetime:
-        """Calculate the sensor value from the entity description."""
-        return self.entity_description.value_fn(self.coordinator.data)