Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 21:06:19 +00:00
Code Activity

Refactor the SMA integration to use a dedicated DataUpdateCoordinator (#154863)

Browse Source
This commit is contained in:
Erwin Douna
2025-10-21 18:26:22 +02:00
committed by GitHub
parent 18d5035877
commit b3e16bd4fa
6 changed files with 203 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

151
homeassistant/components/sma/__init__.py
View File

@@ -1,153 +1,69 @@
"""The sma integration."""
"""The SMA integration."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import TYPE_CHECKING
import pysma
from pysma import SMA
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_CONNECTIONS,
CONF_HOST,
CONF_MAC,
CONF_PASSWORD,
CONF_SCAN_INTERVAL,
CONF_SSL,
CONF_VERIFY_SSL,
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.core import Event, HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
CONF_GROUP,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
PLATFORMS,
PYSMA_COORDINATOR,
PYSMA_DEVICE_INFO,
PYSMA_OBJECT,
PYSMA_REMOVE_LISTENER,
PYSMA_SENSORS,
)
from .const import CONF_GROUP
from .coordinator import SMADataUpdateCoordinator
PLATFORMS = [Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
type SMAConfigEntry = ConfigEntry[SMADataUpdateCoordinator]
async def async_setup_entry(hass: HomeAssistant, entry: SMAConfigEntry) -> bool:
"""Set up sma from a config entry."""
# Init the SMA interface
protocol = "https" if entry.data[CONF_SSL] else "http"
url = f"{protocol}://{entry.data[CONF_HOST]}"
verify_ssl = entry.data[CONF_VERIFY_SSL]
group = entry.data[CONF_GROUP]
password = entry.data[CONF_PASSWORD]
session = async_get_clientsession(hass, verify_ssl=verify_ssl)
sma = pysma.SMA(session, url, password, group)
try:
# Get updated device info
sma_device_info = await sma.device_info()
# Get all device sensors
sensor_def = await sma.get_sensors()
except (
pysma.exceptions.SmaReadException,
pysma.exceptions.SmaConnectionException,
) as exc:
raise ConfigEntryNotReady from exc
except pysma.exceptions.SmaAuthenticationException as exc:
raise ConfigEntryAuthFailed from exc
if TYPE_CHECKING:
assert entry.unique_id
# Create DeviceInfo object from sma_device_info
device_info = DeviceInfo(
configuration_url=url,
identifiers={(DOMAIN, entry.unique_id)},
manufacturer=sma_device_info["manufacturer"],
model=sma_device_info["type"],
name=sma_device_info["name"],
sw_version=sma_device_info["sw_version"],
serial_number=sma_device_info["serial"],
sma = SMA(
session=async_get_clientsession(
hass=hass, verify_ssl=entry.data[CONF_VERIFY_SSL]
),
url=url,
password=entry.data[CONF_PASSWORD],
group=entry.data[CONF_GROUP],
)
# Add the MAC address to connections, if it comes via DHCP
if CONF_MAC in entry.data:
device_info[ATTR_CONNECTIONS] = {
(dr.CONNECTION_NETWORK_MAC, entry.data[CONF_MAC])
}
# Define the coordinator
async def async_update_data():
"""Update the used SMA sensors."""
try:
await sma.read(sensor_def)
except (
pysma.exceptions.SmaReadException,
pysma.exceptions.SmaConnectionException,
) as exc:
raise UpdateFailed(exc) from exc
interval = timedelta(
seconds=entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
)
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
config_entry=entry,
name="sma",
update_method=async_update_data,
update_interval=interval,
)
try:
await coordinator.async_config_entry_first_refresh()
except ConfigEntryNotReady:
await sma.close_session()
raise
# Ensure we logout on shutdown
async def async_close_session(event):
"""Close the session."""
await sma.close_session()
remove_stop_listener = hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, async_close_session
)
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = {
PYSMA_OBJECT: sma,
PYSMA_COORDINATOR: coordinator,
PYSMA_SENSORS: sensor_def,
PYSMA_REMOVE_LISTENER: remove_stop_listener,
PYSMA_DEVICE_INFO: device_info,
}
coordinator = SMADataUpdateCoordinator(hass, entry, sma)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# Ensure the SMA session closes when Home Assistant stops
async def _async_handle_shutdown(event: Event) -> None:
await coordinator.async_close_sma_session()
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_handle_shutdown)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: SMAConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
data = hass.data[DOMAIN].pop(entry.entry_id)
await data[PYSMA_OBJECT].close_session()
data[PYSMA_REMOVE_LISTENER]()
return unload_ok
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@@ -156,7 +72,6 @@ async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
_LOGGER.debug("Migrating from version %s", entry.version)
if entry.version == 1:
# 1 -> 2: Unique ID from integer to string
if entry.minor_version == 1:
minor_version = 2
hass.config_entries.async_update_entry(

3
homeassistant/components/sma/const.py
View File

@@ -1,7 +1,5 @@
"""Constants for the sma integration."""
from homeassistant.const import Platform
DOMAIN = "sma"
PYSMA_COORDINATOR = "coordinator"
@@ -10,7 +8,6 @@ PYSMA_REMOVE_LISTENER = "remove_listener"
PYSMA_SENSORS = "pysma_sensors"
PYSMA_DEVICE_INFO = "device_info"
PLATFORMS = [Platform.SENSOR]
CONF_GROUP = "group"

113
homeassistant/components/sma/coordinator.py Normal file
View File

@@ -0,0 +1,113 @@
"""Coordinator for the SMA integration."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from pysma import SMA
from pysma.exceptions import (
SmaAuthenticationException,
SmaConnectionException,
SmaReadException,
)
from pysma.sensor import Sensor
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_SCAN_INTERVAL
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN
_LOGGER = logging.getLogger(__name__)
@dataclass(slots=True)
class SMACoordinatorData:
"""Data class for SMA sensors."""
sma_device_info: dict[str, str]
sensors: list[Sensor]
class SMADataUpdateCoordinator(DataUpdateCoordinator[SMACoordinatorData]):
"""Data Update Coordinator for SMA."""
config_entry: ConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: ConfigEntry,
sma: SMA,
) -> None:
"""Initialize the SMA Data Update Coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=DOMAIN,
update_interval=timedelta(
seconds=config_entry.options.get(
CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL
)
),
)
self.sma = sma
self._sma_device_info: dict[str, str] = {}
self._sensors: list[Sensor] = []
async def _async_setup(self) -> None:
"""Setup the SMA Data Update Coordinator."""
try:
self._sma_device_info = await self.sma.device_info()
self._sensors = await self.sma.get_sensors()
except (
SmaReadException,
SmaConnectionException,
) as err:
await self.async_close_sma_session()
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": repr(err)},
) from err
except SmaAuthenticationException as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="invalid_auth",
translation_placeholders={"error": repr(err)},
) from err
async def _async_update_data(self) -> SMACoordinatorData:
"""Update the used SMA sensors."""
try:
await self.sma.read(self._sensors)
except (
SmaReadException,
SmaConnectionException,
) as err:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": repr(err)},
) from err
except SmaAuthenticationException as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="invalid_auth",
translation_placeholders={"error": repr(err)},
) from err
return SMACoordinatorData(
sma_device_info=self._sma_device_info,
sensors=self._sensors,
)
async def async_close_sma_session(self) -> None:
"""Close the SMA session."""
await self.sma.close_session()
_LOGGER.debug("SMA session closed")

57
homeassistant/components/sma/sensor.py
View File

@@ -2,8 +2,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
import pysma
from homeassistant.components.sensor import (
@@ -12,8 +10,9 @@ from homeassistant.components.sensor import (
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_SSL,
PERCENTAGE,
EntityCategory,
UnitOfApparentPower,
@@ -29,12 +28,11 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, PYSMA_COORDINATOR, PYSMA_DEVICE_INFO, PYSMA_SENSORS
from . import SMAConfigEntry
from .const import DOMAIN
from .coordinator import SMADataUpdateCoordinator
SENSOR_ENTITIES: dict[str, SensorEntityDescription] = {
"status": SensorEntityDescription(
@@ -837,41 +835,32 @@ SENSOR_ENTITIES: dict[str, SensorEntityDescription] = {
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
entry: SMAConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up SMA sensors."""
sma_data = hass.data[DOMAIN][config_entry.entry_id]
coordinator = sma_data[PYSMA_COORDINATOR]
used_sensors = sma_data[PYSMA_SENSORS]
device_info = sma_data[PYSMA_DEVICE_INFO]
if TYPE_CHECKING:
assert config_entry.unique_id
"""Setup SMA sensors."""
coordinator = entry.runtime_data
async_add_entities(
SMAsensor(
coordinator,
config_entry.unique_id,
SENSOR_ENTITIES.get(sensor.name),
device_info,
sensor,
entry,
)
for sensor in used_sensors
for sensor in coordinator.data.sensors
)
class SMAsensor(CoordinatorEntity, SensorEntity):
class SMAsensor(CoordinatorEntity[SMADataUpdateCoordinator], SensorEntity):
"""Representation of a SMA sensor."""
def __init__(
self,
coordinator: DataUpdateCoordinator,
config_entry_unique_id: str,
coordinator: SMADataUpdateCoordinator,
description: SensorEntityDescription | None,
device_info: DeviceInfo,
pysma_sensor: pysma.sensor.Sensor,
entry: SMAConfigEntry,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
@@ -880,11 +869,23 @@ class SMAsensor(CoordinatorEntity, SensorEntity):
else:
self._attr_name = pysma_sensor.name
self._sensor = pysma_sensor
protocol = "https" if entry.data[CONF_SSL] else "http"
url = f"{protocol}://{entry.data[CONF_HOST]}"
self._attr_device_info = device_info
self._sensor = pysma_sensor
assert entry.unique_id
self._attr_device_info = DeviceInfo(
configuration_url=url,
identifiers={(DOMAIN, entry.unique_id)},
manufacturer=coordinator.data.sma_device_info["manufacturer"],
model=coordinator.data.sma_device_info["type"],
name=coordinator.data.sma_device_info["name"],
sw_version=coordinator.data.sma_device_info["sw_version"],
serial_number=coordinator.data.sma_device_info["serial"],
)
self._attr_unique_id = (
f"{config_entry_unique_id}-{pysma_sensor.key}_{pysma_sensor.key_idx}"
f"{entry.unique_id}-{pysma_sensor.key}_{pysma_sensor.key_idx}"
)
# Set sensor enabled to False.

31
tests/components/sma/conftest.py
View File

@@ -46,13 +46,19 @@ def mock_setup_entry() -> Generator[AsyncMock]:
@pytest.fixture
def mock_sma_client() -> Generator[MagicMock]:
"""Mock the SMA client."""
with patch("homeassistant.components.sma.pysma.SMA", autospec=True) as client:
client.return_value.device_info.return_value = MOCK_DEVICE
client.new_session.return_value = True
client.return_value.get_sensors.return_value = Sensors(
sensor_map[GENERIC_SENSORS]
+ sensor_map[OPTIMIZERS_VIA_INVERTER]
+ sensor_map[ENERGY_METER_VIA_INVERTER]
with patch(
"homeassistant.components.sma.coordinator.SMA", autospec=True
) as sma_cls:
sma_instance: MagicMock = sma_cls.return_value
sma_instance.device_info = AsyncMock(return_value=MOCK_DEVICE)
sma_instance.new_session = AsyncMock(return_value=True)
sma_instance.close_session = AsyncMock(return_value=True)
sma_instance.get_sensors = AsyncMock(
return_value=Sensors(
sensor_map[GENERIC_SENSORS]
+ sensor_map[OPTIMIZERS_VIA_INVERTER]
+ sensor_map[ENERGY_METER_VIA_INVERTER]
)
)
default_sensor_values = {
@@ -65,12 +71,17 @@ def mock_sma_client() -> Generator[MagicMock]:
"6100_00499700": 1000,
}
def mock_read(sensors):
async def _async_mock_read(sensors) -> bool:
for sensor in sensors:
if sensor.key in default_sensor_values:
sensor.value = default_sensor_values[sensor.key]
return True
client.return_value.read.side_effect = mock_read
sma_instance.read = AsyncMock(side_effect=_async_mock_read)
yield client
with (
patch("homeassistant.components.sma.config_flow.pysma.SMA", new=sma_cls),
patch("homeassistant.components.sma.SMA", new=sma_cls),
patch("pysma.SMA", new=sma_cls),
):
yield sma_instance

13
tests/components/sma/test_config_flow.py
View File

@@ -1,6 +1,6 @@
"""Test the sma config flow."""
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
from pysma.exceptions import (
SmaAuthenticationException,
@@ -47,7 +47,7 @@ DHCP_DISCOVERY_DUPLICATE_001 = DhcpServiceInfo(
async def test_form(
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_sma_client: AsyncMock
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_sma_client: MagicMock
) -> None:
"""Test we get the form."""
@@ -91,7 +91,8 @@ async def test_form_exceptions(
)
with patch(
"homeassistant.components.sma.pysma.SMA.new_session", side_effect=exception
"homeassistant.components.sma.config_flow.pysma.SMA.new_session",
side_effect=exception,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@@ -210,7 +211,7 @@ async def test_dhcp_exceptions(
data=DHCP_DISCOVERY,
)
with patch("homeassistant.components.sma.pysma.SMA") as mock_sma:
with patch("homeassistant.components.sma.config_flow.pysma.SMA") as mock_sma:
mock_sma_instance = mock_sma.return_value
mock_sma_instance.new_session = AsyncMock(side_effect=exception)
@@ -222,7 +223,7 @@ async def test_dhcp_exceptions(
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
with patch("homeassistant.components.sma.pysma.SMA") as mock_sma:
with patch("homeassistant.components.sma.config_flow.pysma.SMA") as mock_sma:
mock_sma_instance = mock_sma.return_value
mock_sma_instance.new_session = AsyncMock(return_value=True)
mock_sma_instance.device_info = AsyncMock(return_value=MOCK_DEVICE)
@@ -290,7 +291,7 @@ async def test_reauth_flow_exceptions(
result = await entry.start_reauth_flow(hass)
with patch("homeassistant.components.sma.pysma.SMA") as mock_sma:
with patch("homeassistant.components.sma.config_flow.pysma.SMA") as mock_sma:
mock_sma_instance = mock_sma.return_value
mock_sma_instance.new_session = AsyncMock(side_effect=exception)
result = await hass.config_entries.flow.async_configure(