mirror of https://github.com/home-assistant/core
157 lines
4.4 KiB
Python
157 lines
4.4 KiB
Python
"""The sma integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
import pysma
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.const import (
|
|
CONF_HOST,
|
|
CONF_PASSWORD,
|
|
CONF_SCAN_INTERVAL,
|
|
CONF_SSL,
|
|
CONF_VERIFY_SSL,
|
|
EVENT_HOMEASSISTANT_STOP,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ConfigEntryNotReady
|
|
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,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> 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
|
|
|
|
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"],
|
|
)
|
|
|
|
# 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,
|
|
}
|
|
|
|
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
|
|
|
return True
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> 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
|
|
|
|
|
|
async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|
"""Migrate entry."""
|
|
|
|
_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(
|
|
entry, unique_id=str(entry.unique_id), minor_version=minor_version
|
|
)
|
|
|
|
_LOGGER.debug("Migration successful")
|
|
|
|
return True
|