From 3dc0b9537ca5bf8c9e11c521e2af5d9d67e1c4f6 Mon Sep 17 00:00:00 2001 From: Ullrich Neiss Date: Thu, 18 Nov 2021 16:06:32 +0100 Subject: [PATCH] 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 --- .coveragerc | 2 + .../components/kostal_plenticore/__init__.py | 2 +- .../components/kostal_plenticore/const.py | 58 +++++- .../components/kostal_plenticore/helper.py | 117 +++++++++++- .../components/kostal_plenticore/select.py | 129 +++++++++++++ .../components/kostal_plenticore/switch.py | 171 ++++++++++++++++++ 6 files changed, 472 insertions(+), 7 deletions(-) create mode 100644 homeassistant/components/kostal_plenticore/select.py create mode 100644 homeassistant/components/kostal_plenticore/switch.py diff --git a/.coveragerc b/.coveragerc index bbe00155b2f..2b247c2c923 100644 --- a/.coveragerc +++ b/.coveragerc @@ -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/* diff --git a/homeassistant/components/kostal_plenticore/__init__.py b/homeassistant/components/kostal_plenticore/__init__.py index f00e6ee1327..8e2beb73cc2 100644 --- a/homeassistant/components/kostal_plenticore/__init__.py +++ b/homeassistant/components/kostal_plenticore/__init__.py @@ -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: diff --git a/homeassistant/components/kostal_plenticore/const.py b/homeassistant/components/kostal_plenticore/const.py index 68c2baffbdb..b025738a7b8 100644 --- a/homeassistant/components/kostal_plenticore/const.py +++ b/homeassistant/components/kostal_plenticore/const.py @@ -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", + ) +] diff --git a/homeassistant/components/kostal_plenticore/helper.py b/homeassistant/components/kostal_plenticore/helper.py index 32dfc9b2fd9..264d7e90efb 100644 --- a/homeassistant/components/kostal_plenticore/helper.py +++ b/homeassistant/components/kostal_plenticore/helper.py @@ -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.""" diff --git a/homeassistant/components/kostal_plenticore/select.py b/homeassistant/components/kostal_plenticore/select.py new file mode 100644 index 00000000000..1f9d11dc334 --- /dev/null +++ b/homeassistant/components/kostal_plenticore/select.py @@ -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() diff --git a/homeassistant/components/kostal_plenticore/switch.py b/homeassistant/components/kostal_plenticore/switch.py new file mode 100644 index 00000000000..9598e12aac0 --- /dev/null +++ b/homeassistant/components/kostal_plenticore/switch.py @@ -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}