Coverage for custom_components/supernotify/sentences.py: 99%
139 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 21:14 +0000
1"""Voice and chat commands for Home Assistant's built-in conversation agent - beta.
3The built-in agent doesn't use an AI model, so it can't use the tools in llm.py. It matches fixed
4sentences instead, so Supernotify registers a few of its own, the same way an automation with a
5conversation trigger does. English and Italian for now, and switched on in the options.
6"""
8from __future__ import annotations
10import datetime as dt
11import logging
12import re
13from typing import TYPE_CHECKING, Any
15from homeassistant.core import CALLBACK_TYPE, HomeAssistant
16from homeassistant.core import Context as HAContext
17from homeassistant.helpers.script import ScriptRunResult
18from homeassistant.helpers.trigger import async_initialize_triggers, async_validate_trigger_config
19from homeassistant.util import dt as dt_util
21from . import DOMAIN
22from .model import CommandType, GlobalTargetType, RecipientType
23from .schema import EnvelopeOutcome
25if TYPE_CHECKING:
26 from .engine import SupernotifyEngine
28_LOGGER = logging.getLogger(__name__)
30EVERYONE = ("everyone", "everybody", "all", "tutti", "tutte")
31DEFAULT_LANGUAGE = "en"
32# 3:30pm, 3 pm, 12 a.m.
33MERIDIEM_TIME = re.compile(r"(\d{1,2})(?::(\d{2}))?\s*([ap])\.?\s*m\.?", re.IGNORECASE)
34# 15:30, 15.30, 1530, 3:30 or a bare hour like 3
35CLOCK_TIME = re.compile(r"(\d{1,2})(?:[:.]?(\d{2}))?")
36# "half three" isn't taken, since some places mean 2:30 by it
37HALF_PAST = re.compile(r"half past (\w+)", re.IGNORECASE)
38HOURS_SAID = {
39 word: hour
40 for hour, word in enumerate(
41 ["one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten", "eleven", "twelve"], start=1
42 )
43} | {str(hour): hour for hour in range(1, 13)}
44NAMED_TIMES = {"midnight": 0, "noon": 12}
45# Optional words before the minutes, which the wildcard can take in with the number
46MINUTES_PREFIXES = ("the", "next", "i", "prossimi")
48# hassil sentence templates by language - (a|b) is a choice, [a] is optional and {slot} is a wildcard.
49# A wildcard needs fixed words between it and the next one, so the message is always introduced.
50# Every language is registered, since the built-in agent matches trigger sentences whatever the
51# language of the assistant, and the answer is given in the language of the sentence that matched.
52SENTENCES: dict[str, dict[str, list[str]]] = {
53 "en": {
54 "notify": [
55 "(tell|notify|message) {name} (that|saying) {message}",
56 "send [a] (message|notification) to {name} (that|saying) {message}",
57 ],
58 "snooze_minutes": ["(snooze|mute|pause) [all] [my] notifications for [the] [next] {minutes} minutes"],
59 "snooze_until": ["(snooze|mute|pause) [all] [my] notifications until {time}"],
60 "snooze_hour": ["(snooze|mute|pause) [all] [my] notifications for [an|one] hour"],
61 # "mute ... until I say" is left to snooze_until, so it doesn't match both
62 "silence": ["(silence|mute) [all] [my] notifications", "silence [all] [my] notifications until I say"],
63 "resume": [
64 "(unsnooze|unmute|resume) [all] [my] notifications",
65 "turn [all] [my] notifications back on",
66 ],
67 "last": ["what was the last notification", "what notification was sent last"],
68 },
69 "it": {
70 "notify": [
71 "(avvisa|avverti|informa) {name} che {message}",
72 "(di|dì|di'|dici|scrivi) a {name} che {message}",
73 "(manda|invia) (un messaggio|una notifica) a {name} (che dice|dicendo|con scritto|che) {message}",
74 ],
75 "snooze_minutes": [
76 "(posticipa|sospendi|silenzia|metti in pausa) [tutte] [le] [mie] notifiche per [i] [prossimi] {minutes} minuti"
77 ],
78 "snooze_until": ["(posticipa|sospendi|silenzia|metti in pausa) [tutte] [le] [mie] notifiche fino alle {time}"],
79 "snooze_hour": [
80 "(posticipa|sospendi|silenzia|metti in pausa) [tutte] [le] [mie] notifiche per (un'ora|un ora|una ora)"
81 ],
82 "silence": ["(silenzia|zittisci|disattiva) [tutte] [le] [mie] notifiche [finché non lo dico|fino a nuovo ordine]"],
83 "resume": [
84 "(riattiva|ripristina|riprendi) [tutte] [le] [mie] notifiche",
85 "(riaccendi|rimetti) [tutte] [le] [mie] notifiche",
86 ],
87 "last": [
88 "(qual è|qual era|quale è|quale era) [stata] (l'ultima|l ultima) notifica",
89 "(dimmi|leggi|ripeti) (l'ultima|l ultima) notifica",
90 ],
91 },
92}
94# What to say back, by the language of the sentence that matched
95RESPONSES: dict[str, dict[str, str]] = {
96 "en": {
97 "sent": "Sent",
98 "sent_to": "Sent to {name}",
99 "not_sent": "Sorry, the notification wasn't sent",
100 "which": "Which {name} do you mean: {choices}?",
101 "or": " or ",
102 "unknown_name": "I don't know anyone called {name}. I can notify {known}, or everyone",
103 "nobody_yet": "nobody yet",
104 "minutes_as_number": "Say how many minutes as a number, for example snooze notifications for 30 minutes",
105 "time_as_clock": "Say a time like 15:30 or 3:30pm, for example snooze notifications until 15:30",
106 "resumed_yours": "Turned your notifications back on",
107 "resumed_all": "Turned all notifications back on",
108 "silenced_yours": "Silenced your notifications until you turn them back on",
109 "silenced_all": "Silenced all notifications until you turn them back on",
110 "snoozed_yours": "Snoozed your notifications until {until}",
111 "snoozed_all": "Snoozed all notifications until {until}",
112 "no_last": "There haven't been any notifications since Home Assistant started",
113 "last_sent": "At {when}: {message}. Sent by {sent}",
114 "last_not_sent": "At {when}: {message}. It wasn't sent by anything",
115 "unknown_command": "Supernotify doesn't know the command {command}",
116 },
117 "it": {
118 "sent": "Inviata",
119 "sent_to": "Inviata a {name}",
120 "not_sent": "Mi dispiace, la notifica non è stata inviata",
121 "which": "Quale {name} intendi: {choices}?",
122 "or": " o ",
123 "unknown_name": "Non conosco nessuno che si chiama {name}. Posso avvisare {known}, oppure tutti",
124 "nobody_yet": "ancora nessuno",
125 "minutes_as_number": "Dimmi i minuti con un numero, per esempio posticipa le notifiche per 30 minuti",
126 "time_as_clock": "Dimmi un orario come 15 o 15:30, per esempio posticipa le notifiche fino alle 15:30",
127 "resumed_yours": "Ho riattivato le tue notifiche",
128 "resumed_all": "Ho riattivato tutte le notifiche",
129 "silenced_yours": "Ho silenziato le tue notifiche finché non le riattivi",
130 "silenced_all": "Ho silenziato tutte le notifiche finché non le riattivi",
131 "snoozed_yours": "Ho posticipato le tue notifiche fino alle {until}",
132 "snoozed_all": "Ho posticipato tutte le notifiche fino alle {until}",
133 "no_last": "Non ci sono state notifiche da quando Home Assistant si è avviato",
134 "last_sent": "Alle {when}: {message}. Inviata da {sent}",
135 "last_not_sent": "Alle {when}: {message}. Non è stata inviata da nessun canale",
136 "unknown_command": "Supernotify non conosce il comando {command}",
137 },
138}
141def _say(language: str, key: str, **values: Any) -> str:
142 """A response in the given language, or in English if there's none"""
143 return RESPONSES.get(language, RESPONSES[DEFAULT_LANGUAGE])[key].format(**values)
146async def async_register_sentences(hass: HomeAssistant, engine: SupernotifyEngine) -> CALLBACK_TYPE | None:
147 """Register the sentences with the built-in conversation agent, returning how to remove them"""
149 async def action(run_variables: dict[str, Any], _context: HAContext | None = None) -> ScriptRunResult:
150 trigger: dict[str, Any] = run_variables["trigger"]
151 language, _, command = str(trigger["id"]).rpartition(".")
152 response = await async_respond(
153 engine, command, trigger.get("slots", {}), _requester(trigger), language or DEFAULT_LANGUAGE
154 )
155 return ScriptRunResult(conversation_response=response, service_response=None, variables={})
157 try:
158 config = await async_validate_trigger_config(
159 hass,
160 [
161 {"platform": "conversation", "id": f"{language}.{command}", "command": sentences}
162 for language, commands in SENTENCES.items()
163 for command, sentences in commands.items()
164 ],
165 )
166 except Exception as e:
167 _LOGGER.warning("SUPERNOTIFY Unable to register sentences with the built-in conversation agent: %s", e)
168 return None
169 return await async_initialize_triggers(hass, config, action, DOMAIN, "Supernotify sentences", _log)
172def _log(level: int, msg: str, **kwargs: Any) -> None:
173 _LOGGER.log(level, "SUPERNOTIFY Sentences: %s", msg, **kwargs)
176def _requester(trigger: dict[str, Any]) -> HAContext:
177 """The context of the conversation, so notifications and snoozes are linked to who asked"""
178 context: dict[str, Any] = (trigger.get("user_input") or {}).get("context") or {}
179 return HAContext(user_id=context.get("user_id"), parent_id=context.get("id"))
182async def async_respond(
183 engine: SupernotifyEngine, command: str, slots: dict[str, Any], context: HAContext, language: str = DEFAULT_LANGUAGE
184) -> str:
185 """Carry out a command, returning what to say back in the language of the sentence"""
186 if command == "notify":
187 return await _notify(engine, str(slots.get("name", "")), str(slots.get("message", "")), context, language)
188 if command == "last":
189 return _last(engine, language)
190 if command == "snooze_minutes":
191 minutes = _minutes(str(slots.get("minutes", "")))
192 if not minutes.isdigit() or int(minutes) < 1:
193 return _say(language, "minutes_as_number")
194 return _snooze(engine, CommandType.SNOOZE, context, language, dt.timedelta(minutes=int(minutes)))
195 if command == "snooze_until":
196 spoken = str(slots.get("time", ""))
197 if spoken.strip().casefold() == "i say":
198 return _snooze(engine, CommandType.SILENCE, context, language)
199 until = _next_time(spoken, language)
200 if until is None:
201 return _say(language, "time_as_clock")
202 return _snooze(engine, CommandType.SNOOZE, context, language, until - dt_util.now())
203 if command == "snooze_hour":
204 return _snooze(engine, CommandType.SNOOZE, context, language, dt.timedelta(hours=1))
205 if command == "silence":
206 return _snooze(engine, CommandType.SILENCE, context, language)
207 if command == "resume":
208 return _snooze(engine, CommandType.NORMAL, context, language)
209 return _say(language, "unknown_command", command=command)
212def _minutes(spoken: str) -> str:
213 """The number of minutes, without optional words the wildcard took in, like 'the next' or 'i prossimi'"""
214 words: list[str] = spoken.split()
215 while words and words[0].casefold() in MINUTES_PREFIXES:
216 words.pop(0)
217 return " ".join(words)
220async def _notify(engine: SupernotifyEngine, name: str, message: str, context: HAContext, language: str) -> str:
221 people = engine.context.people_registry
222 target: list[str] | None = None
223 if name.strip().casefold() not in EVERYONE:
224 named = people.people_named(name)
225 if len(named) > 1:
226 choices = _say(language, "or").join(sorted(r.alias or r.name for r in named))
227 return _say(language, "which", name=name, choices=choices)
228 if not named:
229 known = ", ".join(sorted(r.alias or r.name for r in people.enabled_recipients())) or _say(language, "nobody_yet")
230 return _say(language, "unknown_name", name=name, known=known)
231 target = [named[0].entity_id]
232 notification = await engine.async_send_message(message, target=target, context=context)
233 if notification is None or not notification.delivered:
234 return _say(language, "not_sent")
235 return _say(language, "sent") if target is None else _say(language, "sent_to", name=name)
238def _next_time(spoken: str, language: str = DEFAULT_LANGUAGE) -> dt.datetime | None:
239 """The next time the clock shows this time, today or tomorrow
241 In English, 3:30 without am or pm is whichever 3:30 comes next, while 03:30 is always the morning.
242 Italian uses the 24 hour clock, so 3:30 is always the morning there.
243 """
244 spoken = spoken.strip()
245 either_half_of_day = False
246 if (match := MERIDIEM_TIME.fullmatch(spoken)) is not None:
247 hour, minute = int(match.group(1)), int(match.group(2) or 0)
248 if not 1 <= hour <= 12:
249 return None
250 hour = hour % 12 + (12 if match.group(3).lower() == "p" else 0)
251 elif (match := CLOCK_TIME.fullmatch(spoken)) is not None:
252 hour, minute = int(match.group(1)), int(match.group(2) or 0)
253 either_half_of_day = language != "it" and not match.group(1).startswith("0")
254 elif language != "it" and (match := HALF_PAST.fullmatch(spoken)) is not None:
255 if match.group(1).casefold() not in HOURS_SAID:
256 return None
257 hour, minute, either_half_of_day = HOURS_SAID[match.group(1).casefold()], 30, True
258 elif spoken.casefold() in NAMED_TIMES:
259 hour, minute = NAMED_TIMES[spoken.casefold()], 0
260 else:
261 return None
262 if hour > 23 or minute > 59:
263 return None
264 hours = [hour % 12, hour % 12 + 12] if either_half_of_day and 1 <= hour <= 12 else [hour]
265 now = dt_util.as_local(dt_util.now())
266 return min(_next_occurrence(now, candidate, minute) for candidate in hours)
269def _next_occurrence(now: dt.datetime, hour: int, minute: int) -> dt.datetime:
270 until = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
271 return until if until > now else until + dt.timedelta(days=1)
274def _snooze(
275 engine: SupernotifyEngine, cmd: CommandType, context: HAContext, language: str, snooze_for: dt.timedelta | None = None
276) -> str:
277 """Snooze, silence or resume everything for the person asking, or for everyone if they aren't known"""
278 person_id: str | None = engine.context.people_registry.person_id_for_user_id(context.user_id)
279 recipient_type = RecipientType.USER if person_id else RecipientType.EVERYONE
280 engine.context.snoozer.register_snooze(
281 cmd, GlobalTargetType.EVERYTHING, None, recipient_type, person_id, snooze_for, reason="Voice command"
282 )
283 whose = "yours" if person_id else "all"
284 if cmd == CommandType.NORMAL:
285 return _say(language, f"resumed_{whose}")
286 if cmd == CommandType.SILENCE:
287 return _say(language, f"silenced_{whose}")
288 until = dt_util.as_local(dt_util.now() + (snooze_for or engine.context.snoozer.snooze_period))
289 return _say(language, f"snoozed_{whose}", until=_clock(until, language))
292def _clock(when: dt.datetime, language: str) -> str:
293 """A time to say back, with am or pm in English so a wrong guess at the half of the day can be put right"""
294 if language == "it":
295 return when.strftime("%H:%M")
296 return f"{when.hour % 12 or 12}:{when.minute:02d}{'am' if when.hour < 12 else 'pm'}"
299def _last(engine: SupernotifyEngine, language: str) -> str:
300 notification = engine.last_notification
301 if notification is None:
302 return _say(language, "no_last")
303 when = dt_util.as_local(notification.created).strftime("%H:%M")
304 sent = [name for name, outcomes in notification.deliveries.items() if outcomes.get(EnvelopeOutcome.SUCCESS)]
305 if sent:
306 return _say(language, "last_sent", when=when, message=notification.message, sent=", ".join(sent))
307 return _say(language, "last_not_sent", when=when, message=notification.message)