diff --git a/homeassistant/components/risco/__init__.py b/homeassistant/components/risco/__init__.py index 9c62447ee047..c58721e4e280 100644 --- a/homeassistant/components/risco/__init__.py +++ b/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 diff --git a/homeassistant/components/risco/alarm_control_panel.py b/homeassistant/components/risco/alarm_control_panel.py index a72efe1629cc..8a233d0b5fe4 100644 --- a/homeassistant/components/risco/alarm_control_panel.py +++ b/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], diff --git a/homeassistant/components/risco/binary_sensor.py b/homeassistant/components/risco/binary_sensor.py index f60b0bf3c352..ea7153b2aee5 100644 --- a/homeassistant/components/risco/binary_sensor.py +++ b/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, diff --git a/homeassistant/components/risco/config_flow.py b/homeassistant/components/risco/config_flow.py index ef96714742d0..61a452a7ecbe 100644 --- a/homeassistant/components/risco/config_flow.py +++ b/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,7 +232,9 @@ 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} @@ -232,7 +242,9 @@ async def async_step_init(self, user_input=None): 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 diff --git a/homeassistant/components/risco/entity.py b/homeassistant/components/risco/entity.py index e522c29ce193..ac3c04cfc2e6 100644 --- a/homeassistant/components/risco/entity.py +++ b/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.""" diff --git a/homeassistant/components/risco/sensor.py b/homeassistant/components/risco/sensor.py index 1d60ea4d7c27..138c08c18f6b 100644 --- a/homeassistant/components/risco/sensor.py +++ b/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: diff --git a/homeassistant/components/risco/switch.py b/homeassistant/components/risco/switch.py index 9b34479f8a21..d22b2bb21923 100644 --- a/homeassistant/components/risco/switch.py +++ b/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)