Improve risco typing (#108041)

homeassistant/components/risco/__init__.py

@@ -1,4 +1,6 @@
 """The Risco integration."""
+from __future__ import annotations
+
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from datetime import timedelta

homeassistant/components/risco/alarm_control_panel.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 from pyrisco.common import Partition
+from pyrisco.local.partition import Partition as LocalPartition
 
 from homeassistant.components.alarm_control_panel import (
     AlarmControlPanelEntity,
@@ -132,7 +133,7 @@ def state(self) -> str | None:
 
         return None
 
-    def _validate_code(self, code):
+    def _validate_code(self, code: str | None) -> bool:
         """Validate given code."""
         return code == self._code
 
@@ -159,7 +160,7 @@ async def async_alarm_arm_custom_bypass(self, code: str | None = None) -> None:
         """Send arm custom bypass command."""
         await self._arm(STATE_ALARM_ARMED_CUSTOM_BYPASS, code)
 
-    async def _arm(self, mode, code):
+    async def _arm(self, mode: str, code: str | None) -> None:
         if self.code_arm_required and not self._validate_code(code):
             _LOGGER.warning("Wrong code entered for %s", mode)
             return
@@ -205,7 +206,7 @@ def __init__(
     def _get_data_from_coordinator(self) -> None:
         self._partition = self.coordinator.data.partitions[self._partition_id]
 
-    async def _call_alarm_method(self, method, *args):
+    async def _call_alarm_method(self, method: str, *args: Any) -> None:
         alarm = await getattr(self._risco, method)(self._partition_id, *args)
         self._partition = alarm.partitions[self._partition_id]
         self.async_write_ha_state()
@@ -220,7 +221,7 @@ def __init__(
         self,
         system_id: str,
         partition_id: int,
-        partition: Partition,
+        partition: LocalPartition,
         partition_updates: dict[int, Callable[[], Any]],
         code: str,
         options: dict[str, Any],

homeassistant/components/risco/binary_sensor.py

@@ -4,7 +4,8 @@
 from collections.abc import Mapping
 from typing import Any
 
-from pyrisco.common import Zone
+from pyrisco.cloud.zone import Zone as CloudZone
+from pyrisco.local.zone import Zone as LocalZone
 
 from homeassistant.components.binary_sensor import (
     BinarySensorDeviceClass,
@@ -53,7 +54,7 @@ class RiscoCloudBinarySensor(RiscoCloudZoneEntity, BinarySensorEntity):
     _attr_name = None
 
     def __init__(
-        self, coordinator: RiscoDataUpdateCoordinator, zone_id: int, zone: Zone
+        self, coordinator: RiscoDataUpdateCoordinator, zone_id: int, zone: CloudZone
     ) -> None:
         """Init the zone."""
         super().__init__(coordinator=coordinator, suffix="", zone_id=zone_id, zone=zone)
@@ -70,7 +71,7 @@ class RiscoLocalBinarySensor(RiscoLocalZoneEntity, BinarySensorEntity):
     _attr_device_class = BinarySensorDeviceClass.MOTION
     _attr_name = None
 
-    def __init__(self, system_id: str, zone_id: int, zone: Zone) -> None:
+    def __init__(self, system_id: str, zone_id: int, zone: LocalZone) -> None:
         """Init the zone."""
         super().__init__(system_id=system_id, suffix="", zone_id=zone_id, zone=zone)
 
@@ -93,7 +94,7 @@ class RiscoLocalAlarmedBinarySensor(RiscoLocalZoneEntity, BinarySensorEntity):
 
     _attr_translation_key = "alarmed"
 
-    def __init__(self, system_id: str, zone_id: int, zone: Zone) -> None:
+    def __init__(self, system_id: str, zone_id: int, zone: LocalZone) -> None:
         """Init the zone."""
         super().__init__(
             system_id=system_id,
@@ -113,7 +114,7 @@ class RiscoLocalArmedBinarySensor(RiscoLocalZoneEntity, BinarySensorEntity):
 
     _attr_translation_key = "armed"
 
-    def __init__(self, system_id: str, zone_id: int, zone: Zone) -> None:
+    def __init__(self, system_id: str, zone_id: int, zone: LocalZone) -> None:
         """Init the zone."""
         super().__init__(
             system_id=system_id,

homeassistant/components/risco/config_flow.py

@@ -63,7 +63,9 @@
 ]
 
 
-async def validate_cloud_input(hass: core.HomeAssistant, data) -> dict[str, str]:
+async def validate_cloud_input(
+    hass: core.HomeAssistant, data: dict[str, Any]
+) -> dict[str, str]:
     """Validate the user input allows us to connect to Risco Cloud.
 
     Data has the keys from CLOUD_SCHEMA with values provided by the user.
@@ -124,16 +126,20 @@ def async_get_options_flow(
         """Define the config flow to handle options."""
         return RiscoOptionsFlowHandler(config_entry)
 
-    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."""
         return self.async_show_menu(
             step_id="user",
             menu_options=["cloud", "local"],
         )
 
-    async def async_step_cloud(self, user_input=None):
+    async def async_step_cloud(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Configure a cloud based alarm."""
-        errors = {}
+        errors: dict[str, str] = {}
         if user_input is not None:
             if not self._reauth_entry:
                 await self.async_set_unique_id(user_input[CONF_USERNAME])
@@ -168,14 +174,16 @@ async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
         self._reauth_entry = await self.async_set_unique_id(entry_data[CONF_USERNAME])
         return await self.async_step_cloud()
 
-    async def async_step_local(self, user_input=None):
+    async def async_step_local(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Configure a local based alarm."""
-        errors = {}
+        errors: dict[str, str] = {}
         if user_input is not None:
             try:
                 info = await validate_local_input(self.hass, user_input)
-            except CannotConnectError:
-                _LOGGER.debug("Cannot connect", exc_info=1)
+            except CannotConnectError as ex:
+                _LOGGER.debug("Cannot connect", exc_info=ex)
                 errors["base"] = "cannot_connect"
             except UnauthorizedError:
                 errors["base"] = "invalid_auth"
@@ -208,7 +216,7 @@ def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
         self.config_entry = config_entry
         self._data = {**DEFAULT_OPTIONS, **config_entry.options}
 
-    def _options_schema(self):
+    def _options_schema(self) -> vol.Schema:
         return vol.Schema(
             {
                 vol.Required(
@@ -224,15 +232,19 @@ def _options_schema(self):
             }
         )
 
-    async def async_step_init(self, user_input=None):
+    async def async_step_init(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Manage the options."""
         if user_input is not None:
             self._data = {**self._data, **user_input}
             return await self.async_step_risco_to_ha()
 
         return self.async_show_form(step_id="init", data_schema=self._options_schema())
 
-    async def async_step_risco_to_ha(self, user_input=None):
+    async def async_step_risco_to_ha(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Map Risco states to HA states."""
         if user_input is not None:
             self._data[CONF_RISCO_STATES_TO_HA] = user_input
@@ -250,7 +262,9 @@ async def async_step_risco_to_ha(self, user_input=None):
 
         return self.async_show_form(step_id="risco_to_ha", data_schema=options)
 
-    async def async_step_ha_to_risco(self, user_input=None):
+    async def async_step_ha_to_risco(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Map HA states to Risco states."""
         if user_input is not None:
             self._data[CONF_HA_STATES_TO_RISCO] = user_input

homeassistant/components/risco/entity.py

@@ -3,7 +3,9 @@
 
 from typing import Any
 
-from pyrisco.common import Zone
+from pyrisco import RiscoCloud
+from pyrisco.cloud.zone import Zone as CloudZone
+from pyrisco.local.zone import Zone as LocalZone
 
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
@@ -14,7 +16,7 @@
 from .const import DOMAIN
 
 
-def zone_unique_id(risco, zone_id: int) -> str:
+def zone_unique_id(risco: RiscoCloud, zone_id: int) -> str:
     """Return unique id for a cloud zone."""
     return f"{risco.site_uuid}_zone_{zone_id}"
 
@@ -36,7 +38,7 @@ def _handle_coordinator_update(self) -> None:
         self.async_write_ha_state()
 
     @property
-    def _risco(self):
+    def _risco(self) -> RiscoCloud:
         """Return the Risco API object."""
         return self.coordinator.risco
 
@@ -52,7 +54,7 @@ def __init__(
         coordinator: RiscoDataUpdateCoordinator,
         suffix: str,
         zone_id: int,
-        zone: Zone,
+        zone: CloudZone,
         **kwargs: Any,
     ) -> None:
         """Init the zone."""
@@ -84,7 +86,7 @@ def __init__(
         system_id: str,
         suffix: str,
         zone_id: int,
-        zone: Zone,
+        zone: LocalZone,
         **kwargs: Any,
     ) -> None:
         """Init the zone."""

homeassistant/components/risco/sensor.py

@@ -2,8 +2,11 @@
 from __future__ import annotations
 
 from collections.abc import Collection, Mapping
+from datetime import datetime
 from typing import Any
 
+from pyrisco.cloud.event import Event
+
 from homeassistant.components.binary_sensor import DOMAIN as BS_DOMAIN
 from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
 from homeassistant.config_entries import ConfigEntry
@@ -66,22 +69,23 @@ async def async_setup_entry(
 class RiscoSensor(CoordinatorEntity[RiscoEventsDataUpdateCoordinator], SensorEntity):
     """Sensor for Risco events."""
 
+    _entity_registry: er.EntityRegistry
+
     def __init__(
         self,
         coordinator: RiscoEventsDataUpdateCoordinator,
         category_id: int | None,
-        excludes: Collection[int] | None,
+        excludes: Collection[int],
         name: str,
         entry_id: str,
     ) -> None:
         """Initialize sensor."""
         super().__init__(coordinator)
-        self._event = None
+        self._event: Event | None = None
         self._category_id = category_id
         self._excludes = excludes
         self._name = name
         self._entry_id = entry_id
-        self._entity_registry: er.EntityRegistry | None = None
         self._attr_unique_id = f"events_{name}_{self.coordinator.risco.site_uuid}"
         self._attr_name = f"Risco {self.coordinator.risco.site_name} {name} Events"
         self._attr_device_class = SensorDeviceClass.TIMESTAMP
@@ -91,7 +95,7 @@ async def async_added_to_hass(self) -> None:
         await super().async_added_to_hass()
         self._entity_registry = er.async_get(self.hass)
 
-    def _handle_coordinator_update(self):
+    def _handle_coordinator_update(self) -> None:
         events = self.coordinator.data
         for event in reversed(events):
             if event.category_id in self._excludes:
@@ -103,14 +107,14 @@ def _handle_coordinator_update(self):
             self.async_write_ha_state()
 
     @property
-    def native_value(self):
+    def native_value(self) -> datetime | None:
         """Value of sensor."""
         if self._event is None:
             return None
 
-        return dt_util.parse_datetime(self._event.time).replace(
-            tzinfo=dt_util.DEFAULT_TIME_ZONE
-        )
+        if res := dt_util.parse_datetime(self._event.time):
+            return res.replace(tzinfo=dt_util.DEFAULT_TIME_ZONE)
+        return None
 
     @property
     def extra_state_attributes(self) -> Mapping[str, Any] | None:

homeassistant/components/risco/switch.py

@@ -1,6 +1,8 @@
 """Support for bypassing Risco alarm zones."""
 from __future__ import annotations
 
+from typing import Any
+
 from pyrisco.common import Zone
 
 from homeassistant.components.switch import SwitchEntity
@@ -58,11 +60,11 @@ def is_on(self) -> bool | None:
         """Return true if the zone is bypassed."""
         return self._zone.bypassed
 
-    async def async_turn_on(self, **kwargs):
+    async def async_turn_on(self, **kwargs: Any) -> None:
         """Turn the entity on."""
         await self._bypass(True)
 
-    async def async_turn_off(self, **kwargs):
+    async def async_turn_off(self, **kwargs: Any) -> None:
         """Turn the entity off."""
         await self._bypass(False)
 
@@ -92,11 +94,11 @@ def is_on(self) -> bool | None:
         """Return true if the zone is bypassed."""
         return self._zone.bypassed
 
-    async def async_turn_on(self, **kwargs):
+    async def async_turn_on(self, **kwargs: Any) -> None:
         """Turn the entity on."""
         await self._bypass(True)
 
-    async def async_turn_off(self, **kwargs):
+    async def async_turn_off(self, **kwargs: Any) -> None:
         """Turn the entity off."""
         await self._bypass(False)