149 lines
4.6 KiB
Python
149 lines
4.6 KiB
Python
"""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)
|