diff --git a/custom_components/ais_receiver/sensor.py b/custom_components/ais_receiver/sensor.py index ac62176..d12bcf2 100644 --- a/custom_components/ais_receiver/sensor.py +++ b/custom_components/ais_receiver/sensor.py @@ -1,6 +1,9 @@ """Sensors for AIS receiver.""" -from pyais.constants import NavigationStatus +from collections.abc import Callable +from dataclasses import dataclass + +from pyais.constants import ManeuverIndicator, NavigationStatus from homeassistant.components.sensor import ( SensorDeviceClass, @@ -8,34 +11,84 @@ SensorEntityDescription, ) from homeassistant.config_entries import ConfigEntry -from homeassistant.const import UnitOfSpeed +from homeassistant.const import DEGREE, UnitOfSpeed from homeassistant.core import Event, HomeAssistant from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback from .const import CONF_MMSIS, DOMAIN -SENSORS: tuple[SensorEntityDescription, ...] = ( - SensorEntityDescription( - key="course", name="course", entity_registry_enabled_default=False + +@dataclass(frozen=True, kw_only=True) +class AisSensorEntityDescription(SensorEntityDescription): + """A class that describes AEMET OpenData sensor entities.""" + + keys: list[str] | None = None + value_fn: Callable[[str | int | float], float | int | str | None] = ( + lambda value: value + ) + + +def _course_value(value: float) -> float | None: + if value == 360: + return None + return value + + +def _heading_value(value: int) -> int | None: + if value == 511: + return None + return value + + +def _turn_value(value: float) -> float | None: + if value == -128.0: + return None + return (value / 4.733) ^ 2 + + +SENSORS: tuple[AisSensorEntityDescription, ...] = ( + AisSensorEntityDescription( + key="course", + name="course", + native_unit_of_measurement=DEGREE, + value_fn=_course_value, + icon="mdi:compass", ), - SensorEntityDescription( - key="heading", name="heading", entity_registry_enabled_default=False + AisSensorEntityDescription( + key="heading", + name="heading", + native_unit_of_measurement=DEGREE, + value_fn=_heading_value, + icon="mdi:angle-acute", ), - SensorEntityDescription( + AisSensorEntityDescription( + key="maneuver", + name="maneuver", + device_class=SensorDeviceClass.ENUM, + options={status.name for status in ManeuverIndicator}, + value_fn=lambda value: ManeuverIndicator(value).name, + ), + AisSensorEntityDescription( key="speed", name="speed", device_class=SensorDeviceClass.SPEED, native_unit_of_measurement=UnitOfSpeed.KNOTS, + value_fn=lambda value: value / 10, ), - SensorEntityDescription( + AisSensorEntityDescription( key="status", name="status", device_class=SensorDeviceClass.ENUM, options={status.name for status in NavigationStatus}, + value_fn=lambda value: NavigationStatus(value).name, + icon="mdi:check-decagram", ), - SensorEntityDescription( - key="turn", name="turn", entity_registry_enabled_default=False + AisSensorEntityDescription( + key="turn", + name="turn", + native_unit_of_measurement="°/min", + value_fn=_turn_value, ), ) @@ -56,8 +109,9 @@ class AisReceiverSensorEntity(SensorEntity): """Represent a tracked device.""" _attr_should_poll = False + entity_description: AisSensorEntityDescription - def __init__(self, mmsi: str, description: SensorEntityDescription) -> None: + def __init__(self, mmsi: str, description: AisSensorEntityDescription) -> None: """Set up AIS receiver tracker entity.""" self.entity_description = description self._mmsi = mmsi @@ -75,13 +129,8 @@ async def async_update_data_from_msg(self, event: Event) -> None: msg.get("msg_type") in [1, 2, 3] and (value := msg.get(self.entity_description.key)) is not None ): - if self.entity_description.key == "speed": - value = value / 10 - if self.entity_description.key == "status": - value = NavigationStatus(value).name - self._attr_native_value = value - - if msg.get("msg_type") == 5: + self._attr_native_value = self.entity_description.value_fn(value) + elif msg.get("msg_type") == 5: self._attr_extra_state_attributes["shipname"] = msg.get("shipname") self._attr_extra_state_attributes["callsign"] = msg.get("callsign")