evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

New integration for Kostal Plenticore solar inverters (#43404)

Browse source 
* New integration for Kostal Plenticore solar inverters.

* Fix errors from github pipeline.

* Fixed test for py37.

* Add more test for test coverage check.

* Try to fix test coverage check.

* Fix import sort order.

* Try fix test code coverage .

* Mock api client for tests.

* Fix typo.

* Fix order of rebased code from dev.

* Add new data point for home power.

* Modifications to review.
Remove service for write access (for first pull request). Refactor
update coordinator to not use the entity API.

* Fixed mock imports.

* Ignore new python module on coverage.

* Changes after review.

* Fixed unit test because of config title.

* Fixes from review.

* Changes from review (unique id and mocking of tests)

* Use async update method. Change unique id. Remove _dict

* Remove _data field.

* Removed login flag from PlenticoreUpdateCoordinator.

* Removed Dynamic SoC sensor because it should be a binary sensor.

* Remove more sensors because they are binary sensors.
This commit is contained in:
stegm 2021-04-07 09:18:07 +02:00 • committed by GitHub
parent 815db999da
commit 589f2240b1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 1383 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.coveragerc
View file

@ -506,6 +506,10 @@ omit =
homeassistant/components/kodi/media_player.py
homeassistant/components/kodi/notify.py
homeassistant/components/konnected/*
homeassistant/components/kostal_plenticore/__init__.py
homeassistant/components/kostal_plenticore/const.py
homeassistant/components/kostal_plenticore/helper.py
homeassistant/components/kostal_plenticore/sensor.py
homeassistant/components/kwb/sensor.py
homeassistant/components/lacrosse/sensor.py
homeassistant/components/lametric/*

1
CODEOWNERS
View file

@ -250,6 +250,7 @@ homeassistant/components/kmtronic/* @dgomes
homeassistant/components/knx/* @Julius2342 @farmio @marvin-w
homeassistant/components/kodi/* @OnFreund @cgtobi
homeassistant/components/konnected/* @heythisisnate @kit-klein
homeassistant/components/kostal_plenticore/* @stegm
homeassistant/components/kulersky/* @emlove
homeassistant/components/lametric/* @robbiet480
homeassistant/components/launch_library/* @ludeeus

60
homeassistant/components/kostal_plenticore/__init__.py Normal file
View file

@ -0,0 +1,60 @@
"""The Kostal Plenticore Solar Inverter integration."""
import asyncio
import logging
from kostal.plenticore import PlenticoreApiException
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import DOMAIN
from .helper import Plenticore
_LOGGER = logging.getLogger(__name__)
PLATFORMS = ["sensor"]
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
"""Set up the Kostal Plenticore Solar Inverter component."""
hass.data.setdefault(DOMAIN, {})
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Kostal Plenticore Solar Inverter from a config entry."""
plenticore = Plenticore(hass, entry)
if not await plenticore.async_setup():
return False
hass.data[DOMAIN][entry.entry_id] = plenticore
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)
if unload_ok:
# remove API object
plenticore = hass.data[DOMAIN].pop(entry.entry_id)
try:
await plenticore.async_unload()
except PlenticoreApiException as err:
_LOGGER.error("Error logging out from inverter: %s", err)
return unload_ok

78
homeassistant/components/kostal_plenticore/config_flow.py Normal file
View file

@ -0,0 +1,78 @@
"""Config flow for Kostal Plenticore Solar Inverter integration."""
import asyncio
import logging
from aiohttp.client_exceptions import ClientError
from kostal.plenticore import PlenticoreApiClient, PlenticoreAuthenticationException
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_BASE, CONF_HOST, CONF_PASSWORD
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_PASSWORD): str,
}
)
@callback
def configured_instances(hass):
"""Return a set of configured Kostal Plenticore HOSTS."""
return {
entry.data[CONF_HOST] for entry in hass.config_entries.async_entries(DOMAIN)
}
async def test_connection(hass: HomeAssistant, data) -> str:
"""Test the connection to the inverter.
Data has the keys from DATA_SCHEMA with values provided by the user.
"""
session = async_get_clientsession(hass)
async with PlenticoreApiClient(session, data["host"]) as client:
await client.login(data["password"])
values = await client.get_setting_values("scb:network", "Hostname")
return values["scb:network"]["Hostname"]
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Kostal Plenticore Solar Inverter."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
errors = {}
hostname = None
if user_input is not None:
if user_input[CONF_HOST] in configured_instances(self.hass):
return self.async_abort(reason="already_configured")
try:
hostname = await test_connection(self.hass, user_input)
except PlenticoreAuthenticationException as ex:
errors[CONF_PASSWORD] = "invalid_auth"
_LOGGER.error("Error response: %s", ex)
except (ClientError, asyncio.TimeoutError):
errors[CONF_HOST] = "cannot_connect"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors[CONF_BASE] = "unknown"
if not errors:
return self.async_create_entry(title=hostname, data=user_input)
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA, errors=errors
)

521
homeassistant/components/kostal_plenticore/const.py Normal file
View file

@ -0,0 +1,521 @@
"""Constants for the Kostal Plenticore Solar Inverter integration."""
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_ICON,
ATTR_UNIT_OF_MEASUREMENT,
DEVICE_CLASS_BATTERY,
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_POWER,
ENERGY_KILO_WATT_HOUR,
PERCENTAGE,
POWER_WATT,
)
DOMAIN = "kostal_plenticore"
ATTR_ENABLED_DEFAULT = "entity_registry_enabled_default"
# Defines all entities for process data.
#
# Each entry is defined with a tuple of these values:
# - module id (str)
# - process data id (str)
# - entity name suffix (str)
# - sensor properties (dict)
# - value formatter (str)
SENSOR_PROCESS_DATA = [
(
"devices:local",
"Inverter:State",
"Inverter State",
{ATTR_ICON: "mdi:state-machine"},
"format_inverter_state",
),
(
"devices:local",
"Dc_P",
"Solar Power",
{
ATTR_UNIT_OF_MEASUREMENT: POWER_WATT,
ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER,
ATTR_ENABLED_DEFAULT: True,
},
"format_round",
),
(
"devices:local",
"Grid_P",
"Grid Power",
{
ATTR_UNIT_OF_MEASUREMENT: POWER_WATT,
ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER,
ATTR_ENABLED_DEFAULT: True,
},
"format_round",
),
(
"devices:local",
"HomeBat_P",
"Home Power from Battery",
{ATTR_UNIT_OF_MEASUREMENT: POWER_WATT, ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER},
"format_round",
),
(
"devices:local",
"HomeGrid_P",
"Home Power from Grid",
{ATTR_UNIT_OF_MEASUREMENT: POWER_WATT, ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER},
"format_round",
),
(
"devices:local",
"HomeOwn_P",
"Home Power from Own",
{ATTR_UNIT_OF_MEASUREMENT: POWER_WATT, ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER},
"format_round",
),
(
"devices:local",
"HomePv_P",
"Home Power from PV",
{ATTR_UNIT_OF_MEASUREMENT: POWER_WATT, ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER},
"format_round",
),
(
"devices:local",
"Home_P",
"Home Power",
{ATTR_UNIT_OF_MEASUREMENT: POWER_WATT, ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER},
"format_round",
),
(
"devices:local:ac",
"P",
"AC Power",
{
ATTR_UNIT_OF_MEASUREMENT: POWER_WATT,
ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER,
ATTR_ENABLED_DEFAULT: True,
},
"format_round",
),
(
"devices:local:pv1",
"P",
"DC1 Power",
{ATTR_UNIT_OF_MEASUREMENT: POWER_WATT, ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER},
"format_round",
),
(
"devices:local:pv2",
"P",
"DC2 Power",
{ATTR_UNIT_OF_MEASUREMENT: POWER_WATT, ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER},
"format_round",
),
(
"devices:local",
"PV2Bat_P",
"PV to Battery Power",
{ATTR_UNIT_OF_MEASUREMENT: POWER_WATT, ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER},
"format_round",
),
(
"devices:local",
"EM_State",
"Energy Manager State",
{ATTR_ICON: "mdi:state-machine"},
"format_em_manager_state",
),
(
"devices:local:battery",
"Cycles",
"Battery Cycles",
{ATTR_ICON: "mdi:recycle"},
"format_round",
),
(
"devices:local:battery",
"P",
"Battery Power",
{ATTR_UNIT_OF_MEASUREMENT: POWER_WATT, ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER},
"format_round",
),
(
"devices:local:battery",
"SoC",
"Battery SoC",
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_DEVICE_CLASS: DEVICE_CLASS_BATTERY},
"format_round",
),
(
"scb:statistic:EnergyFlow",
"Statistic:Autarky:Day",
"Autarky Day",
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_ICON: "mdi:chart-donut"},
"format_round",
),
(
"scb:statistic:EnergyFlow",
"Statistic:Autarky:Month",
"Autarky Month",
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_ICON: "mdi:chart-donut"},
"format_round",
),
(
"scb:statistic:EnergyFlow",
"Statistic:Autarky:Total",
"Autarky Total",
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_ICON: "mdi:chart-donut"},
"format_round",
),
(
"scb:statistic:EnergyFlow",
"Statistic:Autarky:Year",
"Autarky Year",
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_ICON: "mdi:chart-donut"},
"format_round",
),
(
"scb:statistic:EnergyFlow",
"Statistic:OwnConsumptionRate:Day",
"Own Consumption Rate Day",
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_ICON: "mdi:chart-donut"},
"format_round",
),
(
"scb:statistic:EnergyFlow",
"Statistic:OwnConsumptionRate:Month",
"Own Consumption Rate Month",
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_ICON: "mdi:chart-donut"},
"format_round",
),
(
"scb:statistic:EnergyFlow",
"Statistic:OwnConsumptionRate:Total",
"Own Consumption Rate Total",
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_ICON: "mdi:chart-donut"},
"format_round",
),
(
"scb:statistic:EnergyFlow",
"Statistic:OwnConsumptionRate:Year",
"Own Consumption Rate Year",
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_ICON: "mdi:chart-donut"},
"format_round",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHome:Day",
"Home Consumption Day",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHome:Month",
"Home Consumption Month",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHome:Year",
"Home Consumption Year",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHome:Total",
"Home Consumption Total",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomeBat:Day",
"Home Consumption from Battery Day",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomeBat:Month",
"Home Consumption from Battery Month",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomeBat:Year",
"Home Consumption from Battery Year",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomeBat:Total",
"Home Consumption from Battery Total",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomeGrid:Day",
"Home Consumption from Grid Day",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomeGrid:Month",
"Home Consumption from Grid Month",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomeGrid:Year",
"Home Consumption from Grid Year",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomeGrid:Total",
"Home Consumption from Grid Total",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomePv:Day",
"Home Consumption from PV Day",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomePv:Month",
"Home Consumption from PV Month",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomePv:Year",
"Home Consumption from PV Year",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyHomePv:Total",
"Home Consumption from PV Total",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyPv1:Day",
"Energy PV1 Day",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyPv1:Month",
"Energy PV1 Month",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyPv1:Year",
"Energy PV1 Year",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyPv1:Total",
"Energy PV1 Total",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyPv2:Day",
"Energy PV2 Day",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyPv2:Month",
"Energy PV2 Month",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyPv2:Year",
"Energy PV2 Year",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:EnergyPv2:Total",
"Energy PV2 Total",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:Yield:Day",
"Energy Yield Day",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
ATTR_ENABLED_DEFAULT: True,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:Yield:Month",
"Energy Yield Month",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:Yield:Year",
"Energy Yield Year",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
(
"scb:statistic:EnergyFlow",
"Statistic:Yield:Total",
"Energy Yield Total",
{
ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
ATTR_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"format_energy",
),
]
# Defines all entities for settings.
#
# Each entry is defined with a tuple of these values:
# - module id (str)
# - process data id (str)
# - entity name suffix (str)
# - sensor properties (dict)
# - value formatter (str)
SENSOR_SETTINGS_DATA = [
(
"devices:local",
"Battery:MinHomeComsumption",
"Battery min Home Consumption",
{ATTR_UNIT_OF_MEASUREMENT: POWER_WATT, ATTR_DEVICE_CLASS: DEVICE_CLASS_POWER},
"format_round",
),
(
"devices:local",
"Battery:MinSoc",
"Battery min Soc",
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_ICON: "mdi:battery-negative"},
"format_round",
),
(
"devices:local",
"Battery:Strategy",
"Battery Strategy",
{},
"format_round",
),
]

259
homeassistant/components/kostal_plenticore/helper.py Normal file
View file

@ -0,0 +1,259 @@
"""Code to handle the Plenticore API."""
import asyncio
from collections import defaultdict
from datetime import datetime, timedelta
import logging
from typing import Dict, Union
from aiohttp.client_exceptions import ClientError
from kostal.plenticore import PlenticoreApiClient, PlenticoreAuthenticationException
from homeassistant.const import CONF_HOST, CONF_PASSWORD, 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.event import async_call_later
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class Plenticore:
"""Manages the Plenticore API."""
def __init__(self, hass, config_entry):
"""Create a new plenticore manager instance."""
self.hass = hass
self.config_entry = config_entry
self._client = None
self._shutdown_remove_listener = None
self.device_info = {}
@property
def host(self) -> str:
"""Return the host of the Plenticore inverter."""
return self.config_entry.data[CONF_HOST]
@property
def client(self) -> PlenticoreApiClient:
"""Return the Plenticore API client."""
return self._client
async def async_setup(self) -> bool:
"""Set up Plenticore API client."""
self._client = PlenticoreApiClient(
async_get_clientsession(self.hass), host=self.host
)
try:
await self._client.login(self.config_entry.data[CONF_PASSWORD])
except PlenticoreAuthenticationException as err:
_LOGGER.error(
"Authentication exception connecting to %s: %s", self.host, err
)
return False
except (ClientError, asyncio.TimeoutError) as err:
_LOGGER.error("Error connecting to %s", self.host)
raise ConfigEntryNotReady from err
else:
_LOGGER.debug("Log-in successfully to %s", self.host)
self._shutdown_remove_listener = self.hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, self._async_shutdown
)
# get some device meta data
settings = await self._client.get_setting_values(
{
"devices:local": [
"Properties:SerialNo",
"Branding:ProductName1",
"Branding:ProductName2",
"Properties:VersionIOC",
"Properties:VersionMC",
],
"scb:network": ["Hostname"],
}
)
device_local = settings["devices:local"]
prod1 = device_local["Branding:ProductName1"]
prod2 = device_local["Branding:ProductName2"]
self.device_info = {
"identifiers": {(DOMAIN, device_local["Properties:SerialNo"])},
"manufacturer": "Kostal",
"model": f"{prod1} {prod2}",
"name": settings["scb:network"]["Hostname"],
"sw_version": f'IOC: {device_local["Properties:VersionIOC"]}'
+ f' MC: {device_local["Properties:VersionMC"]}',
}
return True
async def _async_shutdown(self, event):
"""Call from Homeassistant shutdown event."""
# unset remove listener otherwise calling it would raise an exception
self._shutdown_remove_listener = None
await self.async_unload()
async def async_unload(self) -> None:
"""Unload the Plenticore API client."""
if self._shutdown_remove_listener:
self._shutdown_remove_listener()
await self._client.logout()
self._client = None
_LOGGER.debug("Logged out from %s", self.host)
class PlenticoreUpdateCoordinator(DataUpdateCoordinator):
"""Base implementation of DataUpdateCoordinator for Plenticore data."""
def __init__(
self,
hass: HomeAssistant,
logger: logging.Logger,
name: str,
update_inverval: timedelta,
plenticore: Plenticore,
):
"""Create a new update coordinator for plenticore data."""
super().__init__(
hass=hass,
logger=logger,
name=name,
update_interval=update_inverval,
)
# data ids to poll
self._fetch = defaultdict(list)
self._plenticore = plenticore
def start_fetch_data(self, module_id: str, data_id: str) -> None:
"""Start fetching the given data (module-id and data-id)."""
self._fetch[module_id].append(data_id)
# Force an update of all data. Multiple refresh calls
# are ignored by the debouncer.
async def force_refresh(event_time: datetime) -> None:
await self.async_request_refresh()
async_call_later(self.hass, 2, force_refresh)
def stop_fetch_data(self, module_id: str, data_id: str) -> None:
"""Stop fetching the given data (module-id and data-id)."""
self._fetch[module_id].remove(data_id)
class ProcessDataUpdateCoordinator(PlenticoreUpdateCoordinator):
"""Implementation of PlenticoreUpdateCoordinator for process data."""
async def _async_update_data(self) -> Dict[str, Dict[str, str]]:
client = self._plenticore.client
if not self._fetch or client is None:
return {}
_LOGGER.debug("Fetching %s for %s", self.name, self._fetch)
fetched_data = await client.get_process_data_values(self._fetch)
return {
module_id: {
process_data.id: process_data.value
for process_data in fetched_data[module_id]
}
for module_id in fetched_data
}
class SettingDataUpdateCoordinator(PlenticoreUpdateCoordinator):
"""Implementation of PlenticoreUpdateCoordinator for settings data."""
async def _async_update_data(self) -> Dict[str, Dict[str, str]]:
client = self._plenticore.client
if not self._fetch or client is None:
return {}
_LOGGER.debug("Fetching %s for %s", self.name, self._fetch)
fetched_data = await client.get_setting_values(self._fetch)
return fetched_data
class PlenticoreDataFormatter:
"""Provides method to format values of process or settings data."""
INVERTER_STATES = {
0: "Off",
1: "Init",
2: "IsoMEas",
3: "GridCheck",
4: "StartUp",
6: "FeedIn",
7: "Throttled",
8: "ExtSwitchOff",
9: "Update",
10: "Standby",
11: "GridSync",
12: "GridPreCheck",
13: "GridSwitchOff",
14: "Overheating",
15: "Shutdown",
16: "ImproperDcVoltage",
17: "ESB",
}
EM_STATES = {
0: "Idle",
1: "n/a",
2: "Emergency Battery Charge",
4: "n/a",
8: "Winter Mode Step 1",
16: "Winter Mode Step 2",
}
@classmethod
def get_method(cls, name: str) -> callable:
"""Return a callable formatter of the given name."""
return getattr(cls, name)
@staticmethod
def format_round(state: str) -> Union[int, str]:
"""Return the given state value as rounded integer."""
try:
return round(float(state))
except (TypeError, ValueError):
return state
@staticmethod
def format_energy(state: str) -> Union[float, str]:
"""Return the given state value as energy value, scaled to kWh."""
try:
return round(float(state) / 1000, 1)
except (TypeError, ValueError):
return state
@staticmethod
def format_inverter_state(state: str) -> str:
"""Return a readable string of the inverter state."""
try:
value = int(state)
except (TypeError, ValueError):
return state
return PlenticoreDataFormatter.INVERTER_STATES.get(value)
@staticmethod
def format_em_manager_state(state: str) -> str:
"""Return a readable state of the energy manager."""
try:
value = int(state)
except (TypeError, ValueError):
return state
return PlenticoreDataFormatter.EM_STATES.get(value)

10
homeassistant/components/kostal_plenticore/manifest.json Normal file
View file

@ -0,0 +1,10 @@
{
"domain": "kostal_plenticore",
"name": "Kostal Plenticore Solar Inverter",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/kostal_plenticore",
"requirements": ["kostal_plenticore==0.2.0"],
"codeowners": [
"@stegm"
]
}

193
homeassistant/components/kostal_plenticore/sensor.py Normal file
View file

@ -0,0 +1,193 @@
"""Platform for Kostal Plenticore sensors."""
from datetime import timedelta
import logging
from typing import Any, Callable, Dict, Optional
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_DEVICE_CLASS, ATTR_ICON, ATTR_UNIT_OF_MEASUREMENT
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
ATTR_ENABLED_DEFAULT,
DOMAIN,
SENSOR_PROCESS_DATA,
SENSOR_SETTINGS_DATA,
)
from .helper import (
PlenticoreDataFormatter,
ProcessDataUpdateCoordinator,
SettingDataUpdateCoordinator,
)
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities
):
"""Add kostal plenticore Sensors."""
plenticore = hass.data[DOMAIN][entry.entry_id]
entities = []
available_process_data = await plenticore.client.get_process_data()
process_data_update_coordinator = ProcessDataUpdateCoordinator(
hass,
_LOGGER,
"Process Data",
timedelta(seconds=10),
plenticore,
)
for module_id, data_id, name, sensor_data, fmt in SENSOR_PROCESS_DATA:
if (
module_id not in available_process_data
or data_id not in available_process_data[module_id]
):
_LOGGER.debug(
"Skipping non existing process data %s/%s", module_id, data_id
)
continue
entities.append(
PlenticoreDataSensor(
process_data_update_coordinator,
entry.entry_id,
entry.title,
module_id,
data_id,
name,
sensor_data,
PlenticoreDataFormatter.get_method(fmt),
plenticore.device_info,
)
)
available_settings_data = await plenticore.client.get_settings()
settings_data_update_coordinator = SettingDataUpdateCoordinator(
hass,
_LOGGER,
"Settings Data",
timedelta(seconds=300),
plenticore,
)
for module_id, data_id, name, sensor_data, fmt in SENSOR_SETTINGS_DATA:
if module_id not in available_settings_data or data_id not in (
setting.id for setting in available_settings_data[module_id]
):
_LOGGER.debug(
"Skipping non existing setting data %s/%s", module_id, data_id
)
continue
entities.append(
PlenticoreDataSensor(
settings_data_update_coordinator,
entry.entry_id,
entry.title,
module_id,
data_id,
name,
sensor_data,
PlenticoreDataFormatter.get_method(fmt),
plenticore.device_info,
)
)
async_add_entities(entities)
class PlenticoreDataSensor(CoordinatorEntity, SensorEntity):
"""Representation of a Plenticore data Sensor."""
def __init__(
self,
coordinator,
entry_id: str,
platform_name: str,
module_id: str,
data_id: str,
sensor_name: str,
sensor_data: Dict[str, Any],
formatter: Callable[[str], Any],
device_info: Dict[str, Any],
):
"""Create a new Sensor Entity for Plenticore process data."""
super().__init__(coordinator)
self.entry_id = entry_id
self.platform_name = platform_name
self.module_id = module_id
self.data_id = data_id
self._sensor_name = sensor_name
self._sensor_data = sensor_data
self._formatter = formatter
self._device_info = device_info
@property
def available(self) -> bool:
"""Return if entity is available."""
return (
super().available
and self.coordinator.data is not None
and self.module_id in self.coordinator.data
and self.data_id in self.coordinator.data[self.module_id]
)
async def async_added_to_hass(self) -> None:
"""Register this entity on the Update Coordinator."""
await super().async_added_to_hass()
self.coordinator.start_fetch_data(self.module_id, self.data_id)
async def async_will_remove_from_hass(self) -> None:
"""Unregister this entity from the Update Coordinator."""
self.coordinator.stop_fetch_data(self.module_id, self.data_id)
await super().async_will_remove_from_hass()
@property
def device_info(self) -> Dict[str, Any]:
"""Return the device info."""
return self._device_info
@property
def unique_id(self) -> str:
"""Return the unique id of this Sensor Entity."""
return f"{self.entry_id}_{self.module_id}_{self.data_id}"
@property
def name(self) -> str:
"""Return the name of this Sensor Entity."""
return f"{self.platform_name} {self._sensor_name}"
@property
def unit_of_measurement(self) -> Optional[str]:
"""Return the unit of this Sensor Entity or None."""
return self._sensor_data.get(ATTR_UNIT_OF_MEASUREMENT)
@property
def icon(self) -> Optional[str]:
"""Return the icon name of this Sensor Entity or None."""
return self._sensor_data.get(ATTR_ICON)
@property
def device_class(self) -> Optional[str]:
"""Return the class of this device, from component DEVICE_CLASSES."""
return self._sensor_data.get(ATTR_DEVICE_CLASS)
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
return self._sensor_data.get(ATTR_ENABLED_DEFAULT, False)
@property
def state(self) -> Optional[Any]:
"""Return the state of the sensor."""
if self.coordinator.data is None:
# None is translated to STATE_UNKNOWN
return None
raw_value = self.coordinator.data[self.module_id][self.data_id]
return self._formatter(raw_value) if self._formatter else raw_value

21
homeassistant/components/kostal_plenticore/strings.json Normal file
View file

@ -0,0 +1,21 @@
{
"title": "Kostal Plenticore Solar Inverter",
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

22
homeassistant/components/kostal_plenticore/translations/en.json Normal file
View file

@ -0,0 +1,22 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"timeout": "Timeout/No answer",
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"host": "Host",
"password": "Password"
}
}
}
},
"title": "Kostal Plenticore Solar Inverter"
}

1
homeassistant/generated/config_flows.py
View file

@ -124,6 +124,7 @@ FLOWS = [
"kmtronic",
"kodi",
"konnected",
"kostal_plenticore",
"kulersky",
"life360",
"lifx",

3
requirements_all.txt
View file

@ -848,6 +848,9 @@ kiwiki-client==0.1.1
# homeassistant.components.konnected
konnected==1.2.0
# homeassistant.components.kostal_plenticore
kostal_plenticore==0.2.0
# homeassistant.components.eufy
lakeside==0.12

3
requirements_test_all.txt
View file

@ -468,6 +468,9 @@ jsonpath==0.82
# homeassistant.components.konnected
konnected==1.2.0
# homeassistant.components.kostal_plenticore
kostal_plenticore==0.2.0
# homeassistant.components.dyson
libpurecool==0.6.4

1
tests/components/kostal_plenticore/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Tests for the Kostal Plenticore Solar Inverter integration."""

206
tests/components/kostal_plenticore/test_config_flow.py Normal file
View file

@ -0,0 +1,206 @@
"""Test the Kostal Plenticore Solar Inverter config flow."""
import asyncio
from unittest.mock import ANY, AsyncMock, MagicMock, patch
from kostal.plenticore import PlenticoreAuthenticationException
from homeassistant import config_entries, setup
from homeassistant.components.kostal_plenticore import config_flow
from homeassistant.components.kostal_plenticore.const import DOMAIN
from tests.common import MockConfigEntry
async def test_formx(hass):
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] == {}
with patch(
"homeassistant.components.kostal_plenticore.config_flow.PlenticoreApiClient"
) as mock_api_class, patch(
"homeassistant.components.kostal_plenticore.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.kostal_plenticore.async_setup_entry",
return_value=True,
) as mock_setup_entry:
# mock of the context manager instance
mock_api_ctx = MagicMock()
mock_api_ctx.login = AsyncMock()
mock_api_ctx.get_setting_values = AsyncMock(
return_value={"scb:network": {"Hostname": "scb"}}
)
# mock of the return instance of PlenticoreApiClient
mock_api = MagicMock()
mock_api.__aenter__.return_value = mock_api_ctx
mock_api.__aexit__ = AsyncMock()
mock_api_class.return_value = mock_api
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"password": "test-password",
},
)
mock_api_class.assert_called_once_with(ANY, "1.1.1.1")
mock_api.__aenter__.assert_called_once()
mock_api.__aexit__.assert_called_once()
mock_api_ctx.login.assert_called_once_with("test-password")
mock_api_ctx.get_setting_values.assert_called_once()
assert result2["type"] == "create_entry"
assert result2["title"] == "scb"
assert result2["data"] == {
"host": "1.1.1.1",
"password": "test-password",
}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_invalid_auth(hass):
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.kostal_plenticore.config_flow.PlenticoreApiClient"
) as mock_api_class:
# mock of the context manager instance
mock_api_ctx = MagicMock()
mock_api_ctx.login = AsyncMock(
side_effect=PlenticoreAuthenticationException(404, "invalid user"),
)
# mock of the return instance of PlenticoreApiClient
mock_api = MagicMock()
mock_api.__aenter__.return_value = mock_api_ctx
mock_api.__aexit__.return_value = None
mock_api_class.return_value = mock_api
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"password": "test-password",
},
)
assert result2["type"] == "form"
assert result2["errors"] == {"password": "invalid_auth"}
async def test_form_cannot_connect(hass):
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.kostal_plenticore.config_flow.PlenticoreApiClient"
) as mock_api_class:
# mock of the context manager instance
mock_api_ctx = MagicMock()
mock_api_ctx.login = AsyncMock(
side_effect=asyncio.TimeoutError(),
)
# mock of the return instance of PlenticoreApiClient
mock_api = MagicMock()
mock_api.__aenter__.return_value = mock_api_ctx
mock_api.__aexit__.return_value = None
mock_api_class.return_value = mock_api
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"password": "test-password",
},
)
assert result2["type"] == "form"
assert result2["errors"] == {"host": "cannot_connect"}
async def test_form_unexpected_error(hass):
"""Test we handle unexpected error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.kostal_plenticore.config_flow.PlenticoreApiClient"
) as mock_api_class:
# mock of the context manager instance
mock_api_ctx = MagicMock()
mock_api_ctx.login = AsyncMock(
side_effect=Exception(),
)
# mock of the return instance of PlenticoreApiClient
mock_api = MagicMock()
mock_api.__aenter__.return_value = mock_api_ctx
mock_api.__aexit__.return_value = None
mock_api_class.return_value = mock_api
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"password": "test-password",
},
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "unknown"}
async def test_already_configured(hass):
"""Test we handle already configured error."""
MockConfigEntry(
domain="kostal_plenticore",
data={"host": "1.1.1.1", "password": "foobar"},
unique_id="112233445566",
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"password": "test-password",
},
)
assert result2["type"] == "abort"
assert result2["reason"] == "already_configured"
def test_configured_instances(hass):
"""Test configured_instances returns all configured hosts."""
MockConfigEntry(
domain="kostal_plenticore",
data={"host": "2.2.2.2", "password": "foobar"},
unique_id="112233445566",
).add_to_hass(hass)
result = config_flow.configured_instances(hass)
assert result == {"2.2.2.2"}