Add Sensor for Refoss Integration

.coveragerc

@@ -1120,6 +1120,7 @@ omit =
     homeassistant/components/refoss/bridge.py
     homeassistant/components/refoss/coordinator.py
     homeassistant/components/refoss/entity.py
+    homeassistant/components/refoss/sensor.py
     homeassistant/components/refoss/switch.py
     homeassistant/components/refoss/util.py
     homeassistant/components/rejseplanen/sensor.py

homeassistant/components/refoss/__init__.py

@@ -15,6 +15,7 @@
 from .util import refoss_discovery_server
 
 PLATFORMS: Final = [
+    Platform.SENSOR,
     Platform.SWITCH,
 ]
 

homeassistant/components/refoss/const.py

@@ -2,8 +2,18 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
+from dataclasses import dataclass, field
 from logging import Logger, getLogger
 
+from homeassistant.components.sensor import SensorDeviceClass
+from homeassistant.const import (
+    UnitOfElectricCurrent,
+    UnitOfElectricPotential,
+    UnitOfEnergy,
+    UnitOfPower,
+)
+
 _LOGGER: Logger = getLogger(__package__)
 
 COORDINATORS = "coordinators"
@@ -19,3 +29,65 @@
 COORDINATOR = "coordinator"
 
 MAX_ERRORS = 2
+
+ENERGY = "energy"
+ENERGY_RETURNED = "energy_returned"
+
+
+@dataclass
+class UnitOfMeasurement:
+    """Describes a unit of measurement."""
+
+    unit: str
+    device_classes: set[str]
+    aliases: set[str] = field(default_factory=set)
+    conversion_unit: str | None = None
+    conversion_fn: Callable[[float], str] | None = None
+
+
+UNITS = (
+    UnitOfMeasurement(
+        unit=UnitOfElectricPotential.VOLT,
+        aliases={"v"},
+        device_classes={SensorDeviceClass.VOLTAGE},
+        conversion_fn=lambda x: f"{x / 1000:.2f}",
+    ),
+    UnitOfMeasurement(
+        unit=UnitOfElectricCurrent.AMPERE,
+        aliases={"a"},
+        device_classes={SensorDeviceClass.CURRENT},
+        conversion_fn=lambda x: f"{x / 1000:.2f}",
+    ),
+    UnitOfMeasurement(
+        unit=UnitOfPower.WATT,
+        aliases={"w"},
+        device_classes={SensorDeviceClass.POWER},
+        conversion_fn=lambda x: f"{x / 1000:.2f}",
+    ),
+    UnitOfMeasurement(
+        unit=UnitOfEnergy.KILO_WATT_HOUR,
+        aliases={"kwh"},
+        device_classes={SensorDeviceClass.ENERGY},
+        conversion_fn=lambda x: f"{x / 1000:.2f}",
+    ),
+)
+
+
+DEVICE_CLASS_UNITS: dict[str, dict[str, UnitOfMeasurement]] = {}
+for uom in UNITS:
+    for device_class in uom.device_classes:
+        DEVICE_CLASS_UNITS.setdefault(device_class, {})[uom.unit] = uom
+        for unit_alias in uom.aliases:
+            DEVICE_CLASS_UNITS[device_class][unit_alias] = uom
+
+
+CHANNEL_DISPLAY_NAME: dict[str, dict[int, str]] = {
+    "em06": {
+        1: "A1",
+        2: "B1",
+        3: "C1",
+        4: "A2",
+        5: "B2",
+        6: "C2",
+    }
+}

homeassistant/components/refoss/manifest.json

@@ -5,5 +5,5 @@
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/refoss",
   "iot_class": "local_polling",
-  "requirements": ["refoss-ha==1.2.0"]
+  "requirements": ["refoss-ha==1.2.1"]
 }

homeassistant/components/refoss/sensor.py

@@ -0,0 +1,194 @@
+"""Support for refoss sensors."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from refoss_ha.controller.electricity import ElectricityXMix
+
+from homeassistant.components.sensor import (
+    SensorDeviceClass,
+    SensorEntity,
+    SensorEntityDescription,
+    SensorStateClass,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import (
+    UnitOfElectricCurrent,
+    UnitOfElectricPotential,
+    UnitOfPower,
+)
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import StateType
+
+from .bridge import RefossDataUpdateCoordinator
+from .const import (
+    CHANNEL_DISPLAY_NAME,
+    COORDINATORS,
+    DEVICE_CLASS_UNITS,
+    DISPATCH_DEVICE_DISCOVERED,
+    DOMAIN,
+    ENERGY,
+    ENERGY_RETURNED,
+    UnitOfEnergy,
+    UnitOfMeasurement,
+)
+from .entity import RefossEntity
+
+
+@dataclass(frozen=True)
+class RefossSensorEntityDescription(SensorEntityDescription):
+    """Describes Refoss sensor entity."""
+
+    subkey: str | None = None
+
+
+SENSORS: dict[str, tuple[RefossSensorEntityDescription, ...]] = {
+    "em06": (
+        RefossSensorEntityDescription(
+            key="power",
+            name="Power",
+            device_class=SensorDeviceClass.POWER,
+            state_class=SensorStateClass.MEASUREMENT,
+            native_unit_of_measurement=UnitOfPower.WATT,
+            subkey="power",
+        ),
+        RefossSensorEntityDescription(
+            key="voltage",
+            name="Voltage",
+            device_class=SensorDeviceClass.VOLTAGE,
+            state_class=SensorStateClass.MEASUREMENT,
+            native_unit_of_measurement=UnitOfElectricPotential.VOLT,
+            subkey="voltage",
+        ),
+        RefossSensorEntityDescription(
+            key="current",
+            name="Current",
+            device_class=SensorDeviceClass.CURRENT,
+            state_class=SensorStateClass.MEASUREMENT,
+            native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
+            subkey="current",
+        ),
+        RefossSensorEntityDescription(
+            key="factor",
+            name="Power Factor",
+            device_class=SensorDeviceClass.POWER_FACTOR,
+            state_class=SensorStateClass.MEASUREMENT,
+            subkey="factor",
+        ),
+        RefossSensorEntityDescription(
+            key="energy",
+            name="This Month Energy",
+            device_class=SensorDeviceClass.ENERGY,
+            state_class=SensorStateClass.TOTAL,
+            native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
+            subkey="mConsume",
+        ),
+        RefossSensorEntityDescription(
+            key="energy_returned",
+            name="This Month Energy Returned",
+            device_class=SensorDeviceClass.ENERGY,
+            state_class=SensorStateClass.TOTAL,
+            native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
+            subkey="mConsume",
+        ),
+    ),
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the Refoss device from a config entry."""
+
+    @callback
+    def init_device(coordinator):
+        """Register the device."""
+        device = coordinator.device
+
+        if not isinstance(device, ElectricityXMix):
+            return
+        descriptions = SENSORS.get(device.device_type)
+        new_entities = []
+        for channel in device.channels:
+            for description in descriptions:
+                entity = RefossSensor(
+                    coordinator=coordinator,
+                    channel=channel,
+                    description=description,
+                )
+                new_entities.append(entity)
+
+        async_add_entities(new_entities)
+
+    for coordinator in hass.data[DOMAIN][COORDINATORS]:
+        init_device(coordinator)
+
+    config_entry.async_on_unload(
+        async_dispatcher_connect(hass, DISPATCH_DEVICE_DISCOVERED, init_device)
+    )
+
+
+class RefossSensor(RefossEntity, SensorEntity):
+    """Refoss Sensor Device."""
+
+    entity_description: RefossSensorEntityDescription
+    _uom: UnitOfMeasurement | None = None
+
+    def __init__(
+        self,
+        coordinator: RefossDataUpdateCoordinator,
+        channel: int,
+        description: RefossSensorEntityDescription,
+    ) -> None:
+        """Init Refoss sensor."""
+        super().__init__(coordinator, channel)
+        self.entity_description = description
+        self._attr_unique_id = f"{super().unique_id}{description.key}"
+        device_type = coordinator.device.device_type
+        self._attr_name = (
+            f"{CHANNEL_DISPLAY_NAME[device_type][channel]} {description.name}"
+        )
+
+        if self.device_class is not None:
+            if (
+                self.native_unit_of_measurement is None
+                or self.device_class not in DEVICE_CLASS_UNITS
+            ):
+                self._attr_device_class = None
+                return
+
+            uoms = DEVICE_CLASS_UNITS[self.device_class]
+            self._uom = uoms.get(self.native_unit_of_measurement) or uoms.get(
+                self.native_unit_of_measurement.lower()
+            )
+            if self._uom is None:
+                self._attr_device_class = None
+                return
+            self._attr_native_unit_of_measurement = (
+                self._uom.conversion_unit or self._uom.unit
+            )
+
+    @property
+    def native_value(self) -> StateType:
+        """Return the native value."""
+        value = self.coordinator.device.get_value(
+            self.channel_id, self.entity_description.subkey
+        )
+        if value is None:
+            return None
+        if (self.entity_description.key == ENERGY and value < 0) or (
+            self.entity_description.key == ENERGY_RETURNED and value > 0
+        ):
+            value = 0
+
+        if self._uom and self._uom.conversion_fn is not None:
+            return self._uom.conversion_fn(value)
+
+        if isinstance(value, float):
+            return f"{value:.2f}"
+        return value

requirements_all.txt

@@ -2433,7 +2433,7 @@ rapt-ble==0.1.2
 raspyrfm-client==1.2.8
 
 # homeassistant.components.refoss
-refoss-ha==1.2.0
+refoss-ha==1.2.1
 
 # homeassistant.components.rainmachine
 regenmaschine==2024.03.0

requirements_test_all.txt

@@ -1891,7 +1891,7 @@ radiotherm==2.1.0
 rapt-ble==0.1.2
 
 # homeassistant.components.refoss
-refoss-ha==1.2.0
+refoss-ha==1.2.1
 
 # homeassistant.components.rainmachine
 regenmaschine==2024.03.0