diff --git a/homeassistant/components/risco/__init__.py b/homeassistant/components/risco/__init__.py index f143244d31d..88f8ba9bdfa 100644 --- a/homeassistant/components/risco/__init__.py +++ b/homeassistant/components/risco/__init__.py @@ -12,6 +12,8 @@ from pyrisco import ( RiscoLocal, UnauthorizedError, ) +from pyrisco.cloud.alarm import Alarm +from pyrisco.cloud.event import Event from pyrisco.common import Partition, Zone from homeassistant.config_entries import ConfigEntry @@ -175,10 +177,12 @@ async def _update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None: await hass.config_entries.async_reload(entry.entry_id) -class RiscoDataUpdateCoordinator(DataUpdateCoordinator): +class RiscoDataUpdateCoordinator(DataUpdateCoordinator[Alarm]): """Class to manage fetching risco data.""" - def __init__(self, hass, risco, scan_interval): + def __init__( + self, hass: HomeAssistant, risco: RiscoCloud, scan_interval: int + ) -> None: """Initialize global risco data updater.""" self.risco = risco interval = timedelta(seconds=scan_interval) @@ -189,7 +193,7 @@ class RiscoDataUpdateCoordinator(DataUpdateCoordinator): update_interval=interval, ) - async def _async_update_data(self): + async def _async_update_data(self) -> Alarm: """Fetch data from risco.""" try: return await self.risco.get_state() @@ -197,13 +201,15 @@ class RiscoDataUpdateCoordinator(DataUpdateCoordinator): raise UpdateFailed(error) from error -class RiscoEventsDataUpdateCoordinator(DataUpdateCoordinator): +class RiscoEventsDataUpdateCoordinator(DataUpdateCoordinator[list[Event]]): """Class to manage fetching risco data.""" - def __init__(self, hass, risco, eid, scan_interval): + def __init__( + self, hass: HomeAssistant, risco: RiscoCloud, eid: str, scan_interval: int + ) -> None: """Initialize global risco data updater.""" self.risco = risco - self._store = Store( + self._store = Store[dict[str, Any]]( hass, LAST_EVENT_STORAGE_VERSION, f"risco_{eid}_last_event_timestamp" ) interval = timedelta(seconds=scan_interval) @@ -214,7 +220,7 @@ class RiscoEventsDataUpdateCoordinator(DataUpdateCoordinator): update_interval=interval, ) - async def _async_update_data(self): + async def _async_update_data(self) -> list[Event]: """Fetch data from risco.""" last_store = await self._store.async_load() or {} last_timestamp = last_store.get( diff --git a/homeassistant/components/risco/sensor.py b/homeassistant/components/risco/sensor.py index f2cb9821166..bb416b8c550 100644 --- a/homeassistant/components/risco/sensor.py +++ b/homeassistant/components/risco/sensor.py @@ -1,7 +1,7 @@ """Sensor for Risco Events.""" from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Collection, Mapping from typing import Any from homeassistant.components.binary_sensor import DOMAIN as BS_DOMAIN @@ -63,10 +63,17 @@ async def async_setup_entry( async_add_entities(sensors) -class RiscoSensor(CoordinatorEntity, SensorEntity): +class RiscoSensor(CoordinatorEntity[RiscoEventsDataUpdateCoordinator], SensorEntity): """Sensor for Risco events.""" - def __init__(self, coordinator, category_id, excludes, name, entry_id) -> None: + def __init__( + self, + coordinator: RiscoEventsDataUpdateCoordinator, + category_id: int | None, + excludes: Collection[int] | None, + name: str, + entry_id: str, + ) -> None: """Initialize sensor.""" super().__init__(coordinator) self._event = None