mirror of
https://github.com/home-assistant/core.git
synced 2026-06-26 08:35:38 +02:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1993031364 | |||
| fdd5da3f3b |
@@ -51,6 +51,7 @@ base_platforms: &base_platforms
|
||||
- homeassistant/components/switch/**
|
||||
- homeassistant/components/text/**
|
||||
- homeassistant/components/time/**
|
||||
- homeassistant/components/timer_list/**
|
||||
- homeassistant/components/todo/**
|
||||
- homeassistant/components/tts/**
|
||||
- homeassistant/components/update/**
|
||||
|
||||
Generated
+4
@@ -1031,6 +1031,8 @@ CLAUDE.md @home-assistant/core
|
||||
/tests/components/local_calendar/ @allenporter
|
||||
/homeassistant/components/local_ip/ @issacg
|
||||
/tests/components/local_ip/ @issacg
|
||||
/homeassistant/components/local_timer_list/ @home-assistant/core @synesthesiam
|
||||
/tests/components/local_timer_list/ @home-assistant/core @synesthesiam
|
||||
/homeassistant/components/local_todo/ @allenporter
|
||||
/tests/components/local_todo/ @allenporter
|
||||
/homeassistant/components/lock/ @home-assistant/core
|
||||
@@ -1832,6 +1834,8 @@ CLAUDE.md @home-assistant/core
|
||||
/tests/components/time/ @home-assistant/core
|
||||
/homeassistant/components/time_date/ @fabaff
|
||||
/tests/components/time_date/ @fabaff
|
||||
/homeassistant/components/timer_list/ @home-assistant/core @synesthesiam
|
||||
/tests/components/timer_list/ @home-assistant/core @synesthesiam
|
||||
/homeassistant/components/tmb/ @alemuro
|
||||
/homeassistant/components/todo/ @home-assistant/core
|
||||
/tests/components/todo/ @home-assistant/core
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
"""The Local Timer list integration.
|
||||
|
||||
Creates an in-memory timer list entity (provided by the ``timer_list`` base
|
||||
platform) for each config entry, so users can create timer lists from the UI.
|
||||
"""
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
PLATFORMS = [Platform.TIMER_LIST]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up Local Timer list from a config entry."""
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
@@ -0,0 +1,33 @@
|
||||
"""Config flow for the Local Timer list integration."""
|
||||
|
||||
from typing import Any, override
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
|
||||
|
||||
from .const import CONF_TIMER_LIST_NAME, DOMAIN
|
||||
|
||||
STEP_USER_DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_TIMER_LIST_NAME): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class LocalTimerListConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Local Timer list."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
@override
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle the initial step."""
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(
|
||||
title=user_input[CONF_TIMER_LIST_NAME], data=user_input
|
||||
)
|
||||
|
||||
return self.async_show_form(step_id="user", data_schema=STEP_USER_DATA_SCHEMA)
|
||||
@@ -0,0 +1,5 @@
|
||||
"""Constants for the Local Timer list integration."""
|
||||
|
||||
DOMAIN = "local_timer_list"
|
||||
|
||||
CONF_TIMER_LIST_NAME = "timer_list_name"
|
||||
@@ -0,0 +1,10 @@
|
||||
{
|
||||
"domain": "local_timer_list",
|
||||
"name": "Local Timer list",
|
||||
"codeowners": ["@home-assistant/core", "@synesthesiam"],
|
||||
"config_flow": true,
|
||||
"dependencies": ["timer_list"],
|
||||
"documentation": "https://www.home-assistant.io/integrations/local_timer_list",
|
||||
"integration_type": "helper",
|
||||
"iot_class": "local_push"
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"timer_list_name": "Name"
|
||||
},
|
||||
"description": "Choose a name for the new timer list. The entity ID is derived from this name.",
|
||||
"submit": "Create"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
"""Local timer list platform."""
|
||||
|
||||
from homeassistant.components.timer_list import TimerListEntity, TimerListEntityFeature
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
|
||||
from .const import CONF_TIMER_LIST_NAME
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the local timer list entity from a config entry."""
|
||||
async_add_entities([LocalTimerListEntity(config_entry)])
|
||||
|
||||
|
||||
class LocalTimerListEntity(TimerListEntity):
|
||||
"""A local, in-memory timer list."""
|
||||
|
||||
_attr_supported_features = (
|
||||
TimerListEntityFeature.START_TIMER
|
||||
| TimerListEntityFeature.PAUSE_TIMER
|
||||
| TimerListEntityFeature.CANCEL_TIMER
|
||||
| TimerListEntityFeature.ADD_TIME
|
||||
)
|
||||
|
||||
def __init__(self, config_entry: ConfigEntry) -> None:
|
||||
"""Initialize the timer list."""
|
||||
super().__init__()
|
||||
self._attr_name = config_entry.data[CONF_TIMER_LIST_NAME]
|
||||
self._attr_unique_id = config_entry.entry_id
|
||||
@@ -0,0 +1,547 @@
|
||||
"""The Timer list integration.
|
||||
|
||||
A timer list entity holds many independent countdown timers (its *items*),
|
||||
mirroring how a to-do list holds many to-do items. The entity state is the
|
||||
number of active timers. Timers are kept in memory only: they do not survive a
|
||||
restart of Home Assistant in this first version (see the module-level notes on
|
||||
``async_will_remove_from_hass``).
|
||||
"""
|
||||
|
||||
from collections.abc import Callable
|
||||
import copy
|
||||
import dataclasses
|
||||
from datetime import datetime, timedelta
|
||||
from functools import partial
|
||||
import logging
|
||||
from typing import Any, final, override
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components import websocket_api
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import ATTR_NAME
|
||||
from homeassistant.core import (
|
||||
CALLBACK_TYPE,
|
||||
HomeAssistant,
|
||||
ServiceCall,
|
||||
SupportsResponse,
|
||||
callback,
|
||||
)
|
||||
from homeassistant.exceptions import ServiceValidationError
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.entity_component import EntityComponent
|
||||
from homeassistant.helpers.event import async_track_point_in_utc_time
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
from homeassistant.util import dt as dt_util, ulid as ulid_util
|
||||
|
||||
from .const import (
|
||||
ATTR_DURATION,
|
||||
ATTR_FINISH_ACTION,
|
||||
ATTR_STATUS,
|
||||
ATTR_TIMER_ID,
|
||||
DATA_COMPONENT,
|
||||
DOMAIN,
|
||||
TimerFinishAction,
|
||||
TimerListEntityFeature,
|
||||
TimerListEventType,
|
||||
TimerListServices,
|
||||
TimerStatus,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ENTITY_ID_FORMAT = DOMAIN + ".{}"
|
||||
PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA
|
||||
PLATFORM_SCHEMA_BASE = cv.PLATFORM_SCHEMA_BASE
|
||||
|
||||
_FINISHED_STATUSES = (TimerStatus.FINISHED, TimerStatus.CANCELLED)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class TimerItem:
|
||||
"""A single timer within a timer list."""
|
||||
|
||||
timer_id: str
|
||||
"""Generated unique id of the timer."""
|
||||
|
||||
name: str | None
|
||||
"""Optional user-provided name."""
|
||||
|
||||
status: TimerStatus
|
||||
"""Current status of the timer."""
|
||||
|
||||
finish_action: TimerFinishAction
|
||||
"""What happens to the timer once it finishes."""
|
||||
|
||||
duration: timedelta
|
||||
"""Original duration the timer was created with (used by ``restart``)."""
|
||||
|
||||
created_at: datetime
|
||||
"""When the timer was (re)started, in UTC."""
|
||||
|
||||
finishes_at: datetime | None = None
|
||||
"""Absolute time the timer will finish, in UTC. ``None`` unless active."""
|
||||
|
||||
remaining: timedelta | None = None
|
||||
"""Remaining time captured while paused. ``None`` unless paused."""
|
||||
|
||||
finished_at: datetime | None = None
|
||||
"""When the timer finished or was cancelled, in UTC."""
|
||||
|
||||
def remaining_at(self, now: datetime) -> timedelta:
|
||||
"""Return the time left on the timer relative to ``now``."""
|
||||
if self.status == TimerStatus.ACTIVE and self.finishes_at is not None:
|
||||
return max(timedelta(0), self.finishes_at - now)
|
||||
if self.status == TimerStatus.PAUSED and self.remaining is not None:
|
||||
return self.remaining
|
||||
return timedelta(0)
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class TimerListEvent:
|
||||
"""A change to a timer, pushed to subscribers and triggers."""
|
||||
|
||||
event_type: TimerListEventType
|
||||
item: TimerItem
|
||||
|
||||
|
||||
@callback
|
||||
def timer_to_dict(item: TimerItem, now: datetime) -> dict[str, Any]:
|
||||
"""Serialize a timer item for the websocket API and triggers."""
|
||||
return {
|
||||
ATTR_TIMER_ID: item.timer_id,
|
||||
ATTR_NAME: item.name,
|
||||
ATTR_STATUS: item.status.value,
|
||||
ATTR_FINISH_ACTION: item.finish_action.value,
|
||||
"duration": item.duration.total_seconds(),
|
||||
"created_at": item.created_at.isoformat(),
|
||||
"finishes_at": item.finishes_at.isoformat() if item.finishes_at else None,
|
||||
"finished_at": item.finished_at.isoformat() if item.finished_at else None,
|
||||
"remaining": item.remaining_at(now).total_seconds(),
|
||||
}
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the Timer list component."""
|
||||
component = hass.data[DATA_COMPONENT] = EntityComponent[TimerListEntity](
|
||||
_LOGGER, DOMAIN, hass
|
||||
)
|
||||
|
||||
websocket_api.async_register_command(hass, websocket_handle_subscribe)
|
||||
websocket_api.async_register_command(hass, websocket_handle_list)
|
||||
|
||||
component.async_register_entity_service(
|
||||
TimerListServices.START_TIMER,
|
||||
{
|
||||
vol.Optional(ATTR_NAME): cv.string,
|
||||
vol.Required(ATTR_DURATION): cv.positive_time_period,
|
||||
vol.Optional(
|
||||
ATTR_FINISH_ACTION, default=TimerFinishAction.REMOVE
|
||||
): vol.Coerce(TimerFinishAction),
|
||||
},
|
||||
_async_start_timer,
|
||||
required_features=[TimerListEntityFeature.START_TIMER],
|
||||
supports_response=SupportsResponse.OPTIONAL,
|
||||
)
|
||||
component.async_register_entity_service(
|
||||
TimerListServices.PAUSE_TIMER,
|
||||
{vol.Required(ATTR_TIMER_ID): cv.string},
|
||||
"async_pause_timer",
|
||||
required_features=[TimerListEntityFeature.PAUSE_TIMER],
|
||||
)
|
||||
component.async_register_entity_service(
|
||||
TimerListServices.UNPAUSE_TIMER,
|
||||
{vol.Required(ATTR_TIMER_ID): cv.string},
|
||||
"async_unpause_timer",
|
||||
required_features=[TimerListEntityFeature.PAUSE_TIMER],
|
||||
)
|
||||
component.async_register_entity_service(
|
||||
TimerListServices.CANCEL_TIMER,
|
||||
{vol.Required(ATTR_TIMER_ID): cv.string},
|
||||
"async_cancel_timer",
|
||||
required_features=[TimerListEntityFeature.CANCEL_TIMER],
|
||||
)
|
||||
component.async_register_entity_service(
|
||||
TimerListServices.CANCEL_ALL_TIMERS,
|
||||
None,
|
||||
"async_cancel_all_timers",
|
||||
required_features=[TimerListEntityFeature.CANCEL_TIMER],
|
||||
)
|
||||
component.async_register_entity_service(
|
||||
TimerListServices.ADD_TIME,
|
||||
{
|
||||
vol.Required(ATTR_TIMER_ID): cv.string,
|
||||
vol.Required(ATTR_DURATION): cv.positive_time_period,
|
||||
},
|
||||
_async_add_time,
|
||||
required_features=[TimerListEntityFeature.ADD_TIME],
|
||||
)
|
||||
component.async_register_entity_service(
|
||||
TimerListServices.REMOVE_TIME,
|
||||
{
|
||||
vol.Required(ATTR_TIMER_ID): cv.string,
|
||||
vol.Required(ATTR_DURATION): cv.positive_time_period,
|
||||
},
|
||||
_async_remove_time,
|
||||
required_features=[TimerListEntityFeature.ADD_TIME],
|
||||
)
|
||||
component.async_register_entity_service(
|
||||
TimerListServices.REMOVE_TIMER,
|
||||
{vol.Required(ATTR_TIMER_ID): cv.string},
|
||||
"async_remove_timer",
|
||||
)
|
||||
component.async_register_entity_service(
|
||||
TimerListServices.CLEAR_FINISHED_TIMERS,
|
||||
None,
|
||||
"async_clear_finished_timers",
|
||||
)
|
||||
component.async_register_entity_service(
|
||||
TimerListServices.GET_TIMERS,
|
||||
{vol.Optional(ATTR_STATUS): vol.All(cv.ensure_list, [vol.Coerce(TimerStatus)])},
|
||||
_async_get_timers,
|
||||
supports_response=SupportsResponse.ONLY,
|
||||
)
|
||||
|
||||
await component.async_setup(config)
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up a config entry."""
|
||||
return await hass.data[DATA_COMPONENT].async_setup_entry(entry)
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return await hass.data[DATA_COMPONENT].async_unload_entry(entry)
|
||||
|
||||
|
||||
class TimerListEntity(Entity):
|
||||
"""An entity that holds a list of independent countdown timers."""
|
||||
|
||||
_attr_should_poll = False
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the timer list."""
|
||||
self._timers: dict[str, TimerItem] = {}
|
||||
self._cancel_callbacks: dict[str, CALLBACK_TYPE] = {}
|
||||
self._update_listeners: list[Callable[[TimerListEvent], None]] = []
|
||||
|
||||
@property
|
||||
@override
|
||||
def state(self) -> int:
|
||||
"""Return the number of active timers."""
|
||||
return sum(
|
||||
timer.status == TimerStatus.ACTIVE for timer in self._timers.values()
|
||||
)
|
||||
|
||||
@property
|
||||
def timers(self) -> list[TimerItem]:
|
||||
"""Return the timers in the list."""
|
||||
return list(self._timers.values())
|
||||
|
||||
async def async_start_timer(
|
||||
self,
|
||||
*,
|
||||
name: str | None,
|
||||
duration: timedelta,
|
||||
finish_action: TimerFinishAction,
|
||||
) -> str:
|
||||
"""Create and start a new timer, returning its id."""
|
||||
now = dt_util.utcnow()
|
||||
timer_id = ulid_util.ulid_now()
|
||||
timer = TimerItem(
|
||||
timer_id=timer_id,
|
||||
name=name,
|
||||
status=TimerStatus.ACTIVE,
|
||||
finish_action=finish_action,
|
||||
duration=duration,
|
||||
created_at=now,
|
||||
finishes_at=now + duration,
|
||||
)
|
||||
self._timers[timer_id] = timer
|
||||
self._schedule(timer)
|
||||
self._notify(TimerListEventType.STARTED, timer)
|
||||
return timer_id
|
||||
|
||||
async def async_pause_timer(self, timer_id: str) -> None:
|
||||
"""Pause an active timer."""
|
||||
timer = self._get_timer(timer_id)
|
||||
if timer.status != TimerStatus.ACTIVE or timer.finishes_at is None:
|
||||
return
|
||||
timer.remaining = max(timedelta(0), timer.finishes_at - dt_util.utcnow())
|
||||
timer.finishes_at = None
|
||||
timer.status = TimerStatus.PAUSED
|
||||
self._unschedule(timer_id)
|
||||
self._notify(TimerListEventType.UPDATED, timer)
|
||||
|
||||
async def async_unpause_timer(self, timer_id: str) -> None:
|
||||
"""Resume a paused timer."""
|
||||
timer = self._get_timer(timer_id)
|
||||
if timer.status != TimerStatus.PAUSED or timer.remaining is None:
|
||||
return
|
||||
timer.finishes_at = dt_util.utcnow() + timer.remaining
|
||||
timer.remaining = None
|
||||
timer.status = TimerStatus.ACTIVE
|
||||
self._schedule(timer)
|
||||
self._notify(TimerListEventType.UPDATED, timer)
|
||||
|
||||
async def async_cancel_timer(self, timer_id: str) -> None:
|
||||
"""Cancel a timer.
|
||||
|
||||
The timer is retained in the ``cancelled`` state only when its finish
|
||||
action is ``archive``; otherwise it is removed.
|
||||
"""
|
||||
timer = self._get_timer(timer_id)
|
||||
self._unschedule(timer_id)
|
||||
timer.status = TimerStatus.CANCELLED
|
||||
timer.finishes_at = None
|
||||
timer.remaining = None
|
||||
timer.finished_at = dt_util.utcnow()
|
||||
self._notify(TimerListEventType.CANCELLED, timer)
|
||||
if timer.finish_action != TimerFinishAction.ARCHIVE:
|
||||
del self._timers[timer_id]
|
||||
self._notify(TimerListEventType.REMOVED, timer)
|
||||
|
||||
async def async_cancel_all_timers(self) -> None:
|
||||
"""Cancel every active or paused timer."""
|
||||
for timer_id in [
|
||||
timer.timer_id
|
||||
for timer in self._timers.values()
|
||||
if timer.status in (TimerStatus.ACTIVE, TimerStatus.PAUSED)
|
||||
]:
|
||||
await self.async_cancel_timer(timer_id)
|
||||
|
||||
async def async_add_time(self, timer_id: str, duration: timedelta) -> None:
|
||||
"""Add (or, with a negative duration, remove) time on a timer."""
|
||||
timer = self._get_timer(timer_id)
|
||||
if timer.status == TimerStatus.ACTIVE and timer.finishes_at is not None:
|
||||
now = dt_util.utcnow()
|
||||
finishes_at = timer.finishes_at + duration
|
||||
if finishes_at <= now:
|
||||
self._unschedule(timer_id)
|
||||
self._async_timer_finished(timer_id, now)
|
||||
return
|
||||
timer.finishes_at = finishes_at
|
||||
self._schedule(timer)
|
||||
elif timer.status == TimerStatus.PAUSED and timer.remaining is not None:
|
||||
timer.remaining = max(timedelta(0), timer.remaining + duration)
|
||||
else:
|
||||
return
|
||||
self._notify(TimerListEventType.UPDATED, timer)
|
||||
|
||||
async def async_remove_timer(self, timer_id: str) -> None:
|
||||
"""Remove a timer from the list regardless of its status."""
|
||||
timer = self._get_timer(timer_id)
|
||||
self._unschedule(timer_id)
|
||||
del self._timers[timer_id]
|
||||
self._notify(TimerListEventType.REMOVED, timer)
|
||||
|
||||
async def async_clear_finished_timers(self) -> None:
|
||||
"""Remove all finished and cancelled (archived) timers."""
|
||||
for timer_id in [
|
||||
timer.timer_id
|
||||
for timer in self._timers.values()
|
||||
if timer.status in _FINISHED_STATUSES
|
||||
]:
|
||||
timer = self._timers.pop(timer_id)
|
||||
self._notify(TimerListEventType.REMOVED, timer)
|
||||
|
||||
def _get_timer(self, timer_id: str) -> TimerItem:
|
||||
"""Return a timer by id or raise if it does not exist."""
|
||||
if (timer := self._timers.get(timer_id)) is None:
|
||||
raise ServiceValidationError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="timer_not_found",
|
||||
translation_placeholders={"timer_id": timer_id},
|
||||
)
|
||||
return timer
|
||||
|
||||
@callback
|
||||
def _schedule(self, timer: TimerItem) -> None:
|
||||
"""Schedule (or reschedule) the finish callback for a timer."""
|
||||
self._unschedule(timer.timer_id)
|
||||
assert timer.finishes_at is not None
|
||||
self._cancel_callbacks[timer.timer_id] = async_track_point_in_utc_time(
|
||||
self.hass,
|
||||
partial(self._async_timer_finished, timer.timer_id),
|
||||
timer.finishes_at,
|
||||
)
|
||||
|
||||
@callback
|
||||
def _unschedule(self, timer_id: str) -> None:
|
||||
"""Cancel a pending finish callback, if any."""
|
||||
if cancel := self._cancel_callbacks.pop(timer_id, None):
|
||||
cancel()
|
||||
|
||||
@callback
|
||||
def _async_timer_finished(self, timer_id: str, now: datetime) -> None:
|
||||
"""Handle a timer reaching its finish time."""
|
||||
self._cancel_callbacks.pop(timer_id, None)
|
||||
if (timer := self._timers.get(timer_id)) is None:
|
||||
return
|
||||
|
||||
timer.status = TimerStatus.FINISHED
|
||||
timer.finishes_at = None
|
||||
timer.remaining = None
|
||||
timer.finished_at = dt_util.utcnow()
|
||||
self._notify(TimerListEventType.FINISHED, timer)
|
||||
|
||||
if timer.finish_action == TimerFinishAction.REMOVE:
|
||||
self._timers.pop(timer_id, None)
|
||||
self._notify(TimerListEventType.REMOVED, timer)
|
||||
elif timer.finish_action == TimerFinishAction.RESTART:
|
||||
restarted_at = dt_util.utcnow()
|
||||
timer.status = TimerStatus.ACTIVE
|
||||
timer.created_at = restarted_at
|
||||
timer.finishes_at = restarted_at + timer.duration
|
||||
timer.finished_at = None
|
||||
self._schedule(timer)
|
||||
self._notify(TimerListEventType.STARTED, timer)
|
||||
|
||||
@final
|
||||
@callback
|
||||
def async_subscribe_updates(
|
||||
self, listener: Callable[[TimerListEvent], None]
|
||||
) -> CALLBACK_TYPE:
|
||||
"""Subscribe to timer change events.
|
||||
|
||||
Only future changes are pushed; the current set of timers is not
|
||||
replayed on subscribe.
|
||||
"""
|
||||
self._update_listeners.append(listener)
|
||||
|
||||
@callback
|
||||
def unsubscribe() -> None:
|
||||
self._update_listeners.remove(listener)
|
||||
|
||||
return unsubscribe
|
||||
|
||||
@callback
|
||||
def _notify(self, event_type: TimerListEventType, timer: TimerItem) -> None:
|
||||
"""Push a change event to subscribers and write entity state."""
|
||||
event = TimerListEvent(event_type=event_type, item=copy.copy(timer))
|
||||
for listener in list(self._update_listeners):
|
||||
listener(event)
|
||||
self.async_write_ha_state()
|
||||
|
||||
@override
|
||||
async def async_will_remove_from_hass(self) -> None:
|
||||
"""Cancel all pending finish callbacks."""
|
||||
for cancel in self._cancel_callbacks.values():
|
||||
cancel()
|
||||
self._cancel_callbacks.clear()
|
||||
|
||||
|
||||
async def _async_start_timer(
|
||||
entity: TimerListEntity, call: ServiceCall
|
||||
) -> dict[str, Any]:
|
||||
"""Handle the start_timer service."""
|
||||
timer_id = await entity.async_start_timer(
|
||||
name=call.data.get(ATTR_NAME),
|
||||
duration=call.data[ATTR_DURATION],
|
||||
finish_action=call.data[ATTR_FINISH_ACTION],
|
||||
)
|
||||
return {ATTR_TIMER_ID: timer_id}
|
||||
|
||||
|
||||
async def _async_add_time(entity: TimerListEntity, call: ServiceCall) -> None:
|
||||
"""Handle the add_time service."""
|
||||
await entity.async_add_time(call.data[ATTR_TIMER_ID], call.data[ATTR_DURATION])
|
||||
|
||||
|
||||
async def _async_remove_time(entity: TimerListEntity, call: ServiceCall) -> None:
|
||||
"""Handle the remove_time service."""
|
||||
await entity.async_add_time(call.data[ATTR_TIMER_ID], -call.data[ATTR_DURATION])
|
||||
|
||||
|
||||
async def _async_get_timers(
|
||||
entity: TimerListEntity, call: ServiceCall
|
||||
) -> dict[str, Any]:
|
||||
"""Handle the get_timers service."""
|
||||
now = dt_util.utcnow()
|
||||
statuses: list[TimerStatus] | None = call.data.get(ATTR_STATUS)
|
||||
return {
|
||||
"timers": [
|
||||
timer_to_dict(timer, now)
|
||||
for timer in entity.timers
|
||||
if not statuses or timer.status in statuses
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required("type"): "timer_list/item/subscribe",
|
||||
vol.Required("entity_id"): cv.entity_domain(DOMAIN),
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
async def websocket_handle_subscribe(
|
||||
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
|
||||
) -> None:
|
||||
"""Subscribe to timer changes for a timer list, with an initial snapshot."""
|
||||
entity_id: str = msg["entity_id"]
|
||||
if not (entity := hass.data[DATA_COMPONENT].get_entity(entity_id)):
|
||||
connection.send_error(
|
||||
msg["id"],
|
||||
websocket_api.ERR_NOT_FOUND,
|
||||
f"Timer list entity not found: {entity_id}",
|
||||
)
|
||||
return
|
||||
|
||||
@callback
|
||||
def forward_event(event: TimerListEvent) -> None:
|
||||
"""Forward a timer change event to the websocket connection."""
|
||||
connection.send_message(
|
||||
websocket_api.event_message(
|
||||
msg["id"],
|
||||
{
|
||||
"type": "change",
|
||||
"event_type": event.event_type.value,
|
||||
"timer": timer_to_dict(event.item, dt_util.utcnow()),
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
connection.subscriptions[msg["id"]] = entity.async_subscribe_updates(forward_event)
|
||||
connection.send_result(msg["id"])
|
||||
|
||||
now = dt_util.utcnow()
|
||||
connection.send_message(
|
||||
websocket_api.event_message(
|
||||
msg["id"],
|
||||
{
|
||||
"type": "timers",
|
||||
"timers": [timer_to_dict(timer, now) for timer in entity.timers],
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required("type"): "timer_list/item/list",
|
||||
vol.Required("entity_id"): cv.entity_domain(DOMAIN),
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
async def websocket_handle_list(
|
||||
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
|
||||
) -> None:
|
||||
"""Return the current timers for a timer list."""
|
||||
entity_id: str = msg["entity_id"]
|
||||
if not (entity := hass.data[DATA_COMPONENT].get_entity(entity_id)):
|
||||
connection.send_error(
|
||||
msg["id"],
|
||||
websocket_api.ERR_NOT_FOUND,
|
||||
f"Timer list entity not found: {entity_id}",
|
||||
)
|
||||
return
|
||||
|
||||
now = dt_util.utcnow()
|
||||
connection.send_result(
|
||||
msg["id"],
|
||||
{"timers": [timer_to_dict(timer, now) for timer in entity.timers]},
|
||||
)
|
||||
@@ -0,0 +1,76 @@
|
||||
"""Constants for the Timer list integration."""
|
||||
|
||||
from enum import IntFlag, StrEnum
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from homeassistant.util.hass_dict import HassKey
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from homeassistant.helpers.entity_component import EntityComponent
|
||||
|
||||
from . import TimerListEntity
|
||||
|
||||
DOMAIN = "timer_list"
|
||||
DATA_COMPONENT: HassKey[EntityComponent[TimerListEntity]] = HassKey(DOMAIN)
|
||||
|
||||
ATTR_TIMER_ID = "timer_id"
|
||||
ATTR_DURATION = "duration"
|
||||
ATTR_FINISH_ACTION = "finish_action"
|
||||
ATTR_FINISHES_AT = "finishes_at"
|
||||
ATTR_CREATED_AT = "created_at"
|
||||
ATTR_FINISHED_AT = "finished_at"
|
||||
ATTR_REMAINING = "remaining"
|
||||
ATTR_STATUS = "status"
|
||||
ATTR_TIMER = "timer"
|
||||
ATTR_TIMERS = "timers"
|
||||
|
||||
|
||||
class TimerListServices(StrEnum):
|
||||
"""Services for the Timer list integration."""
|
||||
|
||||
START_TIMER = "start_timer"
|
||||
PAUSE_TIMER = "pause_timer"
|
||||
UNPAUSE_TIMER = "unpause_timer"
|
||||
CANCEL_TIMER = "cancel_timer"
|
||||
CANCEL_ALL_TIMERS = "cancel_all_timers"
|
||||
ADD_TIME = "add_time"
|
||||
REMOVE_TIME = "remove_time"
|
||||
REMOVE_TIMER = "remove_timer"
|
||||
CLEAR_FINISHED_TIMERS = "clear_finished_timers"
|
||||
GET_TIMERS = "get_timers"
|
||||
|
||||
|
||||
class TimerStatus(StrEnum):
|
||||
"""Status of a single timer in a timer list."""
|
||||
|
||||
ACTIVE = "active"
|
||||
PAUSED = "paused"
|
||||
FINISHED = "finished"
|
||||
CANCELLED = "cancelled"
|
||||
|
||||
|
||||
class TimerFinishAction(StrEnum):
|
||||
"""What happens to a timer once it finishes."""
|
||||
|
||||
REMOVE = "remove"
|
||||
ARCHIVE = "archive"
|
||||
RESTART = "restart"
|
||||
|
||||
|
||||
class TimerListEventType(StrEnum):
|
||||
"""Type of change pushed to timer list subscribers."""
|
||||
|
||||
STARTED = "started"
|
||||
UPDATED = "updated"
|
||||
FINISHED = "finished"
|
||||
CANCELLED = "cancelled"
|
||||
REMOVED = "removed"
|
||||
|
||||
|
||||
class TimerListEntityFeature(IntFlag):
|
||||
"""Supported features of a timer list entity."""
|
||||
|
||||
START_TIMER = 1
|
||||
PAUSE_TIMER = 2
|
||||
CANCEL_TIMER = 4
|
||||
ADD_TIME = 8
|
||||
@@ -0,0 +1,53 @@
|
||||
{
|
||||
"entity_component": {
|
||||
"_": {
|
||||
"default": "mdi:timer-outline"
|
||||
}
|
||||
},
|
||||
"services": {
|
||||
"add_time": {
|
||||
"service": "mdi:timer-plus-outline"
|
||||
},
|
||||
"cancel_all_timers": {
|
||||
"service": "mdi:timer-cancel-outline"
|
||||
},
|
||||
"cancel_timer": {
|
||||
"service": "mdi:timer-cancel-outline"
|
||||
},
|
||||
"clear_finished_timers": {
|
||||
"service": "mdi:timer-remove-outline"
|
||||
},
|
||||
"get_timers": {
|
||||
"service": "mdi:timer-outline"
|
||||
},
|
||||
"pause_timer": {
|
||||
"service": "mdi:timer-pause-outline"
|
||||
},
|
||||
"remove_time": {
|
||||
"service": "mdi:timer-minus-outline"
|
||||
},
|
||||
"remove_timer": {
|
||||
"service": "mdi:timer-remove-outline"
|
||||
},
|
||||
"start_timer": {
|
||||
"service": "mdi:timer-plus-outline"
|
||||
},
|
||||
"unpause_timer": {
|
||||
"service": "mdi:timer-play-outline"
|
||||
}
|
||||
},
|
||||
"triggers": {
|
||||
"timer_cancelled": {
|
||||
"trigger": "mdi:timer-cancel-outline"
|
||||
},
|
||||
"timer_finished": {
|
||||
"trigger": "mdi:timer-check-outline"
|
||||
},
|
||||
"timer_started": {
|
||||
"trigger": "mdi:timer-play-outline"
|
||||
},
|
||||
"timer_updated": {
|
||||
"trigger": "mdi:timer-edit-outline"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
{
|
||||
"domain": "timer_list",
|
||||
"name": "Timer list",
|
||||
"codeowners": ["@home-assistant/core", "@synesthesiam"],
|
||||
"dependencies": ["websocket_api"],
|
||||
"documentation": "https://www.home-assistant.io/integrations/timer_list",
|
||||
"integration_type": "entity",
|
||||
"quality_scale": "internal"
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
start_timer:
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
supported_features:
|
||||
- timer_list.TimerListEntityFeature.START_TIMER
|
||||
fields:
|
||||
name:
|
||||
example: "Pasta"
|
||||
selector:
|
||||
text:
|
||||
duration:
|
||||
required: true
|
||||
example: "00:05:00"
|
||||
selector:
|
||||
duration:
|
||||
finish_action:
|
||||
default: remove
|
||||
selector:
|
||||
select:
|
||||
translation_key: finish_action
|
||||
options:
|
||||
- remove
|
||||
- archive
|
||||
- restart
|
||||
pause_timer:
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
supported_features:
|
||||
- timer_list.TimerListEntityFeature.PAUSE_TIMER
|
||||
fields:
|
||||
timer_id:
|
||||
required: true
|
||||
example: "01HZ8ABCDEF0123456789ABCDE"
|
||||
selector:
|
||||
text:
|
||||
unpause_timer:
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
supported_features:
|
||||
- timer_list.TimerListEntityFeature.PAUSE_TIMER
|
||||
fields:
|
||||
timer_id:
|
||||
required: true
|
||||
example: "01HZ8ABCDEF0123456789ABCDE"
|
||||
selector:
|
||||
text:
|
||||
cancel_timer:
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
supported_features:
|
||||
- timer_list.TimerListEntityFeature.CANCEL_TIMER
|
||||
fields:
|
||||
timer_id:
|
||||
required: true
|
||||
example: "01HZ8ABCDEF0123456789ABCDE"
|
||||
selector:
|
||||
text:
|
||||
cancel_all_timers:
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
supported_features:
|
||||
- timer_list.TimerListEntityFeature.CANCEL_TIMER
|
||||
add_time:
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
supported_features:
|
||||
- timer_list.TimerListEntityFeature.ADD_TIME
|
||||
fields:
|
||||
timer_id:
|
||||
required: true
|
||||
example: "01HZ8ABCDEF0123456789ABCDE"
|
||||
selector:
|
||||
text:
|
||||
duration:
|
||||
required: true
|
||||
example: "00:01:00"
|
||||
selector:
|
||||
duration:
|
||||
remove_time:
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
supported_features:
|
||||
- timer_list.TimerListEntityFeature.ADD_TIME
|
||||
fields:
|
||||
timer_id:
|
||||
required: true
|
||||
example: "01HZ8ABCDEF0123456789ABCDE"
|
||||
selector:
|
||||
text:
|
||||
duration:
|
||||
required: true
|
||||
example: "00:01:00"
|
||||
selector:
|
||||
duration:
|
||||
remove_timer:
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
fields:
|
||||
timer_id:
|
||||
required: true
|
||||
example: "01HZ8ABCDEF0123456789ABCDE"
|
||||
selector:
|
||||
text:
|
||||
clear_finished_timers:
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
get_timers:
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
fields:
|
||||
status:
|
||||
example: "active"
|
||||
selector:
|
||||
select:
|
||||
translation_key: status
|
||||
multiple: true
|
||||
options:
|
||||
- active
|
||||
- paused
|
||||
- finished
|
||||
- cancelled
|
||||
@@ -0,0 +1,154 @@
|
||||
{
|
||||
"entity_component": {
|
||||
"_": {
|
||||
"name": "[%key:component::timer_list::title%]"
|
||||
}
|
||||
},
|
||||
"exceptions": {
|
||||
"timer_not_found": {
|
||||
"message": "Unable to find timer with id: {timer_id}"
|
||||
}
|
||||
},
|
||||
"selector": {
|
||||
"finish_action": {
|
||||
"options": {
|
||||
"archive": "Archive",
|
||||
"remove": "Remove",
|
||||
"restart": "Restart"
|
||||
}
|
||||
},
|
||||
"status": {
|
||||
"options": {
|
||||
"active": "Active",
|
||||
"cancelled": "Cancelled",
|
||||
"finished": "Finished",
|
||||
"paused": "Paused"
|
||||
}
|
||||
}
|
||||
},
|
||||
"services": {
|
||||
"add_time": {
|
||||
"description": "Adds time to a timer.",
|
||||
"fields": {
|
||||
"duration": {
|
||||
"description": "How much time to add to the timer.",
|
||||
"name": "Duration"
|
||||
},
|
||||
"timer_id": {
|
||||
"description": "The id of the timer to add time to.",
|
||||
"name": "Timer ID"
|
||||
}
|
||||
},
|
||||
"name": "Add time"
|
||||
},
|
||||
"cancel_all_timers": {
|
||||
"description": "Cancels every active and paused timer on a timer list.",
|
||||
"name": "Cancel all timers"
|
||||
},
|
||||
"cancel_timer": {
|
||||
"description": "Cancels a timer.",
|
||||
"fields": {
|
||||
"timer_id": {
|
||||
"description": "The id of the timer to cancel.",
|
||||
"name": "Timer ID"
|
||||
}
|
||||
},
|
||||
"name": "Cancel timer"
|
||||
},
|
||||
"clear_finished_timers": {
|
||||
"description": "Removes all finished and cancelled timers from a timer list.",
|
||||
"name": "Clear finished timers"
|
||||
},
|
||||
"get_timers": {
|
||||
"description": "Gets the timers on a timer list.",
|
||||
"fields": {
|
||||
"status": {
|
||||
"description": "Only return timers with the specified statuses.",
|
||||
"name": "Status"
|
||||
}
|
||||
},
|
||||
"name": "Get timers"
|
||||
},
|
||||
"pause_timer": {
|
||||
"description": "Pauses an active timer.",
|
||||
"fields": {
|
||||
"timer_id": {
|
||||
"description": "The id of the timer to pause.",
|
||||
"name": "Timer ID"
|
||||
}
|
||||
},
|
||||
"name": "Pause timer"
|
||||
},
|
||||
"remove_time": {
|
||||
"description": "Removes time from a timer.",
|
||||
"fields": {
|
||||
"duration": {
|
||||
"description": "How much time to remove from the timer.",
|
||||
"name": "Duration"
|
||||
},
|
||||
"timer_id": {
|
||||
"description": "The id of the timer to remove time from.",
|
||||
"name": "Timer ID"
|
||||
}
|
||||
},
|
||||
"name": "Remove time"
|
||||
},
|
||||
"remove_timer": {
|
||||
"description": "Removes a timer from a timer list.",
|
||||
"fields": {
|
||||
"timer_id": {
|
||||
"description": "The id of the timer to remove.",
|
||||
"name": "Timer ID"
|
||||
}
|
||||
},
|
||||
"name": "Remove timer"
|
||||
},
|
||||
"start_timer": {
|
||||
"description": "Creates and starts a new timer on a timer list.",
|
||||
"fields": {
|
||||
"duration": {
|
||||
"description": "How long the timer should run for.",
|
||||
"name": "Duration"
|
||||
},
|
||||
"finish_action": {
|
||||
"description": "What happens to the timer once it finishes.",
|
||||
"name": "Finish action"
|
||||
},
|
||||
"name": {
|
||||
"description": "Optional name for the timer.",
|
||||
"name": "Name"
|
||||
}
|
||||
},
|
||||
"name": "Start timer"
|
||||
},
|
||||
"unpause_timer": {
|
||||
"description": "Resumes a paused timer.",
|
||||
"fields": {
|
||||
"timer_id": {
|
||||
"description": "The id of the timer to resume.",
|
||||
"name": "Timer ID"
|
||||
}
|
||||
},
|
||||
"name": "Resume timer"
|
||||
}
|
||||
},
|
||||
"title": "Timer list",
|
||||
"triggers": {
|
||||
"timer_cancelled": {
|
||||
"description": "Triggers when a timer is cancelled on a timer list.",
|
||||
"name": "Timer cancelled"
|
||||
},
|
||||
"timer_finished": {
|
||||
"description": "Triggers when a timer finishes on a timer list.",
|
||||
"name": "Timer finished"
|
||||
},
|
||||
"timer_started": {
|
||||
"description": "Triggers when a timer is started on a timer list.",
|
||||
"name": "Timer started"
|
||||
},
|
||||
"timer_updated": {
|
||||
"description": "Triggers when a timer is paused, resumed, or has time added or removed.",
|
||||
"name": "Timer updated"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
"""Provides triggers for timer lists."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from functools import partial
|
||||
from typing import TYPE_CHECKING, cast, override
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.const import ATTR_ENTITY_ID, CONF_OPTIONS, CONF_TARGET
|
||||
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback, split_entity_id
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.target import TargetEntityChangeTracker, TargetSelection
|
||||
from homeassistant.helpers.trigger import (
|
||||
Trigger,
|
||||
TriggerActionRunner,
|
||||
TriggerConfig,
|
||||
TriggerNotTriggeredReporter,
|
||||
)
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from . import TimerListEvent, timer_to_dict
|
||||
from .const import ATTR_TIMER, DATA_COMPONENT, DOMAIN, TimerListEventType
|
||||
|
||||
TRIGGER_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_TARGET): cv.TARGET_FIELDS,
|
||||
vol.Required(CONF_OPTIONS, default={}): {},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class TimerEventListener(TargetEntityChangeTracker):
|
||||
"""Subscribe to timer change events for the targeted timer list entities."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
target_selection: TargetSelection,
|
||||
listener: Callable[[str, TimerListEvent], None],
|
||||
) -> None:
|
||||
"""Initialize the listener."""
|
||||
|
||||
def entity_filter(entities: set[str]) -> set[str]:
|
||||
return {
|
||||
entity_id
|
||||
for entity_id in entities
|
||||
if split_entity_id(entity_id)[0] == DOMAIN
|
||||
}
|
||||
|
||||
super().__init__(hass, target_selection, entity_filter)
|
||||
self._listener = listener
|
||||
self._unsubscribe_listeners: list[CALLBACK_TYPE] = []
|
||||
|
||||
@override
|
||||
@callback
|
||||
def _handle_entities_update(self, tracked_entities: set[str]) -> None:
|
||||
"""Resubscribe when the set of tracked entities changes."""
|
||||
for unsub in self._unsubscribe_listeners:
|
||||
unsub()
|
||||
self._unsubscribe_listeners = []
|
||||
|
||||
component = self._hass.data[DATA_COMPONENT]
|
||||
for entity_id in tracked_entities:
|
||||
if (entity := component.get_entity(entity_id)) is None:
|
||||
continue
|
||||
self._unsubscribe_listeners.append(
|
||||
entity.async_subscribe_updates(partial(self._listener, entity_id))
|
||||
)
|
||||
|
||||
@override
|
||||
@callback
|
||||
def _unsubscribe(self) -> None:
|
||||
"""Unsubscribe from all events."""
|
||||
super()._unsubscribe()
|
||||
for unsub in self._unsubscribe_listeners:
|
||||
unsub()
|
||||
self._unsubscribe_listeners = []
|
||||
|
||||
|
||||
class TimerEventTrigger(Trigger):
|
||||
"""Trigger that fires on a specific timer change event type."""
|
||||
|
||||
_event_type: TimerListEventType
|
||||
|
||||
@override
|
||||
@classmethod
|
||||
async def async_validate_config(
|
||||
cls, hass: HomeAssistant, config: ConfigType
|
||||
) -> ConfigType:
|
||||
"""Validate config."""
|
||||
return cast(ConfigType, TRIGGER_SCHEMA(config))
|
||||
|
||||
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
|
||||
"""Initialize the trigger."""
|
||||
super().__init__(hass, config)
|
||||
if TYPE_CHECKING:
|
||||
assert config.target is not None
|
||||
self._target = config.target
|
||||
|
||||
@override
|
||||
async def async_attach_runner(
|
||||
self,
|
||||
run_action: TriggerActionRunner,
|
||||
did_not_trigger: TriggerNotTriggeredReporter | None = None,
|
||||
) -> CALLBACK_TYPE:
|
||||
"""Attach the trigger to an action runner."""
|
||||
target_selection = TargetSelection(self._target)
|
||||
if not target_selection.has_any_target:
|
||||
raise HomeAssistantError(f"No target defined in {self._target}")
|
||||
|
||||
@callback
|
||||
def handle_event(entity_id: str, event: TimerListEvent) -> None:
|
||||
if event.event_type != self._event_type:
|
||||
return
|
||||
run_action(
|
||||
{
|
||||
ATTR_ENTITY_ID: entity_id,
|
||||
ATTR_TIMER: timer_to_dict(event.item, dt_util.utcnow()),
|
||||
},
|
||||
f"timer {self._event_type.value} on {entity_id}",
|
||||
)
|
||||
|
||||
listener = TimerEventListener(self._hass, target_selection, handle_event)
|
||||
return await listener.async_setup()
|
||||
|
||||
|
||||
class TimerStartedTrigger(TimerEventTrigger):
|
||||
"""Trigger when a timer starts."""
|
||||
|
||||
_event_type = TimerListEventType.STARTED
|
||||
|
||||
|
||||
class TimerUpdatedTrigger(TimerEventTrigger):
|
||||
"""Trigger when a timer is paused, resumed, or has time added/removed."""
|
||||
|
||||
_event_type = TimerListEventType.UPDATED
|
||||
|
||||
|
||||
class TimerFinishedTrigger(TimerEventTrigger):
|
||||
"""Trigger when a timer finishes."""
|
||||
|
||||
_event_type = TimerListEventType.FINISHED
|
||||
|
||||
|
||||
class TimerCancelledTrigger(TimerEventTrigger):
|
||||
"""Trigger when a timer is cancelled."""
|
||||
|
||||
_event_type = TimerListEventType.CANCELLED
|
||||
|
||||
|
||||
TRIGGERS: dict[str, type[Trigger]] = {
|
||||
"timer_started": TimerStartedTrigger,
|
||||
"timer_updated": TimerUpdatedTrigger,
|
||||
"timer_finished": TimerFinishedTrigger,
|
||||
"timer_cancelled": TimerCancelledTrigger,
|
||||
}
|
||||
|
||||
|
||||
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
|
||||
"""Return the triggers for timer lists."""
|
||||
return TRIGGERS
|
||||
@@ -0,0 +1,9 @@
|
||||
.trigger_common: &trigger_common
|
||||
target:
|
||||
entity:
|
||||
domain: timer_list
|
||||
|
||||
timer_started: *trigger_common
|
||||
timer_updated: *trigger_common
|
||||
timer_finished: *trigger_common
|
||||
timer_cancelled: *trigger_common
|
||||
Generated
+1
@@ -12,6 +12,7 @@ FLOWS = {
|
||||
"group",
|
||||
"history_stats",
|
||||
"integration",
|
||||
"local_timer_list",
|
||||
"min_max",
|
||||
"mold_indicator",
|
||||
"otp",
|
||||
|
||||
+1
@@ -46,6 +46,7 @@ class EntityPlatforms(StrEnum):
|
||||
SWITCH = "switch"
|
||||
TEXT = "text"
|
||||
TIME = "time"
|
||||
TIMER_LIST = "timer_list"
|
||||
TODO = "todo"
|
||||
TTS = "tts"
|
||||
UPDATE = "update"
|
||||
|
||||
@@ -8448,6 +8448,12 @@
|
||||
"config_flow": true,
|
||||
"iot_class": "local_push"
|
||||
},
|
||||
"local_timer_list": {
|
||||
"name": "Local Timer list",
|
||||
"integration_type": "helper",
|
||||
"config_flow": true,
|
||||
"iot_class": "local_push"
|
||||
},
|
||||
"manual": {
|
||||
"name": "Manual Alarm Control Panel",
|
||||
"integration_type": "helper",
|
||||
|
||||
@@ -101,6 +101,7 @@ _ENTITY_COMPONENTS: set[str] = set(ENTITY_COMPONENTS).union(
|
||||
"script",
|
||||
"tag",
|
||||
"timer",
|
||||
"timer_list",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@@ -125,6 +125,7 @@ NO_IOT_CLASS = [
|
||||
"tag",
|
||||
"temperature",
|
||||
"timer",
|
||||
"timer_list",
|
||||
"trace",
|
||||
"web_rtc",
|
||||
"webhook",
|
||||
|
||||
@@ -2109,6 +2109,8 @@ NO_QUALITY_SCALE = [
|
||||
"tag",
|
||||
"temperature",
|
||||
"timer",
|
||||
"timer_list",
|
||||
"local_timer_list",
|
||||
"trace",
|
||||
"usage_prediction",
|
||||
"web_rtc",
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Tests for the Local Timer list integration."""
|
||||
@@ -0,0 +1,28 @@
|
||||
"""Test the Local Timer list config flow."""
|
||||
|
||||
from homeassistant.components.local_timer_list.const import CONF_TIMER_LIST_NAME, DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_USER
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
|
||||
async def test_user_flow_creates_entity(hass: HomeAssistant) -> None:
|
||||
"""Test the user config flow creates an entry and a named timer list entity."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_TIMER_LIST_NAME: "Kitchen"}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "Kitchen"
|
||||
assert result["data"] == {CONF_TIMER_LIST_NAME: "Kitchen"}
|
||||
|
||||
state = hass.states.get("timer_list.kitchen")
|
||||
assert state is not None
|
||||
assert state.state == "0"
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Tests for the Timer list integration."""
|
||||
|
||||
from homeassistant.components.timer_list import TimerListEntity
|
||||
from homeassistant.components.timer_list.const import DOMAIN, TimerListEntityFeature
|
||||
from homeassistant.config_entries import ConfigEntry, ConfigFlow
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
|
||||
from tests.common import MockConfigEntry, MockPlatform, mock_platform
|
||||
|
||||
TEST_DOMAIN = "test"
|
||||
|
||||
ALL_FEATURES = (
|
||||
TimerListEntityFeature.START_TIMER
|
||||
| TimerListEntityFeature.PAUSE_TIMER
|
||||
| TimerListEntityFeature.CANCEL_TIMER
|
||||
| TimerListEntityFeature.ADD_TIME
|
||||
)
|
||||
|
||||
|
||||
class MockFlow(ConfigFlow):
|
||||
"""Test flow."""
|
||||
|
||||
|
||||
class MockTimerListEntity(TimerListEntity):
|
||||
"""Test timer list entity."""
|
||||
|
||||
_attr_supported_features = ALL_FEATURES
|
||||
|
||||
def __init__(self, name: str = "Timers") -> None:
|
||||
"""Initialize entity."""
|
||||
super().__init__()
|
||||
self._attr_name = name
|
||||
|
||||
|
||||
async def create_mock_platform(
|
||||
hass: HomeAssistant,
|
||||
entities: list[TimerListEntity],
|
||||
) -> MockConfigEntry:
|
||||
"""Create a timer_list platform with the specified entities."""
|
||||
|
||||
async def async_setup_entry_platform(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up test timer_list platform via config entry."""
|
||||
async_add_entities(entities)
|
||||
|
||||
mock_platform(
|
||||
hass,
|
||||
f"{TEST_DOMAIN}.{DOMAIN}",
|
||||
MockPlatform(async_setup_entry=async_setup_entry_platform),
|
||||
)
|
||||
|
||||
config_entry = MockConfigEntry(domain=TEST_DOMAIN)
|
||||
config_entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
return config_entry
|
||||
@@ -0,0 +1,64 @@
|
||||
"""Fixtures for the Timer list component tests."""
|
||||
|
||||
from collections.abc import Generator
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.timer_list import TimerListEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import TEST_DOMAIN, MockFlow, MockTimerListEntity, create_mock_platform
|
||||
|
||||
from tests.common import MockModule, mock_config_flow, mock_integration, mock_platform
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def config_flow_fixture(hass: HomeAssistant) -> Generator[None]:
|
||||
"""Mock config flow."""
|
||||
mock_platform(hass, f"{TEST_DOMAIN}.config_flow")
|
||||
|
||||
with mock_config_flow(TEST_DOMAIN, MockFlow):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_setup_integration(hass: HomeAssistant) -> None:
|
||||
"""Fixture to set up a mock integration."""
|
||||
|
||||
async def async_setup_entry_init(
|
||||
hass: HomeAssistant, config_entry: ConfigEntry
|
||||
) -> bool:
|
||||
"""Set up test config entry."""
|
||||
await hass.config_entries.async_forward_entry_setups(
|
||||
config_entry, [Platform.TIMER_LIST]
|
||||
)
|
||||
return True
|
||||
|
||||
async def async_unload_entry_init(
|
||||
hass: HomeAssistant, config_entry: ConfigEntry
|
||||
) -> bool:
|
||||
await hass.config_entries.async_unload_platforms(
|
||||
config_entry, [Platform.TIMER_LIST]
|
||||
)
|
||||
return True
|
||||
|
||||
mock_platform(hass, f"{TEST_DOMAIN}.config_flow")
|
||||
mock_integration(
|
||||
hass,
|
||||
MockModule(
|
||||
TEST_DOMAIN,
|
||||
async_setup_entry=async_setup_entry_init,
|
||||
async_unload_entry=async_unload_entry_init,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(name="test_entity")
|
||||
async def mock_test_entity(hass: HomeAssistant) -> TimerListEntity:
|
||||
"""Fixture that creates a test timer list entity."""
|
||||
entity = MockTimerListEntity()
|
||||
entity.entity_id = "timer_list.timers"
|
||||
await create_mock_platform(hass, [entity])
|
||||
return entity
|
||||
@@ -0,0 +1,329 @@
|
||||
"""Tests for the Timer list integration."""
|
||||
|
||||
from datetime import timedelta
|
||||
from typing import Any
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.timer_list import TimerListEntity
|
||||
from homeassistant.components.timer_list.const import DOMAIN
|
||||
from homeassistant.const import ATTR_ENTITY_ID, ATTR_NAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ServiceValidationError
|
||||
|
||||
from tests.common import async_fire_time_changed
|
||||
from tests.typing import WebSocketGenerator
|
||||
|
||||
TEST_ENTITY_ID = "timer_list.timers"
|
||||
|
||||
|
||||
async def _start_timer(
|
||||
hass: HomeAssistant,
|
||||
*,
|
||||
duration: int = 60,
|
||||
name: str | None = None,
|
||||
finish_action: str = "remove",
|
||||
) -> str:
|
||||
"""Start a timer and return its id."""
|
||||
data: dict[str, Any] = {
|
||||
"duration": {"seconds": duration},
|
||||
"finish_action": finish_action,
|
||||
}
|
||||
if name is not None:
|
||||
data[ATTR_NAME] = name
|
||||
result = await hass.services.async_call(
|
||||
DOMAIN,
|
||||
"start_timer",
|
||||
data,
|
||||
target={ATTR_ENTITY_ID: TEST_ENTITY_ID},
|
||||
blocking=True,
|
||||
return_response=True,
|
||||
)
|
||||
return result[TEST_ENTITY_ID]["timer_id"]
|
||||
|
||||
|
||||
async def _get_timers(
|
||||
hass: HomeAssistant, status: list[str] | None = None
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Return the timers via the get_timers service."""
|
||||
data: dict[str, Any] = {}
|
||||
if status is not None:
|
||||
data["status"] = status
|
||||
result = await hass.services.async_call(
|
||||
DOMAIN,
|
||||
"get_timers",
|
||||
data,
|
||||
target={ATTR_ENTITY_ID: TEST_ENTITY_ID},
|
||||
blocking=True,
|
||||
return_response=True,
|
||||
)
|
||||
return result[TEST_ENTITY_ID]["timers"]
|
||||
|
||||
|
||||
async def _call(hass: HomeAssistant, service: str, **fields: Any) -> None:
|
||||
"""Call an entity service targeting the test entity."""
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
service,
|
||||
fields,
|
||||
target={ATTR_ENTITY_ID: TEST_ENTITY_ID},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_start_timer_sets_state_and_returns_id(hass: HomeAssistant) -> None:
|
||||
"""Test starting timers updates the state and returns an id."""
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "0"
|
||||
|
||||
timer_id = await _start_timer(hass, name="Pasta")
|
||||
assert timer_id
|
||||
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "1"
|
||||
|
||||
await _start_timer(hass)
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "2"
|
||||
|
||||
timers = await _get_timers(hass)
|
||||
assert len(timers) == 2
|
||||
assert {timer["status"] for timer in timers} == {"active"}
|
||||
assert timers[0]["name"] == "Pasta"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_get_timers_status_filter(hass: HomeAssistant) -> None:
|
||||
"""Test the get_timers status filter."""
|
||||
await _start_timer(hass)
|
||||
paused_id = await _start_timer(hass)
|
||||
await _call(hass, "pause_timer", timer_id=paused_id)
|
||||
|
||||
assert len(await _get_timers(hass, status=["active"])) == 1
|
||||
assert len(await _get_timers(hass, status=["paused"])) == 1
|
||||
assert len(await _get_timers(hass, status=["active", "paused"])) == 2
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_finish_action_remove(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test a timer is removed after finishing with the remove action."""
|
||||
await _start_timer(hass, duration=60, finish_action="remove")
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "1"
|
||||
|
||||
freezer.tick(timedelta(seconds=61))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "0"
|
||||
assert await _get_timers(hass) == []
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_finish_action_archive(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test a timer is retained as finished with the archive action."""
|
||||
await _start_timer(hass, duration=60, finish_action="archive")
|
||||
|
||||
freezer.tick(timedelta(seconds=61))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "0"
|
||||
timers = await _get_timers(hass)
|
||||
assert len(timers) == 1
|
||||
assert timers[0]["status"] == "finished"
|
||||
assert timers[0]["finished_at"] is not None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_finish_action_restart(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test a timer restarts itself with the restart action."""
|
||||
timer_id = await _start_timer(hass, duration=60, finish_action="restart")
|
||||
|
||||
freezer.tick(timedelta(seconds=61))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "1"
|
||||
timers = await _get_timers(hass)
|
||||
assert len(timers) == 1
|
||||
assert timers[0]["timer_id"] == timer_id
|
||||
assert timers[0]["status"] == "active"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_pause_and_unpause(hass: HomeAssistant) -> None:
|
||||
"""Test pausing and resuming a timer."""
|
||||
timer_id = await _start_timer(hass)
|
||||
|
||||
await _call(hass, "pause_timer", timer_id=timer_id)
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "0"
|
||||
timers = await _get_timers(hass)
|
||||
assert timers[0]["status"] == "paused"
|
||||
assert timers[0]["finishes_at"] is None
|
||||
|
||||
await _call(hass, "unpause_timer", timer_id=timer_id)
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "1"
|
||||
timers = await _get_timers(hass)
|
||||
assert timers[0]["status"] == "active"
|
||||
assert timers[0]["finishes_at"] is not None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_add_and_remove_time(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test adding and removing time on a timer."""
|
||||
timer_id = await _start_timer(hass, duration=60)
|
||||
|
||||
await _call(hass, "add_time", timer_id=timer_id, duration={"seconds": 60})
|
||||
assert (await _get_timers(hass))[0]["remaining"] == pytest.approx(120, abs=1)
|
||||
|
||||
await _call(hass, "remove_time", timer_id=timer_id, duration={"seconds": 90})
|
||||
assert (await _get_timers(hass))[0]["remaining"] == pytest.approx(30, abs=1)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_remove_time_finishes_timer(hass: HomeAssistant) -> None:
|
||||
"""Test removing more time than remaining finishes the timer immediately."""
|
||||
timer_id = await _start_timer(hass, duration=60, finish_action="archive")
|
||||
|
||||
await _call(hass, "remove_time", timer_id=timer_id, duration={"seconds": 120})
|
||||
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "0"
|
||||
assert (await _get_timers(hass))[0]["status"] == "finished"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_cancel_timer_remove(hass: HomeAssistant) -> None:
|
||||
"""Test cancelling a remove-action timer deletes it."""
|
||||
timer_id = await _start_timer(hass, finish_action="remove")
|
||||
await _call(hass, "cancel_timer", timer_id=timer_id)
|
||||
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "0"
|
||||
assert await _get_timers(hass) == []
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_cancel_timer_archive(hass: HomeAssistant) -> None:
|
||||
"""Test cancelling an archive-action timer retains it as cancelled."""
|
||||
timer_id = await _start_timer(hass, finish_action="archive")
|
||||
await _call(hass, "cancel_timer", timer_id=timer_id)
|
||||
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "0"
|
||||
timers = await _get_timers(hass)
|
||||
assert len(timers) == 1
|
||||
assert timers[0]["status"] == "cancelled"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_cancel_all_timers(hass: HomeAssistant) -> None:
|
||||
"""Test cancelling all timers."""
|
||||
await _start_timer(hass)
|
||||
await _start_timer(hass, finish_action="archive")
|
||||
|
||||
await _call(hass, "cancel_all_timers")
|
||||
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "0"
|
||||
# The archived timer is retained as cancelled, the remove timer is deleted.
|
||||
assert len(await _get_timers(hass)) == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_clear_finished_timers(hass: HomeAssistant) -> None:
|
||||
"""Test clearing finished and cancelled timers."""
|
||||
timer_id = await _start_timer(hass, finish_action="archive")
|
||||
await _call(hass, "cancel_timer", timer_id=timer_id)
|
||||
await _start_timer(hass)
|
||||
assert len(await _get_timers(hass)) == 2
|
||||
|
||||
await _call(hass, "clear_finished_timers")
|
||||
|
||||
timers = await _get_timers(hass)
|
||||
assert len(timers) == 1
|
||||
assert timers[0]["status"] == "active"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_remove_timer(hass: HomeAssistant) -> None:
|
||||
"""Test removing a single timer regardless of status."""
|
||||
timer_id = await _start_timer(hass)
|
||||
await _call(hass, "remove_timer", timer_id=timer_id)
|
||||
|
||||
assert hass.states.get(TEST_ENTITY_ID).state == "0"
|
||||
assert await _get_timers(hass) == []
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_timer_not_found(hass: HomeAssistant) -> None:
|
||||
"""Test acting on an unknown timer id raises."""
|
||||
with pytest.raises(ServiceValidationError):
|
||||
await _call(hass, "pause_timer", timer_id="does-not-exist")
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_websocket_subscribe(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
) -> None:
|
||||
"""Test subscribing to timer changes with an initial snapshot."""
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json_auto_id(
|
||||
{"type": "timer_list/item/subscribe", "entity_id": TEST_ENTITY_ID}
|
||||
)
|
||||
msg = await client.receive_json()
|
||||
assert msg["success"]
|
||||
|
||||
msg = await client.receive_json()
|
||||
assert msg["event"] == {"type": "timers", "timers": []}
|
||||
|
||||
timer_id = await _start_timer(hass, name="Pasta")
|
||||
msg = await client.receive_json()
|
||||
assert msg["event"]["type"] == "change"
|
||||
assert msg["event"]["event_type"] == "started"
|
||||
assert msg["event"]["timer"]["timer_id"] == timer_id
|
||||
assert msg["event"]["timer"]["name"] == "Pasta"
|
||||
|
||||
await _call(hass, "cancel_timer", timer_id=timer_id)
|
||||
msg = await client.receive_json()
|
||||
assert msg["event"]["event_type"] == "cancelled"
|
||||
# remove-action timers also emit a removed event after cancellation.
|
||||
msg = await client.receive_json()
|
||||
assert msg["event"]["event_type"] == "removed"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("test_entity")
|
||||
async def test_websocket_list(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
) -> None:
|
||||
"""Test the one-shot websocket list command."""
|
||||
await _start_timer(hass, name="Pasta")
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json_auto_id(
|
||||
{"type": "timer_list/item/list", "entity_id": TEST_ENTITY_ID}
|
||||
)
|
||||
msg = await client.receive_json()
|
||||
assert msg["success"]
|
||||
assert len(msg["result"]["timers"]) == 1
|
||||
assert msg["result"]["timers"][0]["name"] == "Pasta"
|
||||
|
||||
|
||||
async def test_websocket_subscribe_unknown_entity(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
test_entity: TimerListEntity,
|
||||
) -> None:
|
||||
"""Test subscribing to an entity that does not exist."""
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json_auto_id(
|
||||
{"type": "timer_list/item/subscribe", "entity_id": "timer_list.unknown"}
|
||||
)
|
||||
msg = await client.receive_json()
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == "not_found"
|
||||
@@ -0,0 +1,153 @@
|
||||
"""Tests for the Timer list triggers."""
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
import pytest
|
||||
|
||||
from homeassistant.components import automation
|
||||
from homeassistant.components.timer_list.const import DOMAIN
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
CONF_ENTITY_ID,
|
||||
CONF_PLATFORM,
|
||||
CONF_TARGET,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, ServiceCall
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import MockTimerListEntity, create_mock_platform
|
||||
|
||||
from tests.common import async_fire_time_changed, async_mock_service
|
||||
from tests.components.common import assert_trigger_options_supported
|
||||
|
||||
TEST_ENTITY_ID = "timer_list.timers"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service_calls(hass: HomeAssistant) -> list[ServiceCall]:
|
||||
"""Track calls to a mock service."""
|
||||
return async_mock_service(hass, "test", "automation")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def setup_entity(hass: HomeAssistant) -> None:
|
||||
"""Create a timer list entity via the mock platform."""
|
||||
entity = MockTimerListEntity()
|
||||
entity.entity_id = TEST_ENTITY_ID
|
||||
entity._attr_unique_id = "timers"
|
||||
await create_mock_platform(hass, [entity])
|
||||
|
||||
|
||||
async def _setup_automation(hass: HomeAssistant, trigger_type: str) -> None:
|
||||
"""Set up an automation for the given timer list trigger."""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: {
|
||||
"triggers": [
|
||||
{
|
||||
CONF_PLATFORM: f"{DOMAIN}.{trigger_type}",
|
||||
CONF_TARGET: {CONF_ENTITY_ID: TEST_ENTITY_ID},
|
||||
}
|
||||
],
|
||||
"action": {
|
||||
"service": "test.automation",
|
||||
"data": {
|
||||
"entity_id": "{{ trigger.entity_id }}",
|
||||
"timer_id": "{{ trigger.timer.timer_id }}",
|
||||
"status": "{{ trigger.timer.status }}",
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def _start_timer(hass: HomeAssistant, finish_action: str = "remove") -> str:
|
||||
"""Start a timer and return its id."""
|
||||
result = await hass.services.async_call(
|
||||
DOMAIN,
|
||||
"start_timer",
|
||||
{"duration": {"seconds": 60}, "finish_action": finish_action},
|
||||
target={ATTR_ENTITY_ID: TEST_ENTITY_ID},
|
||||
blocking=True,
|
||||
return_response=True,
|
||||
)
|
||||
return result[TEST_ENTITY_ID]["timer_id"]
|
||||
|
||||
|
||||
async def test_timer_finished_trigger(
|
||||
hass: HomeAssistant,
|
||||
service_calls: list[ServiceCall],
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test the timer_finished trigger fires when a timer finishes."""
|
||||
await _setup_automation(hass, "timer_finished")
|
||||
timer_id = await _start_timer(hass)
|
||||
|
||||
assert len(service_calls) == 0
|
||||
|
||||
freezer.tick(timedelta(seconds=61))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(service_calls) == 1
|
||||
assert service_calls[0].data == {
|
||||
"entity_id": TEST_ENTITY_ID,
|
||||
"timer_id": timer_id,
|
||||
"status": "finished",
|
||||
}
|
||||
|
||||
|
||||
async def test_timer_started_trigger(
|
||||
hass: HomeAssistant, service_calls: list[ServiceCall]
|
||||
) -> None:
|
||||
"""Test the timer_started trigger fires when a timer starts."""
|
||||
await _setup_automation(hass, "timer_started")
|
||||
timer_id = await _start_timer(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(service_calls) == 1
|
||||
assert service_calls[0].data["timer_id"] == timer_id
|
||||
assert service_calls[0].data["status"] == "active"
|
||||
|
||||
|
||||
async def test_timer_cancelled_trigger(
|
||||
hass: HomeAssistant, service_calls: list[ServiceCall]
|
||||
) -> None:
|
||||
"""Test the timer_cancelled trigger fires and ignores other events."""
|
||||
await _setup_automation(hass, "timer_cancelled")
|
||||
timer_id = await _start_timer(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 0
|
||||
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
"cancel_timer",
|
||||
{"timer_id": timer_id},
|
||||
target={ATTR_ENTITY_ID: TEST_ENTITY_ID},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
assert len(service_calls) == 1
|
||||
assert service_calls[0].data["status"] == "cancelled"
|
||||
|
||||
|
||||
async def test_trigger_options_supported(hass: HomeAssistant) -> None:
|
||||
"""Test the timer list triggers do not advertise behavior or duration."""
|
||||
for trigger_type in (
|
||||
"timer_started",
|
||||
"timer_updated",
|
||||
"timer_finished",
|
||||
"timer_cancelled",
|
||||
):
|
||||
await assert_trigger_options_supported(
|
||||
hass,
|
||||
f"{DOMAIN}.{trigger_type}",
|
||||
None,
|
||||
supports_behavior=False,
|
||||
supports_duration=False,
|
||||
)
|
||||
@@ -94,6 +94,7 @@
|
||||
'text',
|
||||
'time',
|
||||
'timer',
|
||||
'timer_list',
|
||||
'trace',
|
||||
'tts',
|
||||
'update',
|
||||
@@ -202,6 +203,7 @@
|
||||
'text',
|
||||
'time',
|
||||
'timer',
|
||||
'timer_list',
|
||||
'trace',
|
||||
'tts',
|
||||
'update',
|
||||
|
||||
Reference in New Issue
Block a user