Improve Elk-M1 Control typing (#69924)

* Add types to __init__.py

* Fixup typing.

* Fix type error.

* Bump lib to fix login error.

Co-authored-by: Shay Levy <levyshay1@gmail.com>

.strict-typing

@@ -81,6 +81,7 @@ homeassistant.components.dsmr.*
 homeassistant.components.dunehd.*
 homeassistant.components.efergy.*
 homeassistant.components.elgato.*
+homeassistant.components.elkm1.__init__
 homeassistant.components.esphome.*
 homeassistant.components.energy.*
 homeassistant.components.evil_genius_labs.*

homeassistant/components/elkm1/__init__.py

@@ -5,11 +5,12 @@ import asyncio
 import logging
 import re
 from types import MappingProxyType
-from typing import Any
+from typing import Any, cast
 from urllib.parse import urlparse
 
 import async_timeout
-import elkm1_lib as elkm1
+from elkm1_lib.elements import Element
+from elkm1_lib.elk import Elk
 import voluptuous as vol
 
 from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
@@ -22,6 +23,7 @@ from homeassistant.const import (
     CONF_PREFIX,
     CONF_TEMPERATURE_UNIT,
     CONF_USERNAME,
+    CONF_ZONE,
     TEMP_CELSIUS,
     TEMP_FAHRENHEIT,
     Platform,
@@ -50,7 +52,6 @@ from .const import (
     CONF_SETTING,
     CONF_TASK,
     CONF_THERMOSTAT,
-    CONF_ZONE,
     DISCOVER_SCAN_TIMEOUT,
     DISCOVERY_INTERVAL,
     DOMAIN,
@@ -92,7 +93,7 @@ SET_TIME_SERVICE_SCHEMA = vol.Schema(
 )
 
 
-def _host_validator(config):
+def _host_validator(config: dict[str, str]) -> dict[str, str]:
     """Validate that a host is properly configured."""
     if config[CONF_HOST].startswith("elks://"):
         if CONF_USERNAME not in config or CONF_PASSWORD not in config:
@@ -104,14 +105,14 @@ def _host_validator(config):
     return config
 
 
-def _elk_range_validator(rng):
-    def _housecode_to_int(val):
+def _elk_range_validator(rng: str) -> tuple[int, int]:
+    def _housecode_to_int(val: str) -> int:
         match = re.search(r"^([a-p])(0[1-9]|1[0-6]|[1-9])$", val.lower())
         if match:
             return (ord(match.group(1)) - ord("a")) * 16 + int(match.group(2))
         raise vol.Invalid("Invalid range")
 
-    def _elk_value(val):
+    def _elk_value(val: str) -> int:
         return int(val) if val.isdigit() else _housecode_to_int(val)
 
     vals = [s.strip() for s in str(rng).split("-")]
@@ -120,7 +121,7 @@ def _elk_range_validator(rng):
     return (start, end)
 
 
-def _has_all_unique_prefixes(value):
+def _has_all_unique_prefixes(value: list[dict[str, str]]) -> list[dict[str, str]]:
     """Validate that each m1 configured has a unique prefix.
 
     Uniqueness is determined case-independently.
@@ -214,10 +215,13 @@ async def async_setup(hass: HomeAssistant, hass_config: ConfigType) -> bool:
 
 
 @callback
-def _async_find_matching_config_entry(hass, prefix):
+def _async_find_matching_config_entry(
+    hass: HomeAssistant, prefix: str
+) -> ConfigEntry | None:
     for entry in hass.config_entries.async_entries(DOMAIN):
         if entry.unique_id == prefix:
             return entry
+    return None
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@@ -253,7 +257,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
                 _LOGGER.error("Config item: %s; %s", item, err)
                 return False
 
-    elk = elkm1.Elk(
+    elk = Elk(
         {
             "url": conf[CONF_HOST],
             "userid": conf[CONF_USERNAME],
@@ -262,7 +266,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     )
     elk.connect()
 
-    def _element_changed(element, changeset):
+    def _element_changed(element: Element, changeset: dict[str, Any]) -> None:
         if (keypress := changeset.get("last_keypress")) is None:
             return
 
@@ -275,7 +279,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             },
         )
 
-    for keypad in elk.keypads:  # pylint: disable=no-member
+    for keypad in elk.keypads:
         keypad.add_callback(_element_changed)
 
     try:
@@ -284,7 +288,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     except asyncio.TimeoutError as exc:
         raise ConfigEntryNotReady(f"Timed out connecting to {conf[CONF_HOST]}") from exc
 
-    elk_temp_unit = elk.panel.temperature_units  # pylint: disable=no-member
+    elk_temp_unit = elk.panel.temperature_units
     temperature_unit = TEMP_CELSIUS if elk_temp_unit == "C" else TEMP_FAHRENHEIT
     config["temperature_unit"] = temperature_unit
     hass.data[DOMAIN][entry.entry_id] = {
@@ -301,18 +305,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     return True
 
 
-def _included(ranges, set_to, values):
+def _included(ranges: list[tuple[int, int]], set_to: bool, values: list[bool]) -> None:
     for rng in ranges:
         if not rng[0] <= rng[1] <= len(values):
             raise vol.Invalid(f"Invalid range {rng}")
         values[rng[0] - 1 : rng[1]] = [set_to] * (rng[1] - rng[0] + 1)
 
 
-def _find_elk_by_prefix(hass, prefix):
+def _find_elk_by_prefix(hass: HomeAssistant, prefix: str) -> Elk | None:
     """Search all config entries for a given prefix."""
     for entry_id in hass.data[DOMAIN]:
         if hass.data[DOMAIN][entry_id]["prefix"] == prefix:
-            return hass.data[DOMAIN][entry_id]["elk"]
+            return cast(Elk, hass.data[DOMAIN][entry_id]["elk"])
+    return None
 
 
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@@ -329,7 +334,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
 
 async def async_wait_for_elk_to_sync(
-    elk: elkm1.Elk,
+    elk: Elk,
     login_timeout: int,
     sync_timeout: int,
 ) -> bool:
@@ -338,7 +343,9 @@ async def async_wait_for_elk_to_sync(
     sync_event = asyncio.Event()
     login_event = asyncio.Event()
 
-    def login_status(succeeded):
+    success = True
+
+    def login_status(succeeded: bool) -> None:
         nonlocal success
 
         success = succeeded
@@ -351,10 +358,9 @@ async def async_wait_for_elk_to_sync(
             login_event.set()
             sync_event.set()
 
-    def sync_complete():
+    def sync_complete() -> None:
         sync_event.set()
 
-    success = True
     elk.add_handler("login", login_status)
     elk.add_handler("sync_complete", sync_complete)
     for name, event, timeout in (
@@ -374,8 +380,8 @@ async def async_wait_for_elk_to_sync(
     return success
 
 
-def _create_elk_services(hass):
-    def _getelk(service):
+def _create_elk_services(hass: HomeAssistant) -> None:
+    def _getelk(service: ServiceCall) -> Elk:
         prefix = service.data["prefix"]
         elk = _find_elk_by_prefix(hass, prefix)
         if elk is None:
@@ -402,12 +408,18 @@ def _create_elk_services(hass):
     )
 
 
-def create_elk_entities(elk_data, elk_elements, element_type, class_, entities):
+def create_elk_entities(
+    elk_data: dict[str, Any],
+    elk_elements: list[Element],
+    element_type: str,
+    class_: Any,
+    entities: list[ElkEntity],
+) -> list[ElkEntity] | None:
     """Create the ElkM1 devices of a particular class."""
     auto_configure = elk_data["auto_configure"]
 
     if not auto_configure and not elk_data["config"][element_type]["enabled"]:
-        return
+        return None
 
     elk = elk_data["elk"]
     _LOGGER.debug("Creating elk entities for %s", elk)
@@ -427,7 +439,7 @@ def create_elk_entities(elk_data, elk_elements, element_type, class_, entities):
 class ElkEntity(Entity):
     """Base class for all Elk entities."""
 
-    def __init__(self, element, elk, elk_data):
+    def __init__(self, element: Element, elk: Elk, elk_data: dict[str, Any]) -> None:
         """Initialize the base of all Elk devices."""
         self._elk = elk
         self._element = element
@@ -450,12 +462,12 @@ class ElkEntity(Entity):
         self._unique_id = f"{uid_start}_{self._element.default_name('_')}".lower()
 
     @property
-    def name(self):
+    def name(self) -> str:
         """Name of the element."""
         return f"{self._name_prefix}{self._element.name}"
 
     @property
-    def unique_id(self):
+    def unique_id(self) -> str:
         """Return unique id of the element."""
         return self._unique_id
 
@@ -465,31 +477,31 @@ class ElkEntity(Entity):
         return False
 
     @property
-    def extra_state_attributes(self):
+    def extra_state_attributes(self) -> dict[str, Any]:
         """Return the default attributes of the element."""
         return {**self._element.as_dict(), **self.initial_attrs()}
 
     @property
-    def available(self):
+    def available(self) -> bool:
         """Is the entity available to be updated."""
         return self._elk.is_connected()
 
-    def initial_attrs(self):
+    def initial_attrs(self) -> dict[str, int]:
         """Return the underlying element's attributes as a dict."""
         attrs = {}
         attrs["index"] = self._element.index + 1
         return attrs
 
-    def _element_changed(self, element, changeset):
+    def _element_changed(self, element: Element, changeset: dict[str, Any]) -> None:
         pass
 
     @callback
-    def _element_callback(self, element, changeset):
+    def _element_callback(self, element: Element, changeset: dict[str, Any]) -> None:
         """Handle callback from an Elk element that has changed."""
         self._element_changed(element, changeset)
         self.async_write_ha_state()
 
-    async def async_added_to_hass(self):
+    async def async_added_to_hass(self) -> None:
         """Register callback for ElkM1 changes and update entity state."""
         self._element.add_callback(self._element_callback)
         self._element_callback(self._element, {})

homeassistant/components/elkm1/alarm_control_panel.py

@@ -26,7 +26,7 @@ import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.restore_state import RestoreEntity
 
-from . import ElkAttachedEntity, create_elk_entities
+from . import ElkAttachedEntity, ElkEntity, create_elk_entities
 from .const import (
     ATTR_CHANGED_BY_ID,
     ATTR_CHANGED_BY_KEYPAD,
@@ -61,7 +61,7 @@ async def async_setup_entry(
     """Set up the ElkM1 alarm platform."""
     elk_data = hass.data[DOMAIN][config_entry.entry_id]
     elk = elk_data["elk"]
-    entities: list[ElkArea] = []
+    entities: list[ElkEntity] = []
     create_elk_entities(elk_data, elk.areas, "area", ElkArea, entities)
     async_add_entities(entities, True)
 

homeassistant/components/elkm1/climate.py

@@ -37,7 +37,7 @@ async def async_setup_entry(
 ) -> None:
     """Create the Elk-M1 thermostat platform."""
     elk_data = hass.data[DOMAIN][config_entry.entry_id]
-    entities: list[ElkThermostat] = []
+    entities: list[ElkEntity] = []
     elk = elk_data["elk"]
     create_elk_entities(
         elk_data, elk.thermostats, "thermostat", ElkThermostat, entities

homeassistant/components/elkm1/light.py

@@ -21,7 +21,7 @@ async def async_setup_entry(
 ) -> None:
     """Set up the Elk light platform."""
     elk_data = hass.data[DOMAIN][config_entry.entry_id]
-    entities: list[ElkLight] = []
+    entities: list[ElkEntity] = []
     elk = elk_data["elk"]
     create_elk_entities(elk_data, elk.lights, "plc", ElkLight, entities)
     async_add_entities(entities, True)

homeassistant/components/elkm1/manifest.json

@@ -2,7 +2,7 @@
   "domain": "elkm1",
   "name": "Elk-M1 Control",
   "documentation": "https://www.home-assistant.io/integrations/elkm1",
-  "requirements": ["elkm1-lib==1.3.0"],
+  "requirements": ["elkm1-lib==1.3.3"],
   "dhcp": [{ "registered_devices": true }, { "macaddress": "00409D*" }],
   "codeowners": ["@gwww", "@bdraco"],
   "dependencies": ["network"],

homeassistant/components/elkm1/scene.py

@@ -3,12 +3,14 @@ from __future__ import annotations
 
 from typing import Any
 
+from elkm1_lib.tasks import Task
+
 from homeassistant.components.scene import Scene
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import ElkAttachedEntity, create_elk_entities
+from . import ElkAttachedEntity, ElkEntity, create_elk_entities
 from .const import DOMAIN
 
 
@@ -19,7 +21,7 @@ async def async_setup_entry(
 ) -> None:
     """Create the Elk-M1 scene platform."""
     elk_data = hass.data[DOMAIN][config_entry.entry_id]
-    entities: list[ElkTask] = []
+    entities: list[ElkEntity] = []
     elk = elk_data["elk"]
     create_elk_entities(elk_data, elk.tasks, "task", ElkTask, entities)
     async_add_entities(entities, True)
@@ -28,6 +30,8 @@ async def async_setup_entry(
 class ElkTask(ElkAttachedEntity, Scene):
     """Elk-M1 task as scene."""
 
+    _element: Task
+
     async def async_activate(self, **kwargs: Any) -> None:
         """Activate the task."""
         self._element.activate()

homeassistant/components/elkm1/sensor.py

@@ -19,7 +19,7 @@ from homeassistant.helpers import entity_platform
 from homeassistant.helpers.entity import EntityCategory
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import ElkAttachedEntity, create_elk_entities
+from . import ElkAttachedEntity, ElkEntity, create_elk_entities
 from .const import ATTR_VALUE, DOMAIN, ELK_USER_CODE_SERVICE_SCHEMA
 
 SERVICE_SENSOR_COUNTER_REFRESH = "sensor_counter_refresh"
@@ -40,7 +40,7 @@ async def async_setup_entry(
 ) -> None:
     """Create the Elk-M1 sensor platform."""
     elk_data = hass.data[DOMAIN][config_entry.entry_id]
-    entities: list[ElkSensor] = []
+    entities: list[ElkEntity] = []
     elk = elk_data["elk"]
     create_elk_entities(elk_data, elk.counters, "counter", ElkCounter, entities)
     create_elk_entities(elk_data, elk.keypads, "keypad", ElkKeypad, entities)

homeassistant/components/elkm1/switch.py

@@ -1,12 +1,14 @@
 """Support for control of ElkM1 outputs (relays)."""
 from __future__ import annotations
 
+from elkm1_lib.outputs import Output
+
 from homeassistant.components.switch import SwitchEntity
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import ElkAttachedEntity, create_elk_entities
+from . import ElkAttachedEntity, ElkEntity, create_elk_entities
 from .const import DOMAIN
 
 
@@ -17,7 +19,7 @@ async def async_setup_entry(
 ) -> None:
     """Create the Elk-M1 switch platform."""
     elk_data = hass.data[DOMAIN][config_entry.entry_id]
-    entities: list[ElkOutput] = []
+    entities: list[ElkEntity] = []
     elk = elk_data["elk"]
     create_elk_entities(elk_data, elk.outputs, "output", ElkOutput, entities)
     async_add_entities(entities, True)
@@ -26,6 +28,8 @@ async def async_setup_entry(
 class ElkOutput(ElkAttachedEntity, SwitchEntity):
     """Elk output as switch."""
 
+    _element: Output
+
     @property
     def is_on(self) -> bool:
         """Get the current output status."""

mypy.ini

@@ -693,6 +693,17 @@ no_implicit_optional = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.elkm1.__init__]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+no_implicit_optional = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.esphome.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

requirements_all.txt

@@ -574,7 +574,7 @@ elgato==3.0.0
 eliqonline==1.2.2
 
 # homeassistant.components.elkm1
-elkm1-lib==1.3.0
+elkm1-lib==1.3.3
 
 # homeassistant.components.elmax
 elmax_api==0.0.2

requirements_test_all.txt

@@ -408,7 +408,7 @@ dynalite_devices==0.1.46
 elgato==3.0.0
 
 # homeassistant.components.elkm1
-elkm1-lib==1.3.0
+elkm1-lib==1.3.3
 
 # homeassistant.components.elmax
 elmax_api==0.0.2