diff --git a/custom_components/kostal/__init__.py b/custom_components/kostal/__init__.py index 7d6cf2e..1d90681 100755 --- a/custom_components/kostal/__init__.py +++ b/custom_components/kostal/__init__.py @@ -1,37 +1,36 @@ """The Kostal PIKO inverter sensor integration.""" - import logging - -from .piko_holder import PikoHolder +from typing import Any from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry -from homeassistant.helpers.typing import HomeAssistantType -from homeassistant.helpers.dispatcher import async_dispatcher_send, dispatcher_send - -from .const import DEFAULT_NAME, DOMAIN, SENSOR_TYPES -from .configuration_schema import CONFIG_SCHEMA, CONFIG_SCHEMA_ROOT - from homeassistant.const import ( - CONF_USERNAME, - CONF_PASSWORD, CONF_HOST, CONF_MONITORED_CONDITIONS, - EVENT_HOMEASSISTANT_STOP + CONF_PASSWORD, + CONF_USERNAME, + EVENT_HOMEASSISTANT_STOP, ) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.dispatcher import async_dispatcher_send +from homeassistant.helpers.typing import ConfigType + +from .configuration_schema import CONFIG_SCHEMA_ROOT, SENSOR_TYPE_KEY, UserInputType +from .const import DOMAIN +from .piko_holder import PikoHolder _LOGGER = logging.getLogger(__name__) -__version__ = "1.2.0" +__version__ = "1.3.0" VERSION = __version__ -async def async_setup(hass, config): +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up this integration using yaml.""" _LOGGER.info("Setup kostal, %s", __version__) if DOMAIN not in config: return True - data = dict(config.get(DOMAIN)) + data = config.get(DOMAIN, {}) hass.data["yaml_kostal"] = data hass.async_create_task( @@ -42,23 +41,22 @@ async def async_setup(hass, config): return True -async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry): - """Setup KostalPiko component""" +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Create KostalPiko component for config entry.""" _LOGGER.info("Starting kostal, %s", __version__) - if not DOMAIN in hass.data: + if DOMAIN not in hass.data: hass.data[DOMAIN] = {} if entry.source == "import": if entry.options: # config.yaml data = entry.options.copy() + elif "yaml_kostal" in hass.data: + data = hass.data["yaml_kostal"] else: - if "yaml_kostal" in hass.data: - data = hass.data["yaml_kostal"] - else: - data = {} - await hass.config_entries.async_remove(entry.entry_id) + data = {} + await hass.config_entries.async_remove(entry.entry_id) else: data = entry.data.copy() data.update(entry.options) @@ -69,18 +67,21 @@ async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry): return True -async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry): +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload a config entry.""" - instance = hass.data[DOMAIN][entry.entry_id] + instance: KostalInstance = hass.data[DOMAIN][entry.entry_id] await instance.stop() await instance.clean() return True -class KostalInstance(): - """Config instance of Kostal""" +class KostalInstance: + """Config instance of Kostal.""" - def __init__(self, hass: HomeAssistantType, entry: ConfigEntry, conf): + def __init__( + self, hass: HomeAssistant, entry: ConfigEntry, conf: UserInputType + ) -> None: + """Create a new Kostal instance.""" self.hass = hass self.config_entry = entry self.entry_id = self.config_entry.entry_id @@ -91,24 +92,34 @@ def __init__(self, hass: HomeAssistantType, entry: ConfigEntry, conf): hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self.stop) - hass.loop.create_task( - self.start_up() - ) + hass.loop.create_task(self.start_up()) - async def start_up(self): + async def start_up(self) -> None: + """Start the KOSTAL instance.""" await self.hass.async_add_executor_job(self.piko.update) self.add_sensors(self.conf[CONF_MONITORED_CONDITIONS], self.piko) - async def stop(self, _=None): - """Stop Kostal.""" + async def stop(self, _: Any = None) -> None: + """Stop the KOSTAL instance.""" _LOGGER.info("Shutting down Kostal") - def add_sensors(self, sensors, piko: PikoHolder): + def add_sensors(self, sensors: list[SENSOR_TYPE_KEY], piko: PikoHolder) -> None: + """Add sensors to HASS.""" self.hass.add_job(self._asyncadd_sensors(sensors, piko)) - async def _asyncadd_sensors(self, sensors, piko: PikoHolder): - await self.hass.config_entries.async_forward_entry_setup(self.config_entry, "sensor") - async_dispatcher_send(self.hass, "kostal_init_sensors", sensors, piko) + async def _asyncadd_sensors( + self, sensors: list[SENSOR_TYPE_KEY], piko: PikoHolder + ) -> None: + """Add async sensors to HASS.""" + await self.hass.config_entries.async_forward_entry_setup( + self.config_entry, "sensor" + ) + async_dispatcher_send( + self.hass, + "kostal_init_sensors_" + self.config_entry.entry_id, + sensors, + piko, + ) - async def clean(self): - pass + async def clean(self) -> None: + """Cleanup. Not implemented.""" diff --git a/custom_components/kostal/config_flow.py b/custom_components/kostal/config_flow.py index 386ed94..c6d0d74 100644 --- a/custom_components/kostal/config_flow.py +++ b/custom_components/kostal/config_flow.py @@ -1,44 +1,47 @@ """Adds config flow for Kostal.""" import logging -import voluptuous as vol +from typing import Any from kostalpiko.kostalpiko import Piko - -from requests.exceptions import HTTPError, ConnectTimeout +from requests.exceptions import ConnectTimeout, HTTPError +import voluptuous as vol from homeassistant import config_entries, exceptions -import homeassistant.helpers.config_validation as cv - from homeassistant.const import ( - CONF_NAME, CONF_HOST, - CONF_USERNAME, - CONF_PASSWORD, CONF_MONITORED_CONDITIONS, + CONF_NAME, + CONF_PASSWORD, + CONF_USERNAME, ) from homeassistant.core import HomeAssistant, callback +from homeassistant.data_entry_flow import FlowResult +import homeassistant.helpers.config_validation as cv from homeassistant.util import slugify -from .const import DOMAIN, DEFAULT_NAME, SENSOR_TYPES +from .configuration_schema import SENSOR_TYPE_KEY, UserInputType +from .const import DEFAULT_NAME, DOMAIN, SENSOR_TYPES SUPPORTED_SENSOR_TYPES = list(SENSOR_TYPES) -DEFAULT_MONITORED_CONDITIONS = [ - "current_power", - "total_energy", - "daily_energy", - "status", +DEFAULT_MONITORED_CONDITIONS: list[SENSOR_TYPE_KEY] = [ + SENSOR_TYPE_KEY.SENSOR_CURRENT_POWER, + SENSOR_TYPE_KEY.SENSOR_TOTAL_ENERGY, + SENSOR_TYPE_KEY.SENSOR_DAILY_ENERGY, + SENSOR_TYPE_KEY.SENSOR_STATUS, ] _LOGGER = logging.getLogger(__name__) + @callback -def kostal_entries(hass: HomeAssistant): +def kostal_entries(hass: HomeAssistant) -> set[str]: """Return the hosts for the domain.""" - return set( + return { (entry.data[CONF_HOST]) for entry in hass.config_entries.async_entries(DOMAIN) - ) + } + class KostalConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): """Handle a config flow for Kostal piko.""" @@ -47,33 +50,33 @@ class KostalConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL - def __init__(self): + def __init__(self) -> None: """Initialize.""" - self._errors = {} + self._errors: dict[str, str] = {} - def _host_in_configuration_exists(self, host) -> bool: + def _host_in_configuration_exists(self, host: str) -> bool: """Return True if site_id exists in configuration.""" if host in kostal_entries(self.hass): return True return False - def _check_host(self, host, username, password) -> bool: + def _check_host(self, host: str, username: str, password: str) -> bool: """Check if we can connect to the kostal inverter.""" piko = Piko(host, username, password) try: - response = piko._get_info() + piko._get_info() # pylint: disable=protected-access except (ConnectTimeout, HTTPError): self._errors[CONF_HOST] = "could_not_connect" return False return True - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Handle the initial step.""" - errors = {} - + _user_input: UserInputType if user_input is not None: - name = slugify(user_input.get(CONF_NAME, DEFAULT_NAME)) if self._host_in_configuration_exists(user_input[CONF_HOST]): self._errors[CONF_HOST] = "host_exists" @@ -95,13 +98,15 @@ async def async_step_user(self, user_input=None): CONF_MONITORED_CONDITIONS: conditions, }, ) + _user_input = user_input # type: ignore else: - user_input = {} - user_input[CONF_NAME] = DEFAULT_NAME - user_input[CONF_HOST] = "http://" - user_input[CONF_USERNAME] = "" - user_input[CONF_PASSWORD] = "" - user_input[CONF_MONITORED_CONDITIONS] = DEFAULT_MONITORED_CONDITIONS + _user_input = { + CONF_NAME: DEFAULT_NAME, + CONF_HOST: "http://", + CONF_USERNAME: "", + CONF_PASSWORD: "", + CONF_MONITORED_CONDITIONS: DEFAULT_MONITORED_CONDITIONS, + } default_monitored_conditions = ( [] if self._async_current_entries() else DEFAULT_MONITORED_CONDITIONS @@ -109,15 +114,16 @@ async def async_step_user(self, user_input=None): setup_schema = vol.Schema( { vol.Required( - CONF_NAME, default=user_input.get(CONF_NAME, DEFAULT_NAME) + CONF_NAME, default=_user_input.get(CONF_NAME, DEFAULT_NAME) ): str, - vol.Required(CONF_HOST, default=user_input[CONF_HOST]): str, + vol.Required(CONF_HOST, default=_user_input[CONF_HOST]): str, # type: ignore vol.Required( CONF_USERNAME, description={"suggested_value": "pvserver"} ): str, - vol.Required(CONF_PASSWORD, default=user_input[CONF_PASSWORD]): str, + vol.Required(CONF_PASSWORD, default=_user_input[CONF_PASSWORD]): str, # type: ignore vol.Required( - CONF_MONITORED_CONDITIONS, default=default_monitored_conditions + CONF_MONITORED_CONDITIONS, + default=default_monitored_conditions, # type: ignore ): cv.multi_select(SUPPORTED_SENSOR_TYPES), } ) @@ -126,9 +132,13 @@ async def async_step_user(self, user_input=None): step_id="user", data_schema=setup_schema, errors=self._errors ) - async def async_step_import(self, user_input=None): + async def async_step_import( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Import a config entry.""" - if self._host_in_configuration_exists(user_input[CONF_HOST]): + if user_input is not None and self._host_in_configuration_exists( + user_input[CONF_HOST] + ): return self.async_abort(reason="host_exists") return await self.async_step_user(user_input) diff --git a/custom_components/kostal/configuration_schema.py b/custom_components/kostal/configuration_schema.py index 6e99ee2..156f904 100644 --- a/custom_components/kostal/configuration_schema.py +++ b/custom_components/kostal/configuration_schema.py @@ -1,29 +1,40 @@ """Kostal Piko Configuration Schemas.""" +from typing import TypedDict import voluptuous as vol from homeassistant.const import ( - CONF_NAME, CONF_HOST, - CONF_USERNAME, - CONF_PASSWORD, CONF_MONITORED_CONDITIONS, + CONF_NAME, + CONF_PASSWORD, + CONF_USERNAME, ) - import homeassistant.helpers.config_validation as cv -from .const import DEFAULT_NAME, DOMAIN, SENSOR_TYPES - -CONFIG_SCHEMA_ROOT = vol.Schema({ - vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, - vol.Required(CONF_USERNAME): cv.string, - vol.Required(CONF_PASSWORD): cv.string, - vol.Required(CONF_HOST): cv.string, - vol.Required(CONF_MONITORED_CONDITIONS): vol.All( - cv.ensure_list, [vol.In(list(SENSOR_TYPES))] - ), -}) - -CONFIG_SCHEMA = vol.Schema({ - DOMAIN: CONFIG_SCHEMA_ROOT -}, extra=vol.ALLOW_EXTRA) \ No newline at end of file +from .const import DEFAULT_NAME, DOMAIN, SENSOR_TYPE_KEY, SENSOR_TYPES + + +class UserInputType(TypedDict): + """Data entered by user.""" + + name: str + host: str + username: str + password: str + monitored_conditions: list[SENSOR_TYPE_KEY] + + +CONFIG_SCHEMA_ROOT = vol.Schema( + { + vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, # type: ignore + vol.Required(CONF_USERNAME): cv.string, + vol.Required(CONF_PASSWORD): cv.string, + vol.Required(CONF_HOST): cv.string, + vol.Required(CONF_MONITORED_CONDITIONS): vol.All( + cv.ensure_list, [vol.In(list(SENSOR_TYPES))] + ), + } +) + +CONFIG_SCHEMA = vol.Schema({DOMAIN: CONFIG_SCHEMA_ROOT}, extra=vol.ALLOW_EXTRA) diff --git a/custom_components/kostal/const.py b/custom_components/kostal/const.py index a8ba91d..14016dc 100755 --- a/custom_components/kostal/const.py +++ b/custom_components/kostal/const.py @@ -1,19 +1,14 @@ """Constants for the Kostal piko integration.""" from datetime import timedelta +from enum import StrEnum +from typing import Final, TypedDict from homeassistant.const import ( - POWER_WATT, - ENERGY_KILO_WATT_HOUR, - ) - -try: - from homeassistant.const import ( - ELECTRIC_POTENTIAL_VOLT, - ELECTRIC_CURRENT_AMPERE, - ) -except ImportError: - from homeassistant.const import VOLT as ELECTRIC_POTENTIAL_VOLT - from homeassistant.const import ELECTRICAL_CURRENT_AMPERE as ELECTRIC_CURRENT_AMPERE + UnitOfElectricCurrent, + UnitOfElectricPotential, + UnitOfEnergy, + UnitOfPower, +) DOMAIN = "kostal" @@ -21,25 +16,142 @@ MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=30) -SENSOR_TYPES = { - "solar_generator_power": ["Solar generator power", POWER_WATT, "mdi:solar-power"], - "consumption_phase_1": ["Consumption phase 1", POWER_WATT, "mdi:power-socket-eu"], - "consumption_phase_2": ["Consumption phase 2", POWER_WATT, "mdi:power-socket-eu"], - "consumption_phase_3": ["Consumption phase 3", POWER_WATT, "mdi:power-socket-eu"], - "current_power": ["Current power", POWER_WATT, "mdi:solar-power"], - "total_energy": ["Total energy", ENERGY_KILO_WATT_HOUR, "mdi:solar-power"], - "daily_energy": ["Daily energy", ENERGY_KILO_WATT_HOUR, "mdi:solar-power"], - "string1_voltage": ["String 1 voltage", ELECTRIC_POTENTIAL_VOLT, "mdi:current-ac"], - "string1_current": ["String 1 current", ELECTRIC_CURRENT_AMPERE, "mdi:flash"], - "string2_voltage": ["String 2 voltage", ELECTRIC_POTENTIAL_VOLT, "mdi:current-ac"], - "string2_current": ["String 2 current", ELECTRIC_CURRENT_AMPERE, "mdi:flash"], - "string3_voltage": ["String 3 voltage", ELECTRIC_POTENTIAL_VOLT, "mdi:current-ac"], - "string3_current": ["String 3 current", ELECTRIC_CURRENT_AMPERE, "mdi:flash"], - "l1_voltage": ["L1 voltage", ELECTRIC_POTENTIAL_VOLT, "mdi:current-ac"], - "l1_power": ["L1 power", POWER_WATT, "mdi:power-plug"], - "l2_voltage": ["L2 voltage", ELECTRIC_POTENTIAL_VOLT, "mdi:current-ac"], - "l2_power": ["L2 power", POWER_WATT, "mdi:power-plug"], - "l3_voltage": ["L3 voltage", ELECTRIC_POTENTIAL_VOLT, "mdi:current-ac"], - "l3_power": ["L3 power", POWER_WATT, "mdi:power-plug"], - "status": ["Status", None, "mdi:solar-power"], -} \ No newline at end of file + +class SensorDefinition(TypedDict): + """Description of a sensor entity.""" + + name: str + unit: str | None + icon: str + + +class SENSOR_TYPE_KEY(StrEnum): + """Sensor type keys.""" + + SENSOR_SOLAR_GENERATOR_POWER: Final = "solar_generator_power" + SENSOR_CONSUMPTION_PHASE_1: Final = "consumption_phase_1" + SENSOR_CONSUMPTION_PHASE_2: Final = "consumption_phase_2" + SENSOR_CONSUMPTION_PHASE_3: Final = "consumption_phase_3" + SENSOR_CURRENT_POWER: Final = "current_power" + SENSOR_DAILY_ENERGY: Final = "daily_energy" + SENSOR_TOTAL_ENERGY: Final = "total_energy" + SENSOR_STRING1_VOLTAGE: Final = "string1_voltage" + SENSOR_STRING2_VOLTAGE: Final = "string2_voltage" + SENSOR_STRING3_VOLTAGE: Final = "string3_voltage" + SENSOR_L1_VOLTAGE: Final = "l1_voltage" + SENSOR_L2_VOLTAGE: Final = "l2_voltage" + SENSOR_L3_VOLTAGE: Final = "l3_voltage" + SENSOR_STRING1_CURRENT: Final = "string1_current" + SENSOR_STRING2_CURRENT: Final = "string2_current" + SENSOR_STRING3_CURRENT: Final = "string3_current" + SENSOR_L1_CURRENT: Final = "l1_current" + SENSOR_L2_CURRENT: Final = "l2_current" + SENSOR_L3_CURRENT: Final = "l3_current" + SENSOR_L1_POWER: Final = "l1_power" + SENSOR_L2_POWER: Final = "l2_power" + SENSOR_L3_POWER: Final = "l3_power" + SENSOR_STATUS: Final = "status" + + +SENSOR_TYPES: dict[SENSOR_TYPE_KEY, SensorDefinition] = { + SENSOR_TYPE_KEY.SENSOR_SOLAR_GENERATOR_POWER: { + "name": "Solar generator power", + "unit": UnitOfPower.WATT, + "icon": "mdi:solar-power", + }, + SENSOR_TYPE_KEY.SENSOR_CONSUMPTION_PHASE_1: { + "name": "Consumption phase 1", + "unit": UnitOfPower.WATT, + "icon": "mdi:power-socket-eu", + }, + SENSOR_TYPE_KEY.SENSOR_CONSUMPTION_PHASE_2: { + "name": "Consumption phase 2", + "unit": UnitOfPower.WATT, + "icon": "mdi:power-socket-eu", + }, + SENSOR_TYPE_KEY.SENSOR_CONSUMPTION_PHASE_3: { + "name": "Consumption phase 3", + "unit": UnitOfPower.WATT, + "icon": "mdi:power-socket-eu", + }, + SENSOR_TYPE_KEY.SENSOR_CURRENT_POWER: { + "name": "Current power", + "unit": UnitOfPower.WATT, + "icon": "mdi:solar-power", + }, + SENSOR_TYPE_KEY.SENSOR_TOTAL_ENERGY: { + "name": "Total energy", + "unit": UnitOfEnergy.KILO_WATT_HOUR, + "icon": "mdi:solar-power", + }, + SENSOR_TYPE_KEY.SENSOR_DAILY_ENERGY: { + "name": "Daily energy", + "unit": UnitOfEnergy.KILO_WATT_HOUR, + "icon": "mdi:solar-power", + }, + SENSOR_TYPE_KEY.SENSOR_STRING1_VOLTAGE: { + "name": "String 1 voltage", + "unit": UnitOfElectricPotential.VOLT, + "icon": "mdi:current-ac", + }, + SENSOR_TYPE_KEY.SENSOR_STRING1_CURRENT: { + "name": "String 1 current", + "unit": UnitOfElectricCurrent.AMPERE, + "icon": "mdi:flash", + }, + SENSOR_TYPE_KEY.SENSOR_STRING2_VOLTAGE: { + "name": "String 2 voltage", + "unit": UnitOfElectricPotential.VOLT, + "icon": "mdi:current-ac", + }, + SENSOR_TYPE_KEY.SENSOR_STRING2_CURRENT: { + "name": "String 2 current", + "unit": UnitOfElectricCurrent.AMPERE, + "icon": "mdi:flash", + }, + SENSOR_TYPE_KEY.SENSOR_STRING3_VOLTAGE: { + "name": "String 3 voltage", + "unit": UnitOfElectricPotential.VOLT, + "icon": "mdi:current-ac", + }, + SENSOR_TYPE_KEY.SENSOR_STRING3_CURRENT: { + "name": "String 3 current", + "unit": UnitOfElectricCurrent.AMPERE, + "icon": "mdi:flash", + }, + SENSOR_TYPE_KEY.SENSOR_L1_VOLTAGE: { + "name": "L1 voltage", + "unit": UnitOfElectricPotential.VOLT, + "icon": "mdi:current-ac", + }, + SENSOR_TYPE_KEY.SENSOR_L1_POWER: { + "name": "L1 power", + "unit": UnitOfPower.WATT, + "icon": "mdi:power-plug", + }, + SENSOR_TYPE_KEY.SENSOR_L2_VOLTAGE: { + "name": "L2 voltage", + "unit": UnitOfElectricPotential.VOLT, + "icon": "mdi:current-ac", + }, + SENSOR_TYPE_KEY.SENSOR_L2_POWER: { + "name": "L2 power", + "unit": UnitOfPower.WATT, + "icon": "mdi:power-plug", + }, + SENSOR_TYPE_KEY.SENSOR_L3_VOLTAGE: { + "name": "L3 voltage", + "unit": UnitOfElectricPotential.VOLT, + "icon": "mdi:current-ac", + }, + SENSOR_TYPE_KEY.SENSOR_L3_POWER: { + "name": "L3 power", + "unit": UnitOfPower.WATT, + "icon": "mdi:power-plug", + }, + SENSOR_TYPE_KEY.SENSOR_STATUS: { + "name": "Status", + "unit": None, + "icon": "mdi:solar-power", + }, +} diff --git a/custom_components/kostal/manifest.json b/custom_components/kostal/manifest.json index ffa24fb..329440d 100644 --- a/custom_components/kostal/manifest.json +++ b/custom_components/kostal/manifest.json @@ -1,10 +1,14 @@ { - "codeowners": ["@rcasula"], + "codeowners": [ + "@rcasula" + ], "config_flow": true, "dependencies": [], "documentation": "https://github.com/rcasula/kostalpiko-homeassistant", "domain": "kostal", "name": "Kostal Piko", - "requirements": ["kostalpiko>=0.6"], - "version": "1.2.0" -} + "requirements": [ + "kostalpiko>=0.6" + ], + "version": "1.3.0" +} \ No newline at end of file diff --git a/custom_components/kostal/piko_holder.py b/custom_components/kostal/piko_holder.py index dcaf18c..d085923 100644 --- a/custom_components/kostal/piko_holder.py +++ b/custom_components/kostal/piko_holder.py @@ -1,18 +1,29 @@ -from logging import lastResort +"""Wrapper for Piko object.""" +import time + from kostalpiko.kostalpiko import Piko -import time -class PikoHolder(Piko): - last_update = 0 +class PikoHolder(Piko): # type: ignore + """Wrapper for Piko object.""" + + last_update = 0.0 update_running = False - def __init__(self, host=None, username="pvserver", password="pvwr") -> None: - super().__init__(host, username, password) - - def update(self): + + # def __init__( + # self, + # host: str | None = None, + # username: str = "pvserver", + # password: str = "pvwr", + # ) -> None: + # """Create a new PIKO instance to update entities.""" + # super().__init__(host, username, password) + + def update(self) -> None: + """Pull values from PIKO.""" if not self.update_running: if time.time() - self.last_update > 30.0: - self.update_running=True - self.last_update=time.time() + self.update_running = True + self.last_update = time.time() super().update() - self.update_running=False + self.update_running = False diff --git a/custom_components/kostal/sensor.py b/custom_components/kostal/sensor.py index 1f78b95..855c77f 100755 --- a/custom_components/kostal/sensor.py +++ b/custom_components/kostal/sensor.py @@ -1,167 +1,160 @@ """Support for Kostal PIKO Photvoltaic (PV) inverter.""" - -import logging, time -from .piko_holder import PikoHolder - -from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry -from homeassistant.helpers.typing import HomeAssistantType -from homeassistant.helpers.dispatcher import async_dispatcher_connect - -from homeassistant.helpers.entity import Entity -from homeassistant.util import Throttle, dt -from homeassistant.const import ( - CONF_USERNAME, - CONF_PASSWORD, - CONF_HOST, - CONF_MONITORED_CONDITIONS, - DEVICE_CLASS_ENERGY, - ENERGY_KILO_WATT_HOUR, -) +import logging +from typing import TypedDict from homeassistant.components.sensor import ( - STATE_CLASS_MEASUREMENT, + SensorDeviceClass, SensorEntity, + SensorStateClass, ) -try: - from homeassistant.components.sensor import STATE_CLASS_TOTAL_INCREASING -except ImportError: - from homeassistant.components.sensor import STATE_CLASS_MEASUREMENT as STATE_CLASS_TOTAL_INCREASING +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import UnitOfEnergy +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.util import Throttle -from .const import SENSOR_TYPES, MIN_TIME_BETWEEN_UPDATES, DOMAIN +from .configuration_schema import SENSOR_TYPE_KEY +from .const import DOMAIN, MIN_TIME_BETWEEN_UPDATES, SENSOR_TYPES +from .piko_holder import PikoHolder _LOGGER = logging.getLogger(__name__) -async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry, async_add_entities): +class KostalDeviceInfo(TypedDict): + """Device info from Kostal.""" + + serial: str + model: str | None + + +async def async_setup_entry( + hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback +) -> None: """Set up the sensor dynamically.""" _LOGGER.info("Setting up kostal piko sensor") - async def async_add_sensors(sensors, piko: PikoHolder): + + async def async_add_sensors( + sensors: list[SENSOR_TYPE_KEY], piko: PikoHolder + ) -> None: """Add a sensor.""" - info = await hass.async_add_executor_job(piko._get_info) + _info: list[str] | None = await hass.async_add_executor_job(piko._get_info) # pylint: disable=protected-access + info: KostalDeviceInfo = { + "serial": entry.data.get("host", "UNKNOWN"), + "model": None, + } + if _info is not None: + info = {"serial": _info[0], "model": _info[1]} _sensors = [] for sensor in sensors: _sensors.append(PikoSensor(hass, piko, sensor, info, entry.title)) - + async_add_entities(_sensors) # async_add_entities([ # PikoSensor(piko, type, entry.title) # ]) async_dispatcher_connect( - hass, - "kostal_init_sensors", - async_add_sensors + hass, "kostal_init_sensors_" + entry.entry_id, async_add_sensors ) class PikoSensor(SensorEntity): """Representation of a Piko inverter value.""" - def __init__(self, hass: HomeAssistantType, piko: PikoHolder, sensor_type, info={None,None}, name=None): + def __init__( + self, + hass: HomeAssistant, + piko: PikoHolder, + sensor_type: SENSOR_TYPE_KEY, + info: KostalDeviceInfo, + name: str | None = None, + ) -> None: """Initialize the sensor.""" _LOGGER.debug("Initializing PikoSensor: %s", sensor_type) - self._sensor = SENSOR_TYPES[sensor_type][0] + self._sensor = SENSOR_TYPES[sensor_type]["name"] self._name = name self.hass = hass self.type = sensor_type self.piko = piko - self._state = None - self._unit_of_measurement = SENSOR_TYPES[self.type][1] - self._icon = SENSOR_TYPES[self.type][2] - self.serial_number = info[0] - self.model = info[1] - if self._unit_of_measurement == ENERGY_KILO_WATT_HOUR: - self._attr_device_class = DEVICE_CLASS_ENERGY - self._attr_state_class = STATE_CLASS_TOTAL_INCREASING + # self._state: float | int | str | None = None + self._attr_native_unit_of_measurement = SENSOR_TYPES[self.type]["unit"] + self._attr_icon = SENSOR_TYPES[self.type]["icon"] + self._attr_name = f"{name} {self._sensor}" + self._attr_unique_id = f"{info['serial']} {self._sensor}" + self.kostal_info = info + if SENSOR_TYPES[self.type]["unit"] == UnitOfEnergy.KILO_WATT_HOUR: + self._attr_device_class = SensorDeviceClass.ENERGY + self._attr_state_class = SensorStateClass.TOTAL_INCREASING + + # @property + # def state(self) -> str | float | int | None: + # """Return the state of the device.""" + # return self._state @property - def name(self): - """Return the name of the sensor.""" - return "{} {}".format(self._name, self._sensor) - - @property - def state(self): - """Return the state of the device.""" - return self._state - - @property - def unit_of_measurement(self): - """Return the unit of measurement this sensor expresses itself in.""" - return self._unit_of_measurement - - @property - def icon(self): - """Return icon.""" - return self._icon - - @property - def unique_id(self): - """Return unique id based on device serial and variable.""" - return "{} {}".format(self.serial_number, self._sensor) - - @property - def device_info(self): + def device_info(self) -> DeviceInfo: """Return information about the device.""" return { - "identifiers": {(DOMAIN, self.serial_number)}, + "identifiers": {(DOMAIN, self.kostal_info["serial"])}, "name": self._name, "manufacturer": "Kostal", - "model": self.model, + "model": self.kostal_info["model"], } - @Throttle(MIN_TIME_BETWEEN_UPDATES) - async def async_update(self): + async def async_update(self) -> None: + """Update kostal entity async.""" await self.hass.async_add_executor_job(self._update) - - def _update(self): - """Update data.""" + def _update(self) -> None: + """Update kostal entity.""" self.piko.update() data = self.piko.data ba_data = self.piko.ba_data - + value: int | float | str | None = None if data is not None: if self.type == "current_power": - self._state = data.get_current_power() + value = data.get_current_power() elif self.type == "total_energy": - self._state = data.get_total_energy() + value = data.get_total_energy() elif self.type == "daily_energy": - self._state = data.get_daily_energy() + value = data.get_daily_energy() elif self.type == "string1_voltage": - self._state = data.get_string1_voltage() + value = data.get_string1_voltage() elif self.type == "string1_current": - self._state = data.get_string1_current() + value = data.get_string1_current() elif self.type == "string2_voltage": - self._state = data.get_string2_voltage() + value = data.get_string2_voltage() elif self.type == "string2_current": - self._state = data.get_string2_current() + value = data.get_string2_current() elif self.type == "string3_voltage": - self._state = data.get_string3_voltage() + value = data.get_string3_voltage() elif self.type == "string3_current": - self._state = data.get_string3_current() + value = data.get_string3_current() elif self.type == "l1_voltage": - self._state = data.get_l1_voltage() + value = data.get_l1_voltage() elif self.type == "l1_power": - self._state = data.get_l1_power() + value = data.get_l1_power() elif self.type == "l2_voltage": - self._state = data.get_l2_voltage() + value = data.get_l2_voltage() elif self.type == "l2_power": - self._state = data.get_l2_power() + value = data.get_l2_power() elif self.type == "l3_voltage": - self._state = data.get_l3_voltage() + value = data.get_l3_voltage() elif self.type == "l3_power": - self._state = data.get_l3_power() + value = data.get_l3_power() elif self.type == "status": - self._state = data.get_piko_status() + value = data.get_piko_status() if ba_data is not None: if self.type == "solar_generator_power": - self._state = ba_data.get_solar_generator_power() or "No BA sensor installed" + value = ba_data.get_solar_generator_power() or "No BA sensor installed" elif self.type == "consumption_phase_1": - self._state = ba_data.get_consumption_phase_1() or "No BA sensor installed" + value = ba_data.get_consumption_phase_1() or "No BA sensor installed" elif self.type == "consumption_phase_2": - self._state = ba_data.get_consumption_phase_2() or "No BA sensor installed" + value = ba_data.get_consumption_phase_2() or "No BA sensor installed" elif self.type == "consumption_phase_3": - self._state = ba_data.get_consumption_phase_3() or "No BA sensor installed" - - _LOGGER.debug("END - Type: {} - {}".format(self.type, self._state)) + value = ba_data.get_consumption_phase_3() or "No BA sensor installed" + self._attr_native_value = value + _LOGGER.debug("END - Type: %s - %s", self.type, value)