[ci skip] home assistant integration

2026-02-19 14:34:58 +11:00
parent 7d0ce52c27
commit d72e88ab7b
15 changed files with 1397 additions and 1 deletions
--- a/custom_components/victron_mk2_mqtt/sensor.py
+++ b/custom_components/victron_mk2_mqtt/sensor.py
@@ -0,0 +1,148 @@
+"""Sensor entities for Victron MK2 MQTT integration."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+from homeassistant.components.sensor import SensorEntity, SensorStateClass
+from homeassistant.const import (
+    EntityCategory,
+    PERCENTAGE,
+    UnitOfElectricCurrent,
+    UnitOfElectricPotential,
+    UnitOfFrequency,
+)
+from homeassistant.core import HomeAssistant
+
+from .const import DATA_BRIDGE, DOMAIN
+from .coordinator import VictronMqttBridge
+from .entity import VictronMqttEntity
+
+
+@dataclass(frozen=True)
+class MetricDescription:
+    """Description for a telemetry-backed sensor."""
+
+    key: str
+    name: str
+    value_fn: Callable[[VictronMqttBridge], Any]
+    unit: str | None = None
+    state_class: SensorStateClass | None = SensorStateClass.MEASUREMENT
+    entity_category: EntityCategory | None = None
+
+
+METRICS: tuple[MetricDescription, ...] = (
+    MetricDescription(
+        key="battery_voltage",
+        name="Battery Voltage",
+        value_fn=lambda bridge: bridge.metric_float("BatVoltage"),
+        unit=UnitOfElectricPotential.VOLT,
+    ),
+    MetricDescription(
+        key="battery_current",
+        name="Battery Current",
+        value_fn=lambda bridge: bridge.metric_float("BatCurrent"),
+        unit=UnitOfElectricCurrent.AMPERE,
+    ),
+    MetricDescription(
+        key="battery_charge",
+        name="Battery Charge",
+        value_fn=lambda bridge: (
+            bridge.metric_float("ChargeState") * 100.0
+            if bridge.metric_float("ChargeState") is not None
+            else None
+        ),
+        unit=PERCENTAGE,
+    ),
+    MetricDescription(
+        key="input_voltage",
+        name="Input Voltage",
+        value_fn=lambda bridge: bridge.metric_float("InVoltage"),
+        unit=UnitOfElectricPotential.VOLT,
+    ),
+    MetricDescription(
+        key="input_current",
+        name="Input Current",
+        value_fn=lambda bridge: bridge.metric_float("InCurrent"),
+        unit=UnitOfElectricCurrent.AMPERE,
+    ),
+    MetricDescription(
+        key="input_frequency",
+        name="Input Frequency",
+        value_fn=lambda bridge: bridge.metric_float("InFrequency"),
+        unit=UnitOfFrequency.HERTZ,
+    ),
+    MetricDescription(
+        key="output_voltage",
+        name="Output Voltage",
+        value_fn=lambda bridge: bridge.metric_float("OutVoltage"),
+        unit=UnitOfElectricPotential.VOLT,
+    ),
+    MetricDescription(
+        key="output_current",
+        name="Output Current",
+        value_fn=lambda bridge: bridge.metric_float("OutCurrent"),
+        unit=UnitOfElectricCurrent.AMPERE,
+    ),
+    MetricDescription(
+        key="output_frequency",
+        name="Output Frequency",
+        value_fn=lambda bridge: bridge.metric_float("OutFrequency"),
+        unit=UnitOfFrequency.HERTZ,
+    ),
+    MetricDescription(
+        key="input_power",
+        name="Input Power",
+        value_fn=lambda bridge: _product(bridge.metric_float("InVoltage"), bridge.metric_float("InCurrent")),
+        unit="VA",
+    ),
+    MetricDescription(
+        key="output_power",
+        name="Output Power",
+        value_fn=lambda bridge: _product(bridge.metric_float("OutVoltage"), bridge.metric_float("OutCurrent")),
+        unit="VA",
+    ),
+    MetricDescription(
+        key="last_command_error",
+        name="Last Command Error",
+        value_fn=lambda bridge: bridge.last_error,
+        state_class=None,
+        entity_category=EntityCategory.DIAGNOSTIC,
+    ),
+)
+
+
+def _product(a: float | None, b: float | None) -> float | None:
+    if a is None or b is None:
+        return None
+    return a * b
+
+
+async def async_setup_platform(
+    hass: HomeAssistant,
+    config: dict[str, Any],
+    async_add_entities,
+    discovery_info: dict[str, Any] | None = None,
+) -> None:
+    """Set up Victron telemetry sensors."""
+    bridge: VictronMqttBridge = hass.data[DOMAIN][DATA_BRIDGE]
+    async_add_entities(VictronMetricSensor(bridge, metric) for metric in METRICS)
+
+
+class VictronMetricSensor(VictronMqttEntity, SensorEntity):
+    """Generic telemetry sensor."""
+
+    def __init__(self, bridge: VictronMqttBridge, description: MetricDescription) -> None:
+        super().__init__(bridge)
+        self.entity_description = description
+        self._attr_unique_id = f"{bridge.topic_root}_{description.key}"
+        self._attr_name = description.name
+        self._attr_native_unit_of_measurement = description.unit
+        self._attr_state_class = description.state_class
+        self._attr_entity_category = description.entity_category
+
+    @property
+    def native_value(self):
+        return self.entity_description.value_fn(self.bridge)