improve sensors, add maneuver sensor

custom_components/ais_receiver/sensor.py

@@ -1,41 +1,94 @@
 """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,
     SensorEntity,
     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")