Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Rewrite Fronius integration (#59686) 

* Add unique_id and use DataUpdateCoordinator in Fronius (#57879)

* initial refactoring commit - meters

- config_flow (no strings, no tests yet)
- import yaml config
- FroniusSolarNet class for holding Fronius object , coordinators and some common data
- meter descriptions
- update coordinator
- entities (including devices)

* storage controllers

* error handling on init; inverter unique_id

* inverters

* power_flow

* fix VA, var, varh not valid for device_class power/energy

and add custom icons

* add SolarNet device for system wide values

* cleanup

* config_flow strings

* test config_flow

* use pyfronius 0.7.0

* enable strict typing

* remove TODO comments

* fix lint errors; move FroniusSensorEntity to sensor.py

* power_flow as optional coordinator

API V0 doesn't support power_flow endpoint

* show error message in logs

* prevent parallel requests to one host

* logger_info coordinator

* store FroniusSolarNet reference directly in coordinator

* cleanup coordinators when unloading entry

* round floats returned by Fronius API

* default icons for grid im/export tariffs

* small typing fix

* Update homeassistant/components/fronius/sensor.py

Co-authored-by: Brett Adams <Bre77@users.noreply.github.com>

* DC icons

* prepend names with "Fronius" and device type

to get more reasonable default entity_ids (eg. have them next to each other when alphabetically sorted)

* remove config_flow and devices

* rename _FroniusUpdateCoordinator to FroniusCoordinatorBase

and mark ABC

* move SensorEntityDescriptions to sensor.py

* Revert "move SensorEntityDescriptions to sensor.py"

This reverts commit 2e5a726eb6.

* Don't raise ConfigEntryNotReady and use regular refresh method

* move bridge initialization out of helper class

* no coverage tests

* power_flow update interval 10 seconds

* move SensorEntityDescriptions to sensor.py

without introducing a circular dependency

* deprecation warning for CONF_MONITORED_CONDITIONS

* remove extra_state_attributes form meter sensor entities

* readd diagnostic entities

* decouple default entity_id from default name

* use key instead of name for entity_id

and make deprecated config key optional

* adjust tests

* use old entity_ids

these changes are now backwards compatible

* check coverage

* simplify entity description definitions

* restore entity names of previous implementation

Co-authored-by: Brett Adams <Bre77@users.noreply.github.com>

* Add config_flow for Fronius integration (#59677)

* Cleanup Fronius config_flow and tests (#60094)

* Add devices to Fronius integration (#60104)

* New entity names for Fronius entities (#60215)

* Adaptive update interval for Fronius coordinators (#60192)

Co-authored-by: Brett Adams <Bre77@users.noreply.github.com>
pull/60258/head
Matthias Alphart 2021-11-24 02:04:36 +01:00 committed by GitHub
parent 314f593066
commit 3dac661480
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 1721 additions and 407 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -372,7 +372,6 @@ omit =
homeassistant/components/fritzbox_callmonitor/const.py
homeassistant/components/fritzbox_callmonitor/base.py
homeassistant/components/fritzbox_callmonitor/sensor.py
homeassistant/components/fronius/sensor.py
homeassistant/components/frontier_silicon/media_player.py
homeassistant/components/futurenow/light.py
homeassistant/components/garadget/cover.py

1
.strict-typing
View File

@ -49,6 +49,7 @@ homeassistant.components.flunearyou.*
homeassistant.components.flux_led.*
homeassistant.components.forecast_solar.*
homeassistant.components.fritzbox.*
homeassistant.components.fronius.*
homeassistant.components.frontend.*
homeassistant.components.fritz.*
homeassistant.components.geo_location.*

2
CODEOWNERS
View File

@ -186,7 +186,7 @@ homeassistant/components/freebox/* @hacf-fr @Quentame
homeassistant/components/freedompro/* @stefano055415
homeassistant/components/fritz/* @mammuth @AaronDavidSchneider @chemelli74
homeassistant/components/fritzbox/* @mib1185 @flabbamann
homeassistant/components/fronius/* @nielstron
homeassistant/components/fronius/* @nielstron @farmio
homeassistant/components/frontend/* @home-assistant/frontend
homeassistant/components/garages_amsterdam/* @klaasnicolaas
homeassistant/components/gdacs/* @exxamalte

205
homeassistant/components/fronius/__init__.py
View File

@ -1 +1,204 @@
"""The Fronius component."""
"""The Fronius integration."""
from __future__ import annotations
import asyncio
import logging
from typing import Callable, TypeVar
from pyfronius import Fronius, FroniusError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_MODEL, ATTR_SW_VERSION, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity import DeviceInfo
from .const import DOMAIN, SOLAR_NET_ID_SYSTEM, FroniusDeviceInfo
from .coordinator import (
FroniusCoordinatorBase,
FroniusInverterUpdateCoordinator,
FroniusLoggerUpdateCoordinator,
FroniusMeterUpdateCoordinator,
FroniusPowerFlowUpdateCoordinator,
FroniusStorageUpdateCoordinator,
)
_LOGGER = logging.getLogger(__name__)
PLATFORMS: list[str] = ["sensor"]
FroniusCoordinatorType = TypeVar("FroniusCoordinatorType", bound=FroniusCoordinatorBase)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up fronius from a config entry."""
host = entry.data[CONF_HOST]
fronius = Fronius(async_get_clientsession(hass), host)
solar_net = FroniusSolarNet(hass, entry, fronius)
await solar_net.init_devices()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = solar_net
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
# reload on config_entry update
entry.async_on_unload(entry.add_update_listener(async_update_entry))
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:
solar_net = hass.data[DOMAIN].pop(entry.entry_id)
while solar_net.cleanup_callbacks:
solar_net.cleanup_callbacks.pop()()
return unload_ok
async def async_update_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Update a given config entry."""
await hass.config_entries.async_reload(entry.entry_id)
class FroniusSolarNet:
"""The FroniusSolarNet class routes received values to sensor entities."""
def __init__(
self, hass: HomeAssistant, entry: ConfigEntry, fronius: Fronius
) -> None:
"""Initialize FroniusSolarNet class."""
self.hass = hass
self.cleanup_callbacks: list[Callable[[], None]] = []
self.config_entry = entry
self.coordinator_lock = asyncio.Lock()
self.fronius = fronius
self.host: str = entry.data[CONF_HOST]
# entry.unique_id is either logger uid or first inverter uid if no logger available
# prepended by "solar_net_" to have individual device for whole system (power_flow)
self.solar_net_device_id = f"solar_net_{entry.unique_id}"
self.system_device_info: DeviceInfo | None = None
self.inverter_coordinators: list[FroniusInverterUpdateCoordinator] = []
self.logger_coordinator: FroniusLoggerUpdateCoordinator | None = None
self.meter_coordinator: FroniusMeterUpdateCoordinator | None = None
self.power_flow_coordinator: FroniusPowerFlowUpdateCoordinator | None = None
self.storage_coordinator: FroniusStorageUpdateCoordinator | None = None
async def init_devices(self) -> None:
"""Initialize DataUpdateCoordinators for SolarNet devices."""
if self.config_entry.data["is_logger"]:
self.logger_coordinator = FroniusLoggerUpdateCoordinator(
hass=self.hass,
solar_net=self,
logger=_LOGGER,
name=f"{DOMAIN}_logger_{self.host}",
)
await self.logger_coordinator.async_config_entry_first_refresh()
# _create_solar_net_device uses data from self.logger_coordinator when available
self.system_device_info = await self._create_solar_net_device()
_inverter_infos = await self._get_inverter_infos()
for inverter_info in _inverter_infos:
coordinator = FroniusInverterUpdateCoordinator(
hass=self.hass,
solar_net=self,
logger=_LOGGER,
name=f"{DOMAIN}_inverter_{inverter_info.solar_net_id}_{self.host}",
inverter_info=inverter_info,
)
await coordinator.async_config_entry_first_refresh()
self.inverter_coordinators.append(coordinator)
self.meter_coordinator = await self._init_optional_coordinator(
FroniusMeterUpdateCoordinator(
hass=self.hass,
solar_net=self,
logger=_LOGGER,
name=f"{DOMAIN}_meters_{self.host}",
)
)
self.power_flow_coordinator = await self._init_optional_coordinator(
FroniusPowerFlowUpdateCoordinator(
hass=self.hass,
solar_net=self,
logger=_LOGGER,
name=f"{DOMAIN}_power_flow_{self.host}",
)
)
self.storage_coordinator = await self._init_optional_coordinator(
FroniusStorageUpdateCoordinator(
hass=self.hass,
solar_net=self,
logger=_LOGGER,
name=f"{DOMAIN}_storages_{self.host}",
)
)
async def _create_solar_net_device(self) -> DeviceInfo:
"""Create a device for the Fronius SolarNet system."""
solar_net_device: DeviceInfo = DeviceInfo(
configuration_url=self.host,
identifiers={(DOMAIN, self.solar_net_device_id)},
manufacturer="Fronius",
name="SolarNet",
)
if self.logger_coordinator:
_logger_info = self.logger_coordinator.data[SOLAR_NET_ID_SYSTEM]
solar_net_device[ATTR_MODEL] = _logger_info["product_type"]["value"]
solar_net_device[ATTR_SW_VERSION] = _logger_info["software_version"][
"value"
]
device_registry = await dr.async_get_registry(self.hass)
device_registry.async_get_or_create(
config_entry_id=self.config_entry.entry_id,
**solar_net_device,
)
return solar_net_device
async def _get_inverter_infos(self) -> list[FroniusDeviceInfo]:
"""Get information about the inverters in the SolarNet system."""
try:
_inverter_info = await self.fronius.inverter_info()
except FroniusError as err:
raise ConfigEntryNotReady from err
inverter_infos: list[FroniusDeviceInfo] = []
for inverter in _inverter_info["inverters"]:
solar_net_id = inverter["device_id"]["value"]
unique_id = inverter["unique_id"]["value"]
device_info = DeviceInfo(
identifiers={(DOMAIN, unique_id)},
manufacturer=inverter["device_type"].get("manufacturer", "Fronius"),
model=inverter["device_type"].get(
"model", inverter["device_type"]["value"]
),
name=inverter.get("custom_name", {}).get("value"),
via_device=(DOMAIN, self.solar_net_device_id),
)
inverter_infos.append(
FroniusDeviceInfo(
device_info=device_info,
solar_net_id=solar_net_id,
unique_id=unique_id,
)
)
return inverter_infos
@staticmethod
async def _init_optional_coordinator(
coordinator: FroniusCoordinatorType,
) -> FroniusCoordinatorType | None:
"""Initialize an update coordinator and return it if devices are found."""
try:
await coordinator.async_config_entry_first_refresh()
except ConfigEntryNotReady:
return None
# keep coordinator only if devices are found
# else ConfigEntryNotReady raised form KeyError
# in FroniusMeterUpdateCoordinator._get_fronius_device_data
return coordinator

109
homeassistant/components/fronius/config_flow.py Normal file
View File

@ -0,0 +1,109 @@
"""Config flow for Fronius integration."""
from __future__ import annotations
import logging
from typing import Any
from pyfronius import Fronius, FroniusError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_RESOURCE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN, FroniusConfigEntryData
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
}
)
async def validate_input(
hass: HomeAssistant, data: dict[str, Any]
) -> tuple[str, FroniusConfigEntryData]:
"""Validate the user input allows us to connect.
Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
"""
host = data[CONF_HOST]
fronius = Fronius(async_get_clientsession(hass), host)
try:
datalogger_info: dict[str, Any]
datalogger_info = await fronius.current_logger_info()
except FroniusError as err:
_LOGGER.debug(err)
else:
logger_uid: str = datalogger_info["unique_identifier"]["value"]
return logger_uid, FroniusConfigEntryData(
host=host,
is_logger=True,
)
# Gen24 devices don't provide GetLoggerInfo
try:
inverter_info = await fronius.inverter_info()
first_inverter = next(inverter for inverter in inverter_info["inverters"])
except FroniusError as err:
_LOGGER.debug(err)
raise CannotConnect from err
except StopIteration as err:
raise CannotConnect("No supported Fronius SolarNet device found.") from err
first_inverter_uid: str = first_inverter["unique_id"]["value"]
return first_inverter_uid, FroniusConfigEntryData(
host=host,
is_logger=False,
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Fronius."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
errors = {}
try:
unique_id, info = await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(unique_id, raise_on_progress=False)
self._abort_if_unique_id_configured(
updates=dict(info), reload_on_update=False
)
title = (
f"SolarNet {'Datalogger' if info['is_logger'] else 'Inverter'}"
f" at {info['host']}"
)
return self.async_create_entry(title=title, data=info)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
async def async_step_import(self, conf: dict) -> FlowResult:
"""Import a configuration from config.yaml."""
return await self.async_step_user(user_input={CONF_HOST: conf[CONF_RESOURCE]})
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""

25
homeassistant/components/fronius/const.py Normal file
View File

@ -0,0 +1,25 @@
"""Constants for the Fronius integration."""
from typing import Final, NamedTuple, TypedDict
from homeassistant.helpers.entity import DeviceInfo
DOMAIN: Final = "fronius"
SolarNetId = str
SOLAR_NET_ID_POWER_FLOW: SolarNetId = "power_flow"
SOLAR_NET_ID_SYSTEM: SolarNetId = "system"
class FroniusConfigEntryData(TypedDict):
"""ConfigEntry for the Fronius integration."""
host: str
is_logger: bool
class FroniusDeviceInfo(NamedTuple):
"""Information about a Fronius inverter device."""
device_info: DeviceInfo
solar_net_id: SolarNetId
unique_id: str

184
homeassistant/components/fronius/coordinator.py Normal file
View File

@ -0,0 +1,184 @@
"""DataUpdateCoordinators for the Fronius integration."""
from __future__ import annotations
from abc import ABC, abstractmethod
from datetime import timedelta
from typing import TYPE_CHECKING, Any, Dict, TypeVar
from pyfronius import FroniusError
from homeassistant.components.sensor import SensorEntityDescription
from homeassistant.core import callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
SOLAR_NET_ID_POWER_FLOW,
SOLAR_NET_ID_SYSTEM,
FroniusDeviceInfo,
SolarNetId,
)
from .sensor import (
INVERTER_ENTITY_DESCRIPTIONS,
LOGGER_ENTITY_DESCRIPTIONS,
METER_ENTITY_DESCRIPTIONS,
POWER_FLOW_ENTITY_DESCRIPTIONS,
STORAGE_ENTITY_DESCRIPTIONS,
)
if TYPE_CHECKING:
from . import FroniusSolarNet
from .sensor import _FroniusSensorEntity
FroniusEntityType = TypeVar("FroniusEntityType", bound=_FroniusSensorEntity)
class FroniusCoordinatorBase(
ABC, DataUpdateCoordinator[Dict[SolarNetId, Dict[str, Any]]]
):
"""Query Fronius endpoint and keep track of seen conditions."""
default_interval: timedelta
error_interval: timedelta
valid_descriptions: list[SensorEntityDescription]
def __init__(self, *args: Any, solar_net: FroniusSolarNet, **kwargs: Any) -> None:
"""Set up the FroniusCoordinatorBase class."""
self._failed_update_count = 0
self.solar_net = solar_net
# unregistered_keys are used to create entities in platform module
self.unregistered_keys: dict[SolarNetId, set[str]] = {}
super().__init__(*args, update_interval=self.default_interval, **kwargs)
@abstractmethod
async def _update_method(self) -> dict[SolarNetId, Any]:
"""Return data per solar net id from pyfronius."""
async def _async_update_data(self) -> dict[SolarNetId, Any]:
"""Fetch the latest data from the source."""
async with self.solar_net.coordinator_lock:
try:
data = await self._update_method()
except FroniusError as err:
self._failed_update_count += 1
if self._failed_update_count == 3:
self.update_interval = self.error_interval
raise UpdateFailed(err) from err
if self._failed_update_count != 0:
self._failed_update_count = 0
self.update_interval = self.default_interval
for solar_net_id in data:
if solar_net_id not in self.unregistered_keys:
# id seen for the first time
self.unregistered_keys[solar_net_id] = {
desc.key for desc in self.valid_descriptions
}
return data
@callback
def add_entities_for_seen_keys(
self,
async_add_entities: AddEntitiesCallback,
entity_constructor: type[FroniusEntityType],
) -> None:
"""
Add entities for received keys and registers listener for future seen keys.
Called from a platforms `async_setup_entry`.
"""
@callback
def _add_entities_for_unregistered_keys() -> None:
"""Add entities for keys seen for the first time."""
new_entities: list = []
for solar_net_id, device_data in self.data.items():
for key in self.unregistered_keys[solar_net_id].intersection(
device_data
):
new_entities.append(entity_constructor(self, key, solar_net_id))
self.unregistered_keys[solar_net_id].remove(key)
if new_entities:
async_add_entities(new_entities)
_add_entities_for_unregistered_keys()
self.solar_net.cleanup_callbacks.append(
self.async_add_listener(_add_entities_for_unregistered_keys)
)
class FroniusInverterUpdateCoordinator(FroniusCoordinatorBase):
"""Query Fronius device inverter endpoint and keep track of seen conditions."""
default_interval = timedelta(minutes=1)
error_interval = timedelta(minutes=10)
valid_descriptions = INVERTER_ENTITY_DESCRIPTIONS
def __init__(
self, *args: Any, inverter_info: FroniusDeviceInfo, **kwargs: Any
) -> None:
"""Set up a Fronius inverter device scope coordinator."""
super().__init__(*args, **kwargs)
self.inverter_info = inverter_info
async def _update_method(self) -> dict[SolarNetId, Any]:
"""Return data per solar net id from pyfronius."""
data = await self.solar_net.fronius.current_inverter_data(
self.inverter_info.solar_net_id
)
# wrap a single devices data in a dict with solar_net_id key for
# FroniusCoordinatorBase _async_update_data and add_entities_for_seen_keys
return {self.inverter_info.solar_net_id: data}
class FroniusLoggerUpdateCoordinator(FroniusCoordinatorBase):
"""Query Fronius logger info endpoint and keep track of seen conditions."""
default_interval = timedelta(hours=1)
error_interval = timedelta(hours=1)
valid_descriptions = LOGGER_ENTITY_DESCRIPTIONS
async def _update_method(self) -> dict[SolarNetId, Any]:
"""Return data per solar net id from pyfronius."""
data = await self.solar_net.fronius.current_logger_info()
return {SOLAR_NET_ID_SYSTEM: data}
class FroniusMeterUpdateCoordinator(FroniusCoordinatorBase):
"""Query Fronius system meter endpoint and keep track of seen conditions."""
default_interval = timedelta(minutes=1)
error_interval = timedelta(minutes=10)
valid_descriptions = METER_ENTITY_DESCRIPTIONS
async def _update_method(self) -> dict[SolarNetId, Any]:
"""Return data per solar net id from pyfronius."""
data = await self.solar_net.fronius.current_system_meter_data()
return data["meters"] # type: ignore[no-any-return]
class FroniusPowerFlowUpdateCoordinator(FroniusCoordinatorBase):
"""Query Fronius power flow endpoint and keep track of seen conditions."""
default_interval = timedelta(seconds=10)
error_interval = timedelta(minutes=3)
valid_descriptions = POWER_FLOW_ENTITY_DESCRIPTIONS
async def _update_method(self) -> dict[SolarNetId, Any]:
"""Return data per solar net id from pyfronius."""
data = await self.solar_net.fronius.current_power_flow()
return {SOLAR_NET_ID_POWER_FLOW: data}
class FroniusStorageUpdateCoordinator(FroniusCoordinatorBase):
"""Query Fronius system storage endpoint and keep track of seen conditions."""
default_interval = timedelta(minutes=1)
error_interval = timedelta(minutes=10)
valid_descriptions = STORAGE_ENTITY_DESCRIPTIONS
async def _update_method(self) -> dict[SolarNetId, Any]:
"""Return data per solar net id from pyfronius."""
data = await self.solar_net.fronius.current_system_storage_data()
return data["storages"] # type: ignore[no-any-return]

3
homeassistant/components/fronius/manifest.json
View File

@ -1,8 +1,9 @@
{
"domain": "fronius",
"name": "Fronius",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/fronius",
"requirements": ["pyfronius==0.7.0"],
"codeowners": ["@nielstron"],
"codeowners": ["@nielstron", "@farmio"],
"iot_class": "local_polling"
}

996
homeassistant/components/fronius/sensor.py
View File

File diff suppressed because it is too large Load Diff

20
homeassistant/components/fronius/strings.json Normal file
View File

@ -0,0 +1,20 @@
{
"config": {
"step": {
"user": {
"title": "Fronius SolarNet",
"description": "Configure the IP address or local hostname of your Fronius device.",
"data": {
"host": "[%key:common::config_flow::data::host%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

20
homeassistant/components/fronius/translations/en.json Normal file
View File

@ -0,0 +1,20 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"host": "Host"
},
"description": "Configure the IP address or local hostname of your Fronius device.",
"title": "Fronius SolarNet"
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -102,6 +102,7 @@ FLOWS = [
"fritz",
"fritzbox",
"fritzbox_callmonitor",
"fronius",
"garages_amsterdam",
"gdacs",
"geofency",

11
mypy.ini
View File

@ -550,6 +550,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.fronius.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.frontend.*]
check_untyped_defs = true
disallow_incomplete_defs = true

77
tests/components/fronius/__init__.py
View File

@ -1,23 +1,72 @@
"""Tests for the Fronius integration."""
from homeassistant.components.fronius.const import DOMAIN
from homeassistant.const import CONF_HOST
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import CONF_MONITORED_CONDITIONS, CONF_RESOURCE
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, load_fixture
from tests.test_util.aiohttp import AiohttpClientMocker
from .const import DOMAIN, MOCK_HOST
MOCK_HOST = "http://fronius"
MOCK_UID = "123.4567890" # has to match mocked logger unique_id
async def setup_fronius_integration(hass, devices):
async def setup_fronius_integration(hass):
"""Create the Fronius integration."""
assert await async_setup_component(
hass,
SENSOR_DOMAIN,
{
SENSOR_DOMAIN: {
"platform": DOMAIN,
CONF_RESOURCE: MOCK_HOST,
CONF_MONITORED_CONDITIONS: devices,
}
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=MOCK_UID,
data={
CONF_HOST: MOCK_HOST,
"is_logger": True,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
def mock_responses(
aioclient_mock: AiohttpClientMocker,
host: str = MOCK_HOST,
night: bool = False,
) -> None:
"""Mock responses for Fronius Symo inverter with meter."""
aioclient_mock.clear_requests()
_day_or_night = "night" if night else "day"
aioclient_mock.get(
f"{host}/solar_api/GetAPIVersion.cgi",
text=load_fixture("symo/GetAPIVersion.json", "fronius"),
)
aioclient_mock.get(
f"{host}/solar_api/v1/GetInverterRealtimeData.cgi?Scope=Device&"
"DeviceId=1&DataCollection=CommonInverterData",
text=load_fixture(
f"symo/GetInverterRealtimeDate_Device_1_{_day_or_night}.json", "fronius"
),
)
aioclient_mock.get(
f"{host}/solar_api/v1/GetInverterInfo.cgi",
text=load_fixture("symo/GetInverterInfo.json", "fronius"),
)
aioclient_mock.get(
f"{host}/solar_api/v1/GetLoggerInfo.cgi",
text=load_fixture("symo/GetLoggerInfo.json", "fronius"),
)
aioclient_mock.get(
f"{host}/solar_api/v1/GetMeterRealtimeData.cgi?Scope=Device&DeviceId=0",
text=load_fixture("symo/GetMeterRealtimeData_Device_0.json", "fronius"),
)
aioclient_mock.get(
f"{host}/solar_api/v1/GetMeterRealtimeData.cgi?Scope=System",
text=load_fixture("symo/GetMeterRealtimeData_System.json", "fronius"),
)
aioclient_mock.get(
f"{host}/solar_api/v1/GetPowerFlowRealtimeData.fcgi",
text=load_fixture(
f"symo/GetPowerFlowRealtimeData_{_day_or_night}.json", "fronius"
),
)
aioclient_mock.get(
f"{host}/solar_api/v1/GetStorageRealtimeData.cgi?Scope=System",
text=load_fixture("symo/GetStorageRealtimeData_System.json", "fronius"),
)

4
tests/components/fronius/const.py
View File

@ -1,4 +0,0 @@
"""Constants for Fronius tests."""
DOMAIN = "fronius"
MOCK_HOST = "http://fronius"

263
tests/components/fronius/test_config_flow.py Normal file
View File

@ -0,0 +1,263 @@
"""Test the Fronius config flow."""
from unittest.mock import patch
from pyfronius import FroniusError
from homeassistant import config_entries
from homeassistant.components.fronius.const import DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import CONF_HOST, CONF_RESOURCE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from homeassistant.setup import async_setup_component
from . import MOCK_HOST, mock_responses
from tests.common import MockConfigEntry
INVERTER_INFO_RETURN_VALUE = {
"inverters": [
{
"device_id": {"value": "1"},
"unique_id": {"value": "1234567"},
}
]
}
LOGGER_INFO_RETURN_VALUE = {"unique_identifier": {"value": "123.4567"}}
async def test_form_with_logger(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
with patch(
"pyfronius.Fronius.current_logger_info",
return_value=LOGGER_INFO_RETURN_VALUE,
), patch(
"homeassistant.components.fronius.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "10.9.8.1",
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "SolarNet Datalogger at 10.9.8.1"
assert result2["data"] == {
"host": "10.9.8.1",
"is_logger": True,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_with_inverter(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
with patch(
"pyfronius.Fronius.current_logger_info",
side_effect=FroniusError,
), patch(
"pyfronius.Fronius.inverter_info",
return_value=INVERTER_INFO_RETURN_VALUE,
), patch(
"homeassistant.components.fronius.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "10.9.1.1",
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "SolarNet Inverter at 10.9.1.1"
assert result2["data"] == {
"host": "10.9.1.1",
"is_logger": False,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"pyfronius.Fronius.current_logger_info",
side_effect=FroniusError,
), patch(
"pyfronius.Fronius.inverter_info",
side_effect=FroniusError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
},
)
assert result2["type"] == RESULT_TYPE_FORM
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_no_device(hass: HomeAssistant) -> None:
"""Test we handle no device found error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"pyfronius.Fronius.current_logger_info",
side_effect=FroniusError,
), patch(
"pyfronius.Fronius.inverter_info",
return_value={"inverters": []},
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
},
)
assert result2["type"] == RESULT_TYPE_FORM
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_unexpected(hass: HomeAssistant) -> None:
"""Test we handle unexpected error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"pyfronius.Fronius.current_logger_info",
side_effect=KeyError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
},
)
assert result2["type"] == RESULT_TYPE_FORM
assert result2["errors"] == {"base": "unknown"}
async def test_form_already_existing(hass):
"""Test existing entry."""
MockConfigEntry(
domain=DOMAIN,
unique_id="123.4567",
data={CONF_HOST: "10.9.8.1", "is_logger": True},
).add_to_hass(hass)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"pyfronius.Fronius.current_logger_info",
return_value=LOGGER_INFO_RETURN_VALUE,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "10.9.8.1",
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_ABORT
assert result2["reason"] == "already_configured"
async def test_form_updates_host(hass, aioclient_mock):
"""Test existing entry gets updated."""
old_host = "http://10.1.0.1"
new_host = "http://10.1.0.2"
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="123.4567890", # has to match mocked logger unique_id
data={
CONF_HOST: old_host,
"is_logger": True,
},
)
entry.add_to_hass(hass)
mock_responses(aioclient_mock, host=old_host)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_responses(aioclient_mock, host=new_host)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": new_host,
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_ABORT
assert result2["reason"] == "already_configured"
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].data == {
"host": new_host,
"is_logger": True,
}
async def test_import(hass, aioclient_mock):
"""Test import step."""
mock_responses(aioclient_mock)
assert await async_setup_component(
hass,
SENSOR_DOMAIN,
{
SENSOR_DOMAIN: {
"platform": DOMAIN,
CONF_RESOURCE: MOCK_HOST,
}
},
)
await hass.async_block_till_done()
fronius_entries = hass.config_entries.async_entries(DOMAIN)
assert len(fronius_entries) == 1
test_entry = fronius_entries[0]
assert test_entry.unique_id == "123.4567890" # has to match mocked logger unique_id
assert test_entry.data == {
"host": MOCK_HOST,
"is_logger": True,
}

55
tests/components/fronius/test_coordinator.py Normal file
View File

@ -0,0 +1,55 @@
"""Test the Fronius update coordinators."""
from unittest.mock import patch
from pyfronius import FroniusError
from homeassistant.components.fronius.coordinator import (
FroniusInverterUpdateCoordinator,
)
from homeassistant.util import dt
from . import mock_responses, setup_fronius_integration
from tests.common import async_fire_time_changed
async def test_adaptive_update_interval(hass, aioclient_mock):
"""Test coordinators changing their update interval when inverter not available."""
with patch("pyfronius.Fronius.current_inverter_data") as mock_inverter_data:
mock_responses(aioclient_mock)
await setup_fronius_integration(hass)
assert mock_inverter_data.call_count == 1
async_fire_time_changed(
hass, dt.utcnow() + FroniusInverterUpdateCoordinator.default_interval
)
await hass.async_block_till_done()
assert mock_inverter_data.call_count == 2
mock_inverter_data.side_effect = FroniusError
# first 3 requests at default interval - 4th has different interval
for _ in range(4):
async_fire_time_changed(
hass, dt.utcnow() + FroniusInverterUpdateCoordinator.default_interval
)
await hass.async_block_till_done()
assert mock_inverter_data.call_count == 5
async_fire_time_changed(
hass, dt.utcnow() + FroniusInverterUpdateCoordinator.error_interval
)
await hass.async_block_till_done()
assert mock_inverter_data.call_count == 6
mock_inverter_data.side_effect = None
# next successful request resets to default interval
async_fire_time_changed(
hass, dt.utcnow() + FroniusInverterUpdateCoordinator.error_interval
)
await hass.async_block_till_done()
assert mock_inverter_data.call_count == 7
async_fire_time_changed(
hass, dt.utcnow() + FroniusInverterUpdateCoordinator.default_interval
)
await hass.async_block_till_done()
assert mock_inverter_data.call_count == 8

23
tests/components/fronius/test_init.py Normal file
View File

@ -0,0 +1,23 @@
"""Test the Fronius integration."""
from homeassistant.components.fronius.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from . import mock_responses, setup_fronius_integration
async def test_unload_config_entry(hass, aioclient_mock):
"""Test that configuration entry supports unloading."""
mock_responses(aioclient_mock)
await setup_fronius_integration(hass)
fronius_entries = hass.config_entries.async_entries(DOMAIN)
assert len(fronius_entries) == 1
test_entry = fronius_entries[0]
assert test_entry.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(test_entry.entry_id)
await hass.async_block_till_done()
assert test_entry.state is ConfigEntryState.NOT_LOADED
assert not hass.data.get(DOMAIN)

128
tests/components/fronius/test_sensor.py
View File

@ -1,66 +1,15 @@
"""Tests for the Fronius sensor platform."""
from homeassistant.components.fronius.sensor import (
CONF_SCOPE,
DEFAULT_SCAN_INTERVAL,
SCOPE_DEVICE,
TYPE_INVERTER,
TYPE_LOGGER_INFO,
TYPE_METER,
TYPE_POWER_FLOW,
from homeassistant.components.fronius.coordinator import (
FroniusInverterUpdateCoordinator,
FroniusPowerFlowUpdateCoordinator,
)
from homeassistant.const import CONF_DEVICE, CONF_SENSOR_TYPE, STATE_UNKNOWN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import STATE_UNKNOWN
from homeassistant.util import dt
from . import setup_fronius_integration
from .const import MOCK_HOST
from . import mock_responses, setup_fronius_integration
from tests.common import async_fire_time_changed, load_fixture
from tests.test_util.aiohttp import AiohttpClientMocker
def mock_responses(aioclient_mock: AiohttpClientMocker, night: bool = False) -> None:
"""Mock responses for Fronius Symo inverter with meter."""
aioclient_mock.clear_requests()
_day_or_night = "night" if night else "day"
aioclient_mock.get(
f"{MOCK_HOST}/solar_api/GetAPIVersion.cgi",
text=load_fixture("symo/GetAPIVersion.json", "fronius"),
)
aioclient_mock.get(
f"{MOCK_HOST}/solar_api/v1/GetInverterRealtimeData.cgi?Scope=Device&"
"DeviceId=1&DataCollection=CommonInverterData",
text=load_fixture(
f"symo/GetInverterRealtimeDate_Device_1_{_day_or_night}.json", "fronius"
),
)
aioclient_mock.get(
f"{MOCK_HOST}/solar_api/v1/GetInverterInfo.cgi",
text=load_fixture("symo/GetInverterInfo.json", "fronius"),
)
aioclient_mock.get(
f"{MOCK_HOST}/solar_api/v1/GetLoggerInfo.cgi",
text=load_fixture("symo/GetLoggerInfo.json", "fronius"),
)
aioclient_mock.get(
f"{MOCK_HOST}/solar_api/v1/GetMeterRealtimeData.cgi?Scope=Device&DeviceId=0",
text=load_fixture("symo/GetMeterRealtimeData_Device_0.json", "fronius"),
)
aioclient_mock.get(
f"{MOCK_HOST}/solar_api/v1/GetMeterRealtimeData.cgi?Scope=System",
text=load_fixture("symo/GetMeterRealtimeData_System.json", "fronius"),
)
aioclient_mock.get(
f"{MOCK_HOST}/solar_api/v1/GetPowerFlowRealtimeData.fcgi",
text=load_fixture(
f"symo/GetPowerFlowRealtimeData_{_day_or_night}.json", "fronius"
),
)
aioclient_mock.get(
f"{MOCK_HOST}/solar_api/v1/GetStorageRealtimeData.cgi?Scope=System",
text=load_fixture("symo/GetStorageRealtimeData_System.json", "fronius"),
)
from tests.common import async_fire_time_changed
async def test_symo_inverter(hass, aioclient_mock):
@ -72,15 +21,9 @@ async def test_symo_inverter(hass, aioclient_mock):
# Init at night
mock_responses(aioclient_mock, night=True)
config = {
CONF_SENSOR_TYPE: TYPE_INVERTER,
CONF_SCOPE: SCOPE_DEVICE,
CONF_DEVICE: 1,
}
await setup_fronius_integration(hass, [config])
await setup_fronius_integration(hass)
assert len(hass.states.async_all()) == 10
# 5 ignored from DeviceStatus
assert len(hass.states.async_all(domain_filter=SENSOR_DOMAIN)) == 55
assert_state("sensor.current_dc_fronius_inverter_1_http_fronius", 0)
assert_state("sensor.energy_day_fronius_inverter_1_http_fronius", 10828)
assert_state("sensor.energy_total_fronius_inverter_1_http_fronius", 44186900)
@ -89,10 +32,12 @@ async def test_symo_inverter(hass, aioclient_mock):
# Second test at daytime when inverter is producing
mock_responses(aioclient_mock, night=False)
async_fire_time_changed(hass, dt.utcnow() + DEFAULT_SCAN_INTERVAL)
async_fire_time_changed(
hass, dt.utcnow() + FroniusInverterUpdateCoordinator.default_interval
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 14
assert len(hass.states.async_all(domain_filter=SENSOR_DOMAIN)) == 59
# 4 additional AC entities
assert_state("sensor.current_dc_fronius_inverter_1_http_fronius", 2.19)
assert_state("sensor.energy_day_fronius_inverter_1_http_fronius", 1113)
@ -114,12 +59,9 @@ async def test_symo_logger(hass, aioclient_mock):
assert state.state == str(expected_state)
mock_responses(aioclient_mock)
config = {
CONF_SENSOR_TYPE: TYPE_LOGGER_INFO,
}
await setup_fronius_integration(hass, [config])
await setup_fronius_integration(hass)
assert len(hass.states.async_all()) == 12
assert len(hass.states.async_all(domain_filter=SENSOR_DOMAIN)) == 59
# ignored constant entities:
# hardware_platform, hardware_version, product_type
# software_version, time_zone, time_zone_location
@ -128,7 +70,7 @@ async def test_symo_logger(hass, aioclient_mock):
# states are rounded to 2 decimals
assert_state(
"sensor.cash_factor_fronius_logger_info_0_http_fronius",
0.08,
0.078,
)
assert_state(
"sensor.co2_factor_fronius_logger_info_0_http_fronius",
@ -149,21 +91,16 @@ async def test_symo_meter(hass, aioclient_mock):
assert state.state == str(expected_state)
mock_responses(aioclient_mock)
config = {
CONF_SENSOR_TYPE: TYPE_METER,
CONF_SCOPE: SCOPE_DEVICE,
CONF_DEVICE: 0,
}
await setup_fronius_integration(hass, [config])
await setup_fronius_integration(hass)
assert len(hass.states.async_all()) == 39
assert len(hass.states.async_all(domain_filter=SENSOR_DOMAIN)) == 59
# ignored entities:
# manufacturer, model, serial, enable, timestamp, visible, meter_location
#
# states are rounded to 2 decimals
assert_state("sensor.current_ac_phase_1_fronius_meter_0_http_fronius", 7.75)
assert_state("sensor.current_ac_phase_1_fronius_meter_0_http_fronius", 7.755)
assert_state("sensor.current_ac_phase_2_fronius_meter_0_http_fronius", 6.68)
assert_state("sensor.current_ac_phase_3_fronius_meter_0_http_fronius", 10.1)
assert_state("sensor.current_ac_phase_3_fronius_meter_0_http_fronius", 10.102)
assert_state(
"sensor.energy_reactive_ac_consumed_fronius_meter_0_http_fronius", 59960790
)
@ -175,9 +112,9 @@ async def test_symo_meter(hass, aioclient_mock):
assert_state("sensor.energy_real_consumed_fronius_meter_0_http_fronius", 15303334)
assert_state("sensor.energy_real_produced_fronius_meter_0_http_fronius", 35623065)
assert_state("sensor.frequency_phase_average_fronius_meter_0_http_fronius", 50)
assert_state("sensor.power_apparent_phase_1_fronius_meter_0_http_fronius", 1772.79)
assert_state("sensor.power_apparent_phase_2_fronius_meter_0_http_fronius", 1527.05)
assert_state("sensor.power_apparent_phase_3_fronius_meter_0_http_fronius", 2333.56)
assert_state("sensor.power_apparent_phase_1_fronius_meter_0_http_fronius", 1772.793)
assert_state("sensor.power_apparent_phase_2_fronius_meter_0_http_fronius", 1527.048)
assert_state("sensor.power_apparent_phase_3_fronius_meter_0_http_fronius", 2333.562)
assert_state("sensor.power_apparent_fronius_meter_0_http_fronius", 5592.57)
assert_state("sensor.power_factor_phase_1_fronius_meter_0_http_fronius", -0.99)
assert_state("sensor.power_factor_phase_2_fronius_meter_0_http_fronius", -0.99)
@ -215,12 +152,9 @@ async def test_symo_power_flow(hass, aioclient_mock):
# First test at night
mock_responses(aioclient_mock, night=True)
config = {
CONF_SENSOR_TYPE: TYPE_POWER_FLOW,
}
await setup_fronius_integration(hass, [config])
await setup_fronius_integration(hass)
assert len(hass.states.async_all()) == 12
assert len(hass.states.async_all(domain_filter=SENSOR_DOMAIN)) == 55
# ignored: location, mode, timestamp
#
# states are rounded to 2 decimals
@ -263,13 +197,15 @@ async def test_symo_power_flow(hass, aioclient_mock):
# Second test at daytime when inverter is producing
mock_responses(aioclient_mock, night=False)
async_fire_time_changed(hass, dt.utcnow() + DEFAULT_SCAN_INTERVAL)
async_fire_time_changed(
hass, dt.utcnow() + FroniusPowerFlowUpdateCoordinator.default_interval
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 12
# still 55 because power_flow update interval is shorter than others
assert len(hass.states.async_all(domain_filter=SENSOR_DOMAIN)) == 55
assert_state(
"sensor.energy_day_fronius_power_flow_0_http_fronius",
1101.70,
1101.7001,
)
assert_state(
"sensor.energy_total_fronius_power_flow_0_http_fronius",
@ -297,7 +233,7 @@ async def test_symo_power_flow(hass, aioclient_mock):
)
assert_state(
"sensor.relative_autonomy_fronius_power_flow_0_http_fronius",
39.47,
39.4708,
)
assert_state(
"sensor.relative_self_consumption_fronius_power_flow_0_http_fronius",