mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 16:57:53 +00:00
Add binary_sensor to Starlink (#85409)
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Jack Boswell
GitHub
parent
ae9a57b2a8
commit
43cc8a1ebf
@ -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
|
||||
|
||||
@ -8,7 +8,7 @@ from homeassistant.core import HomeAssistant
|
||||
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:
|
||||
|
||||
129
homeassistant/components/starlink/binary_sensor.py
Normal file
129
homeassistant/components/starlink/binary_sensor.py
Normal file
@ -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"],
|
||||
),
|
||||
]
|
||||
@ -1,11 +1,19 @@
|
||||
"""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
|
||||
@ -13,7 +21,16 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
|
||||
_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 @@ class StarlinkUpdateCoordinator(DataUpdateCoordinator[StatusDict]):
|
||||
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
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -1,71 +1,26 @@
|
||||
"""Contains sensors exposed by the Starlink integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
|
||||
from homeassistant.components.sensor import (
|
||||
SensorDeviceClass,
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
SensorStateClass,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import DEGREE, UnitOfDataRate, UnitOfTime
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity import EntityCategory
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import StateType
|
||||
|
||||
from .const import DOMAIN
|
||||
from .entity import StarlinkSensorEntity, StarlinkSensorEntityDescription
|
||||
|
||||
SENSORS: tuple[StarlinkSensorEntityDescription, ...] = (
|
||||
StarlinkSensorEntityDescription(
|
||||
key="ping",
|
||||
name="Ping",
|
||||
icon="mdi:speedometer",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfTime.MILLISECONDS,
|
||||
value_fn=lambda data: round(data["pop_ping_latency_ms"]),
|
||||
),
|
||||
StarlinkSensorEntityDescription(
|
||||
key="azimuth",
|
||||
name="Azimuth",
|
||||
icon="mdi:compass",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
native_unit_of_measurement=DEGREE,
|
||||
value_fn=lambda data: round(data["direction_azimuth"]),
|
||||
),
|
||||
StarlinkSensorEntityDescription(
|
||||
key="elevation",
|
||||
name="Elevation",
|
||||
icon="mdi:compass",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
native_unit_of_measurement=DEGREE,
|
||||
value_fn=lambda data: round(data["direction_elevation"]),
|
||||
),
|
||||
StarlinkSensorEntityDescription(
|
||||
key="uplink_throughput",
|
||||
name="Uplink throughput",
|
||||
icon="mdi:upload",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfDataRate.BITS_PER_SECOND,
|
||||
value_fn=lambda data: round(data["uplink_throughput_bps"]),
|
||||
),
|
||||
StarlinkSensorEntityDescription(
|
||||
key="downlink_throughput",
|
||||
name="Downlink throughput",
|
||||
icon="mdi:download",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfDataRate.BITS_PER_SECOND,
|
||||
value_fn=lambda data: round(data["downlink_throughput_bps"]),
|
||||
),
|
||||
StarlinkSensorEntityDescription(
|
||||
key="last_boot_time",
|
||||
name="Last boot time",
|
||||
icon="mdi:clock",
|
||||
device_class=SensorDeviceClass.TIMESTAMP,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
value_fn=lambda data: datetime.now().astimezone()
|
||||
- timedelta(seconds=data["uptime"]),
|
||||
),
|
||||
)
|
||||
from .coordinator import StarlinkData, StarlinkUpdateCoordinator
|
||||
from .entity import StarlinkEntity
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
@ -77,3 +32,93 @@ async def async_setup_entry(
|
||||
async_add_entities(
|
||||
StarlinkSensorEntity(coordinator, description) for description in SENSORS
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class StarlinkSensorEntityDescriptionMixin:
|
||||
"""Mixin for required keys."""
|
||||
|
||||
value_fn: Callable[[StarlinkData], datetime | StateType]
|
||||
|
||||
|
||||
@dataclass
|
||||
class StarlinkSensorEntityDescription(
|
||||
SensorEntityDescription, StarlinkSensorEntityDescriptionMixin
|
||||
):
|
||||
"""Describes a Starlink sensor entity."""
|
||||
|
||||
|
||||
class StarlinkSensorEntity(StarlinkEntity, SensorEntity):
|
||||
"""A SensorEntity for Starlink devices. Handles creating unique IDs."""
|
||||
|
||||
entity_description: StarlinkSensorEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: StarlinkUpdateCoordinator,
|
||||
description: StarlinkSensorEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.entity_description = description
|
||||
self._attr_unique_id = f"{self.coordinator.data.status['id']}_{description.key}"
|
||||
|
||||
@property
|
||||
def native_value(self) -> StateType | datetime:
|
||||
"""Calculate the sensor value from the entity description."""
|
||||
return self.entity_description.value_fn(self.coordinator.data)
|
||||
|
||||
|
||||
SENSORS: tuple[StarlinkSensorEntityDescription, ...] = (
|
||||
StarlinkSensorEntityDescription(
|
||||
key="ping",
|
||||
name="Ping",
|
||||
icon="mdi:speedometer",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfTime.MILLISECONDS,
|
||||
value_fn=lambda data: round(data.status["pop_ping_latency_ms"]),
|
||||
),
|
||||
StarlinkSensorEntityDescription(
|
||||
key="azimuth",
|
||||
name="Azimuth",
|
||||
icon="mdi:compass",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
native_unit_of_measurement=DEGREE,
|
||||
value_fn=lambda data: round(data.status["direction_azimuth"]),
|
||||
),
|
||||
StarlinkSensorEntityDescription(
|
||||
key="elevation",
|
||||
name="Elevation",
|
||||
icon="mdi:compass",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
native_unit_of_measurement=DEGREE,
|
||||
value_fn=lambda data: round(data.status["direction_elevation"]),
|
||||
),
|
||||
StarlinkSensorEntityDescription(
|
||||
key="uplink_throughput",
|
||||
name="Uplink throughput",
|
||||
icon="mdi:upload",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfDataRate.BITS_PER_SECOND,
|
||||
value_fn=lambda data: round(data.status["uplink_throughput_bps"]),
|
||||
),
|
||||
StarlinkSensorEntityDescription(
|
||||
key="downlink_throughput",
|
||||
name="Downlink throughput",
|
||||
icon="mdi:download",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfDataRate.BITS_PER_SECOND,
|
||||
value_fn=lambda data: round(data.status["downlink_throughput_bps"]),
|
||||
),
|
||||
StarlinkSensorEntityDescription(
|
||||
key="last_boot_time",
|
||||
name="Last boot time",
|
||||
icon="mdi:clock",
|
||||
device_class=SensorDeviceClass.TIMESTAMP,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
value_fn=lambda data: datetime.now().astimezone()
|
||||
- timedelta(seconds=data.status["uptime"]),
|
||||
),
|
||||
)
|
||||
|
||||
@ -3,7 +3,10 @@ from unittest.mock import patch
|
||||
|
||||
from starlink_grpc import StatusDict
|
||||
|
||||
from homeassistant.components.starlink.coordinator import StarlinkUpdateCoordinator
|
||||
from homeassistant.components.starlink.coordinator import (
|
||||
StarlinkData,
|
||||
StarlinkUpdateCoordinator,
|
||||
)
|
||||
|
||||
SETUP_ENTRY_PATCHER = patch(
|
||||
"homeassistant.components.starlink.async_setup_entry", return_value=True
|
||||
@ -12,7 +15,11 @@ SETUP_ENTRY_PATCHER = patch(
|
||||
COORDINATOR_SUCCESS_PATCHER = patch.object(
|
||||
StarlinkUpdateCoordinator,
|
||||
"_async_update_data",
|
||||
return_value=StatusDict(id="1", software_version="1", hardware_version="1"),
|
||||
return_value=StarlinkData(
|
||||
StatusDict(id="1", software_version="1", hardware_version="1"),
|
||||
{},
|
||||
{},
|
||||
),
|
||||
)
|
||||
|
||||
DEVICE_FOUND_PATCHER = patch(
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user