mirror of
https://github.com/home-assistant/core.git
synced 2026-05-08 17:49:37 +01:00
Enrich DeviceInfo with meter metadata in route_b_smart_meter (#164006)
Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
@@ -2,6 +2,7 @@
|
||||
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
import time
|
||||
|
||||
from momonga import Momonga, MomongaError
|
||||
|
||||
@@ -28,9 +29,20 @@ class BRouteData:
|
||||
type BRouteConfigEntry = ConfigEntry[BRouteUpdateCoordinator]
|
||||
|
||||
|
||||
@dataclass
|
||||
class BRouteDeviceInfo:
|
||||
"""Static device information fetched once at setup."""
|
||||
|
||||
serial_number: str | None = None
|
||||
manufacturer_code: str | None = None
|
||||
echonet_version: str | None = None
|
||||
|
||||
|
||||
class BRouteUpdateCoordinator(DataUpdateCoordinator[BRouteData]):
|
||||
"""The B Route update coordinator."""
|
||||
|
||||
device_info_data: BRouteDeviceInfo
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
@@ -40,9 +52,9 @@ class BRouteUpdateCoordinator(DataUpdateCoordinator[BRouteData]):
|
||||
|
||||
self.device = entry.data[CONF_DEVICE]
|
||||
self.bid = entry.data[CONF_ID]
|
||||
password = entry.data[CONF_PASSWORD]
|
||||
self._password = entry.data[CONF_PASSWORD]
|
||||
|
||||
self.api = Momonga(dev=self.device, rbid=self.bid, pwd=password)
|
||||
self.api = Momonga(dev=self.device, rbid=self.bid, pwd=self._password)
|
||||
|
||||
super().__init__(
|
||||
hass,
|
||||
@@ -52,10 +64,34 @@ class BRouteUpdateCoordinator(DataUpdateCoordinator[BRouteData]):
|
||||
update_interval=DEFAULT_SCAN_INTERVAL,
|
||||
)
|
||||
|
||||
self.device_info_data = BRouteDeviceInfo()
|
||||
|
||||
async def _async_setup(self) -> None:
|
||||
await self.hass.async_add_executor_job(
|
||||
self.api.open,
|
||||
)
|
||||
def fetch() -> None:
|
||||
self.api.open()
|
||||
self._fetch_device_info()
|
||||
|
||||
await self.hass.async_add_executor_job(fetch)
|
||||
|
||||
def _fetch_device_info(self) -> None:
|
||||
"""Fetch static device information from the smart meter."""
|
||||
try:
|
||||
self.device_info_data.serial_number = self.api.get_serial_number()
|
||||
except MomongaError:
|
||||
_LOGGER.debug("Failed to fetch serial number", exc_info=True)
|
||||
|
||||
time.sleep(self.api.internal_xmit_interval)
|
||||
try:
|
||||
raw = self.api.get_manufacturer_code()
|
||||
self.device_info_data.manufacturer_code = raw.hex().upper()
|
||||
except MomongaError:
|
||||
_LOGGER.debug("Failed to fetch manufacturer code", exc_info=True)
|
||||
|
||||
time.sleep(self.api.internal_xmit_interval)
|
||||
try:
|
||||
self.device_info_data.echonet_version = self.api.get_standard_version()
|
||||
except MomongaError:
|
||||
_LOGGER.debug("Failed to fetch ECHONET Lite version", exc_info=True)
|
||||
|
||||
def _get_data(self) -> BRouteData:
|
||||
"""Get the data from API."""
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import Literal
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
SensorDeviceClass,
|
||||
@@ -69,6 +70,27 @@ SENSOR_DESCRIPTIONS = (
|
||||
),
|
||||
)
|
||||
|
||||
_DEVICE_INFO_MAPPING: dict[
|
||||
Literal["manufacturer", "serial_number", "sw_version"],
|
||||
Callable[[BRouteUpdateCoordinator], str | None],
|
||||
] = {
|
||||
"manufacturer": lambda coordinator: coordinator.device_info_data.manufacturer_code,
|
||||
"serial_number": lambda coordinator: coordinator.device_info_data.serial_number,
|
||||
"sw_version": lambda coordinator: coordinator.device_info_data.echonet_version,
|
||||
}
|
||||
|
||||
|
||||
def _build_device_info(coordinator: BRouteUpdateCoordinator) -> DeviceInfo:
|
||||
"""Build device information from coordinator data."""
|
||||
device = DeviceInfo(
|
||||
identifiers={(DOMAIN, coordinator.bid)},
|
||||
name=f"Route B Smart Meter {coordinator.bid}",
|
||||
)
|
||||
for key, fn in _DEVICE_INFO_MAPPING.items():
|
||||
if (value := fn(coordinator)) is not None:
|
||||
device[key] = value
|
||||
return device
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
@@ -98,10 +120,7 @@ class SmartMeterBRouteSensor(CoordinatorEntity[BRouteUpdateCoordinator], SensorE
|
||||
super().__init__(coordinator)
|
||||
self.entity_description: SensorEntityDescriptionWithValueAccessor = description
|
||||
self._attr_unique_id = f"{coordinator.bid}_{description.key}"
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, coordinator.bid)},
|
||||
name=f"Route B Smart Meter {coordinator.bid}",
|
||||
)
|
||||
self._attr_device_info = _build_device_info(coordinator)
|
||||
|
||||
@property
|
||||
def native_value(self) -> StateType:
|
||||
|
||||
@@ -44,6 +44,10 @@ def mock_momonga(exception=None) -> Generator[Mock]:
|
||||
}
|
||||
client.get_instantaneous_power.return_value = 3
|
||||
client.get_measured_cumulative_energy.return_value = 4
|
||||
client.get_serial_number.return_value = "TEST_SERIAL"
|
||||
client.get_manufacturer_code.return_value = b"\x00\x00\x16"
|
||||
client.get_standard_version.return_value = "F.0"
|
||||
client.internal_xmit_interval = 0
|
||||
yield mock_momonga
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user