Fix #8 and #23, update deprecated units, add more typings, apply HASS style-rules

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."""

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,29 +98,32 @@ 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
         )
         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)
 

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)
+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)