Skip to content

Commit

Permalink
Improve risco generic typing (#84644)
Browse files Browse the repository at this point in the history
  • Loading branch information
cdce8p committed Dec 28, 2022
1 parent 7feb080 commit cb37df6
Show file tree
Hide file tree
Showing 2 changed files with 23 additions and 10 deletions.
20 changes: 13 additions & 7 deletions homeassistant/components/risco/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
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
Expand Down Expand Up @@ -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)
Expand All @@ -189,21 +193,23 @@ def __init__(self, hass, risco, scan_interval):
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()
except (CannotConnectError, UnauthorizedError, OperationError) as error:
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)
Expand All @@ -214,7 +220,7 @@ def __init__(self, hass, risco, eid, scan_interval):
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(
Expand Down
13 changes: 10 additions & 3 deletions homeassistant/components/risco/sensor.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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
Expand Down

0 comments on commit cb37df6

Please sign in to comment.