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

Move Kostal Plenticore writable settings from sensor to select widget or switch (#56529) 

* Move "Battery:SmartBatteryControl:Enable" from a simple sensor to a switch
Add "Battery:TimeControl:Enable" as a switch

If you want to change charging behavior you need to turn off both switches, before you can enable the function you want. (Same as on Plenticore UI)

* removed:
    @property
    def assumed_state(self) -> bool

was copied from an switchbot integration, does not make sense or does deliver valuable information

Tried to set constant properties in the constructor

* correct typo, add new line at eof

* Initial state of switch was missing after (re)starting HA. Now working.

* Reformatted with black

* correct syntax errors from test run 09.10.2021

* reformat

* update 15.10.2021

* Set select value is working

* update 05.11.2021

* data correctly received

* working completly

* remove old switch definitions, now replaced by select widget

* correct complaints from workflow run on 11/11/2021

* Add explanatory comment for switch and select

* Correct comments

* Removed function async def async_read_data(self, module_id: str, data_id: str)
from class SettingDataUpdateCoordinator

* Add Mixin class for read/write

* try to make select.py less "stale"

* new dev environment 2

* new dev environment 2

* correct syntax

* minor coding standard correction

* Remove BOM

* Remove BOM on select.py

* Updated .coveragerc
pull/59907/head
Ullrich Neiss 2021-11-18 16:06:32 +01:00 committed by GitHub
parent 5e07bc38c1
commit 3dc0b9537c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 472 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -549,6 +549,8 @@ omit =
homeassistant/components/kostal_plenticore/const.py
homeassistant/components/kostal_plenticore/helper.py
homeassistant/components/kostal_plenticore/sensor.py
homeassistant/components/kostal_plenticore/switch.py
homeassistant/components/kostal_plenticore/select.py
homeassistant/components/kwb/sensor.py
homeassistant/components/lacrosse/sensor.py
homeassistant/components/lametric/*

2
homeassistant/components/kostal_plenticore/__init__.py
View File

@ -11,7 +11,7 @@ from .helper import Plenticore
_LOGGER = logging.getLogger(__name__)
PLATFORMS = ["sensor"]
PLATFORMS = ["sensor", "switch", "select"]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:

58
homeassistant/components/kostal_plenticore/const.py
View File

@ -1,4 +1,6 @@
"""Constants for the Kostal Plenticore Solar Inverter integration."""
from collections import namedtuple
from typing import NamedTuple
from homeassistant.components.sensor import (
ATTR_STATE_CLASS,
@ -688,11 +690,59 @@ SENSOR_SETTINGS_DATA = [
{ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE, ATTR_ICON: "mdi:battery-negative"},
"format_round",
),
(
]
# Defines all entities for switches.
#
# Each entry is defined with a tuple of these values:
# - module id (str)
# - process data id (str)
# - entity name suffix (str)
# - on Value (str)
# - on Label (str)
# - off Value (str)
# - off Label (str)
SWITCH = namedtuple(
"SWITCH", "module_id data_id name is_on on_value on_label off_value off_label"
)
SWITCH_SETTINGS_DATA = [
SWITCH(
"devices:local",
"Battery:Strategy",
"Battery Strategy",
{},
"format_round",
"Battery Strategy:",
"1",
"1",
"Automatic",
"2",
"Automatic economical",
),
]
class SelectData(NamedTuple):
"""Representation of a SelectData tuple."""
module_id: str
data_id: str
name: str
options: list
is_on: str
# Defines all entities for select widgets.
#
# Each entry is defined with a tuple of these values:
# - module id (str)
# - process data id (str)
# - entity name suffix (str)
# - options
# - entity is enabled by default (bool)
SELECT_SETTINGS_DATA = [
SelectData(
"devices:local",
"battery_charge",
"Battery Charging / Usage mode",
["None", "Battery:SmartBatteryControl:Enable", "Battery:TimeControl:Enable"],
"1",
)
]

117
homeassistant/components/kostal_plenticore/helper.py
View File

@ -3,11 +3,16 @@ from __future__ import annotations
import asyncio
from collections import defaultdict
from collections.abc import Iterable
from datetime import datetime, timedelta
import logging
from aiohttp.client_exceptions import ClientError
from kostal.plenticore import PlenticoreApiClient, PlenticoreAuthenticationException
from kostal.plenticore import (
PlenticoreApiClient,
PlenticoreApiException,
PlenticoreAuthenticationException,
)
from homeassistant.const import CONF_HOST, CONF_PASSWORD, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
@ -112,6 +117,38 @@ class Plenticore:
_LOGGER.debug("Logged out from %s", self.host)
class DataUpdateCoordinatorMixin:
"""Base implementation for read and write data."""
async def async_read_data(self, module_id: str, data_id: str) -> list[str, bool]:
"""Write settings back to Plenticore."""
client = self._plenticore.client
if client is None:
return False
try:
val = await client.get_setting_values(module_id, data_id)
except PlenticoreApiException:
return False
else:
return val
async def async_write_data(self, module_id: str, value: dict[str, str]) -> bool:
"""Write settings back to Plenticore."""
client = self._plenticore.client
if client is None:
return False
try:
await client.set_setting_values(module_id, value)
except PlenticoreApiException:
return False
else:
return True
class PlenticoreUpdateCoordinator(DataUpdateCoordinator):
"""Base implementation of DataUpdateCoordinator for Plenticore data."""
@ -171,7 +208,9 @@ class ProcessDataUpdateCoordinator(PlenticoreUpdateCoordinator):
}
class SettingDataUpdateCoordinator(PlenticoreUpdateCoordinator):
class SettingDataUpdateCoordinator(
PlenticoreUpdateCoordinator, DataUpdateCoordinatorMixin
):
"""Implementation of PlenticoreUpdateCoordinator for settings data."""
async def _async_update_data(self) -> dict[str, dict[str, str]]:
@ -183,9 +222,83 @@ class SettingDataUpdateCoordinator(PlenticoreUpdateCoordinator):
_LOGGER.debug("Fetching %s for %s", self.name, self._fetch)
fetched_data = await client.get_setting_values(self._fetch)
return fetched_data
class PlenticoreSelectUpdateCoordinator(DataUpdateCoordinator):
"""Base implementation of DataUpdateCoordinator for Plenticore data."""
def __init__(
self,
hass: HomeAssistant,
logger: logging.Logger,
name: str,
update_inverval: timedelta,
plenticore: Plenticore,
) -> None:
"""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, all_options: str) -> None:
"""Start fetching the given data (module-id and entry-id)."""
self._fetch[module_id].append(data_id)
self._fetch[module_id].append(all_options)
# 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, all_options: str) -> None:
"""Stop fetching the given data (module-id and entry-id)."""
self._fetch[module_id].remove(all_options)
self._fetch[module_id].remove(data_id)
class SelectDataUpdateCoordinator(
PlenticoreSelectUpdateCoordinator, DataUpdateCoordinatorMixin
):
"""Implementation of PlenticoreUpdateCoordinator for select data."""
async def _async_update_data(self) -> dict[str, dict[str, str]]:
client = self._plenticore.client
if client is None:
return {}
_LOGGER.debug("Fetching select %s for %s", self.name, self._fetch)
fetched_data = await self.async_get_currentoption(self._fetch)
return fetched_data
async def async_get_currentoption(
self,
module_id: str | dict[str, Iterable[str]],
) -> dict[str, dict[str, str]]:
"""Get current option."""
for mid, pids in module_id.items():
all_options = pids[1]
for all_option in all_options:
if all_option != "None":
val = await self.async_read_data(mid, all_option)
for option in val.values():
if option[all_option] == "1":
fetched = {mid: {pids[0]: all_option}}
return fetched
return {mid: {pids[0]: "None"}}
class PlenticoreDataFormatter:
"""Provides method to format values of process or settings data."""

129
homeassistant/components/kostal_plenticore/select.py Normal file
View File

@ -0,0 +1,129 @@
"""Platform for Kostal Plenticore select widgets."""
from __future__ import annotations
from abc import ABC
from datetime import timedelta
import logging
from homeassistant.components.select import SelectEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import DEVICE_DEFAULT_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, SELECT_SETTINGS_DATA
from .helper import Plenticore, SelectDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities
) -> None:
"""Add kostal plenticore Select widget."""
plenticore: Plenticore = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
PlenticoreDataSelect(
hass=hass,
plenticore=plenticore,
entry_id=entry.entry_id,
platform_name=entry.title,
device_class="kostal_plenticore__battery",
module_id=select.module_id,
data_id=select.data_id,
name=select.name,
current_option="None",
options=select.options,
is_on=select.is_on,
device_info=plenticore.device_info,
unique_id=f"{entry.entry_id}_{select.module_id}",
)
for select in SELECT_SETTINGS_DATA
)
class PlenticoreDataSelect(CoordinatorEntity, SelectEntity, ABC):
"""Representation of a Plenticore Switch."""
def __init__(
self,
hass: HomeAssistant,
plenticore: Plenticore,
entry_id: str,
platform_name: str,
device_class: str | None,
module_id: str,
data_id: str,
name: str,
current_option: str | None,
options: list[str],
is_on: str,
device_info: DeviceInfo,
unique_id: str,
) -> None:
"""Create a new switch Entity for Plenticore process data."""
super().__init__(
coordinator=SelectDataUpdateCoordinator(
hass,
_LOGGER,
"Select Data",
timedelta(seconds=30),
plenticore,
)
)
self.plenticore = plenticore
self.entry_id = entry_id
self.platform_name = platform_name
self._attr_device_class = device_class
self.module_id = module_id
self.data_id = data_id
self._attr_options = options
self.all_options = options
self._attr_current_option = current_option
self._is_on = is_on
self._device_info = device_info
self._attr_name = name or DEVICE_DEFAULT_NAME
self._attr_unique_id = unique_id
@property
def available(self) -> bool:
"""Return if entity is available."""
is_available = (
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]
)
if is_available:
self._attr_current_option = self.coordinator.data[self.module_id][
self.data_id
]
return is_available
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, self.all_options
)
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, self.all_options)
await super().async_will_remove_from_hass()
async def async_select_option(self, option: str) -> None:
"""Update the current selected option."""
self._attr_current_option = option
for all_option in self._attr_options:
if all_option != "None":
await self.coordinator.async_write_data(
self.module_id, {all_option: "0"}
)
if option != "None":
await self.coordinator.async_write_data(self.module_id, {option: "1"})
self.async_write_ha_state()

171
homeassistant/components/kostal_plenticore/switch.py Normal file
View File

@ -0,0 +1,171 @@
"""Platform for Kostal Plenticore switches."""
from __future__ import annotations
from abc import ABC
from datetime import timedelta
import logging
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, SWITCH_SETTINGS_DATA
from .helper import SettingDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities
):
"""Add kostal plenticore Switch."""
plenticore = hass.data[DOMAIN][entry.entry_id]
entities = []
available_settings_data = await plenticore.client.get_settings()
settings_data_update_coordinator = SettingDataUpdateCoordinator(
hass,
_LOGGER,
"Settings Data",
timedelta(seconds=30),
plenticore,
)
for switch in SWITCH_SETTINGS_DATA:
if switch.module_id not in available_settings_data or switch.data_id not in (
setting.id for setting in available_settings_data[switch.module_id]
):
_LOGGER.debug(
"Skipping non existing setting data %s/%s",
switch.module_id,
switch.data_id,
)
continue
entities.append(
PlenticoreDataSwitch(
settings_data_update_coordinator,
entry.entry_id,
entry.title,
switch.module_id,
switch.data_id,
switch.name,
switch.is_on,
switch.on_value,
switch.on_label,
switch.off_value,
switch.off_label,
plenticore.device_info,
f"{entry.title} {switch.name}",
f"{entry.entry_id}_{switch.module_id}_{switch.data_id}",
)
)
async_add_entities(entities)
class PlenticoreDataSwitch(CoordinatorEntity, SwitchEntity, ABC):
"""Representation of a Plenticore Switch."""
def __init__(
self,
coordinator,
entry_id: str,
platform_name: str,
module_id: str,
data_id: str,
name: str,
is_on: str,
on_value: str,
on_label: str,
off_value: str,
off_label: str,
device_info: DeviceInfo,
attr_name: str,
attr_unique_id: str,
):
"""Create a new switch 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._last_run_success: bool | None = None
self._name = name
self._is_on = is_on
self._attr_name = attr_name
self.on_value = on_value
self.on_label = on_label
self.off_value = off_value
self.off_label = off_label
self._attr_unique_id = attr_unique_id
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()
async def async_turn_on(self, **kwargs) -> None:
"""Turn device on."""
if await self.coordinator.async_write_data(
self.module_id, {self.data_id: self.on_value}
):
self._last_run_success = True
self.coordinator.name = f"{self.platform_name} {self._name} {self.on_label}"
await self.coordinator.async_request_refresh()
else:
self._last_run_success = False
async def async_turn_off(self, **kwargs) -> None:
"""Turn device off."""
if await self.coordinator.async_write_data(
self.module_id, {self.data_id: self.off_value}
):
self._last_run_success = True
self.coordinator.name = (
f"{self.platform_name} {self._name} {self.off_label}"
)
await self.coordinator.async_request_refresh()
else:
self._last_run_success = False
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return self._device_info
@property
def is_on(self) -> bool:
"""Return true if device is on."""
if self.coordinator.data[self.module_id][self.data_id] == self._is_on:
self.coordinator.name = f"{self.platform_name} {self._name} {self.on_label}"
else:
self.coordinator.name = (
f"{self.platform_name} {self._name} {self.off_label}"
)
return bool(self.coordinator.data[self.module_id][self.data_id] == self._is_on)
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
return {"last_run_success": self._last_run_success}