mirror of
https://github.com/sascha-hemi/hacs_waste_collection_schedule.git
synced 2026-03-21 02:04:22 +01:00
add some typehints for waste_collection_api + fix typing issue in WCSCoordinator
This commit is contained in:
@@ -1,25 +1,32 @@
|
||||
# This is the class organizing the different sources when using the yaml configuration
|
||||
from datetime import time
|
||||
from random import randrange
|
||||
from typing import Any
|
||||
|
||||
import homeassistant.util.dt as dt_util
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers.dispatcher import dispatcher_send
|
||||
from homeassistant.helpers.event import async_track_time_change
|
||||
from homeassistant.helpers.event import (
|
||||
async_call_later,
|
||||
async_track_time_change,
|
||||
)
|
||||
|
||||
from . import const
|
||||
from .waste_collection_schedule import SourceShell
|
||||
|
||||
from homeassistant.helpers.event import async_call_later # isort:skip
|
||||
from .waste_collection_schedule import Customize, SourceShell
|
||||
|
||||
|
||||
class WasteCollectionApi:
|
||||
"""Class to manage the waste collection sources when using the yaml configuration."""
|
||||
|
||||
def __init__(
|
||||
self, hass, separator, fetch_time, random_fetch_time_offset, day_switch_time
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
separator: str,
|
||||
fetch_time: time,
|
||||
random_fetch_time_offset: int,
|
||||
day_switch_time: time,
|
||||
):
|
||||
self._hass = hass
|
||||
self._source_shells = []
|
||||
self._source_shells: list[SourceShell] = []
|
||||
self._separator = separator
|
||||
self._fetch_time = fetch_time
|
||||
self._random_fetch_time_offset = random_fetch_time_offset
|
||||
@@ -45,7 +52,7 @@ class WasteCollectionApi:
|
||||
)
|
||||
|
||||
# add a timer at midnight (if not already there) to update days-to
|
||||
midnight = dt_util.parse_time("00:00")
|
||||
midnight = time.min
|
||||
if midnight != self._fetch_time and midnight != self._day_switch_time:
|
||||
async_track_time_change(
|
||||
hass,
|
||||
@@ -72,11 +79,11 @@ class WasteCollectionApi:
|
||||
|
||||
def add_source_shell(
|
||||
self,
|
||||
source_name,
|
||||
customize,
|
||||
source_args,
|
||||
calendar_title,
|
||||
day_offset,
|
||||
source_name: str,
|
||||
customize: dict[str, Customize],
|
||||
source_args: Any,
|
||||
calendar_title: str,
|
||||
day_offset: int,
|
||||
):
|
||||
new_shell = SourceShell.create(
|
||||
source_name=source_name,
|
||||
@@ -100,7 +107,7 @@ class WasteCollectionApi:
|
||||
def shells(self):
|
||||
return self._source_shells
|
||||
|
||||
def get_shell(self, index):
|
||||
def get_shell(self, index: int) -> SourceShell | None:
|
||||
return self._source_shells[index] if index < len(self._source_shells) else None
|
||||
|
||||
@callback
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import datetime
|
||||
import logging
|
||||
from random import randrange
|
||||
from typing import Any
|
||||
@@ -28,18 +29,23 @@ class WCSCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
source_shell: SourceShell,
|
||||
separator,
|
||||
fetch_time,
|
||||
random_fetch_time_offset,
|
||||
day_switch_time,
|
||||
separator: str,
|
||||
fetch_time: str,
|
||||
random_fetch_time_offset: int,
|
||||
day_switch_time: str,
|
||||
):
|
||||
self._hass = hass
|
||||
self._shell = source_shell
|
||||
self._aggregator = CollectionAggregator([source_shell])
|
||||
self._separator = separator
|
||||
self._fetch_time = dt_util.parse_time(fetch_time)
|
||||
if not self._fetch_time:
|
||||
raise ValueError(f"Invalid fetch_time: {fetch_time}")
|
||||
self._random_fetch_time_offset = random_fetch_time_offset
|
||||
|
||||
self._day_switch_time = dt_util.parse_time(day_switch_time)
|
||||
if not self._day_switch_time:
|
||||
raise ValueError(f"Invalid day_switch_time: {day_switch_time}")
|
||||
|
||||
super().__init__(hass, _LOGGER, name=const.DOMAIN)
|
||||
|
||||
@@ -63,7 +69,7 @@ class WCSCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
||||
)
|
||||
|
||||
# add a timer at midnight (if not already there) to update days-to
|
||||
midnight = dt_util.parse_time("00:00")
|
||||
midnight = datetime.time.min
|
||||
if midnight != self._fetch_time and midnight != self._day_switch_time:
|
||||
async_track_time_change( # TODO: cancel on unload
|
||||
hass,
|
||||
@@ -73,9 +79,10 @@ class WCSCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
||||
midnight.second,
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> None:
|
||||
async def _async_update_data(self) -> dict[str, Any]:
|
||||
"""Update data via library."""
|
||||
await self._fetch_now()
|
||||
return {}
|
||||
|
||||
@property
|
||||
def shell(self):
|
||||
|
||||
Reference in New Issue
Block a user