Coverage for custom_components/supernotify/sentences.py: 99%

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, NamedTuple

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, QualifiedTargetType, RecipientType, TargetType

23from .schema import EnvelopeOutcome

25if TYPE_CHECKING:

26 from .engine import SupernotifyEngine

27 from .notification import Notification

29_LOGGER = logging.getLogger(__name__)

32class Subject(NamedTuple):

33 """The camera or tag a voice snooze is for, and how it was said"""

35 target_type: QualifiedTargetType

36 target: str

37 spoken: str

40EVERYONE = ("everyone", "everybody", "all", "tutti", "tutte")

41DEFAULT_LANGUAGE = "en"

42# 3:30pm, 3 pm, 12 a.m.

43MERIDIEM_TIME = re.compile(r"(\d{1,2})(?::(\d{2}))?\s*([ap])\.?\s*m\.?", re.IGNORECASE)

44# 15:30, 15.30, 1530, 3:30 or a bare hour like 3

45CLOCK_TIME = re.compile(r"(\d{1,2})(?:[:.]?(\d{2}))?")

46# "half three" isn't taken, since some places mean 2:30 by it

47HALF_PAST = re.compile(r"half past (\w+)", re.IGNORECASE)

48HOURS_SAID = {

49 word: hour

50 for hour, word in enumerate(

51 ["one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten", "eleven", "twelve"], start=1

52 )

53} | {str(hour): hour for hour in range(1, 13)}

54NAMED_TIMES = {"midnight": 0, "noon": 12}

55# Optional words before the minutes, which the wildcard can take in with the number

56MINUTES_PREFIXES = ("the", "next", "i", "prossimi")

57# A tag starting or ending with this names a camera, as "driveway camera" or "telecamera vialetto"

58CAMERA_WORDS = {"en": "camera", "it": "telecamera"}

59HELP_URL = "https://supernotify.rhizomatics.org.uk/help/"

61# hassil sentence templates by language - (a|b) is a choice, [a] is optional and {slot} is a wildcard.

62# A wildcard needs fixed words between it and the next one, so the message is always introduced.

63# Every language is registered, since the built-in agent matches trigger sentences whatever the

64# language of the assistant, and the answer is given in the language of the sentence that matched.

65SENTENCES: dict[str, dict[str, list[str]]] = {

66 "en": {

67 "notify": [

68 "(tell|notify|message) {name} (that|saying) {message}",

69 "send [a] (message|notification) to {name} (that|saying) {message}",

70 ],

71 # the tag sentence comes first, since the agent takes the first that matches, and the minutes

72 # wildcard in the other would otherwise take in "driveway for 20"

73 "snooze_minutes": [

74 "(snooze|mute|pause) [all] [my] notifications for {tag} for [the] [next] {minutes} minutes",

75 "(snooze|mute|pause) [all] [my] notifications for [the] [next] {minutes} minutes",

76 ],

77 "snooze_until": [

78 "(snooze|mute|pause) [all] [my] notifications until {time}",

79 "(snooze|mute|pause) [all] [my] notifications for {tag} until {time}",

80 ],

81 "snooze_hour": [

82 "(snooze|mute|pause) [all] [my] notifications for [an|one] hour",

83 "(snooze|mute|pause) [all] [my] notifications for {tag} for [an|one] hour",

84 ],

85 # "mute ... until I say" is left to snooze_until, so it doesn't match both

86 "silence": ["(silence|mute) [all] [my] notifications", "silence [all] [my] notifications until I say"],

87 # the way out when an alert that is meant to be critical keeps going off

88 "silence_everything": [

89 "(silence|mute) [all] [my] notifications (even|including) [the] critical [ones|alerts|notifications]",

90 "(silence|mute) [all] [my] notifications [and] critical [ones|alerts|notifications] too",

91 ],

92 "resume": [

93 "(unsnooze|unmute|resume) [all] [my] notifications",

94 "turn [all] [my] notifications back on",

95 "(unsnooze|unmute|resume) [all] [my] notifications for {tag}",

96 "turn [all] [my] notifications for {tag} back on",

97 ],

98 "last": ["what was the last notification", "what notification was sent last"],

99 # speech to text often hears the name as two words

100 "help": [

101 "how (do|can) I get help (on|with|for) (supernotify|super notify)",

102 "[get] help (on|with|for) (supernotify|super notify)",

103 "(supernotify|super notify) help",

104 ],

105 },

106 "it": {

107 "notify": [

108 "(avvisa|avverti|informa) {name} che {message}",

109 "(di|dì|di'|dici|scrivi) a {name} che {message}",

110 "(manda|invia) (un messaggio|una notifica) a {name} (che dice|dicendo|con scritto|che) {message}",

111 ],

112 "snooze_minutes": [

113 "(posticipa|sospendi|silenzia|metti in pausa) [tutte] [le] [mie] notifiche per [i] [prossimi] {minutes} minuti",

114 (

115 "(posticipa|sospendi|silenzia|metti in pausa) [tutte] [le] [mie] notifiche "

116 "(di|del|della|dello|dei|delle) {tag} per [i] [prossimi] {minutes} minuti"

117 ),

118 ],

119 "snooze_until": [

120 "(posticipa|sospendi|silenzia|metti in pausa) [tutte] [le] [mie] notifiche fino alle {time}",

121 (

122 "(posticipa|sospendi|silenzia|metti in pausa) [tutte] [le] [mie] notifiche "

123 "(di|del|della|dello|dei|delle) {tag} fino alle {time}"

124 ),

125 ],

126 "snooze_hour": [

128 (

129 "(posticipa|sospendi|silenzia|metti in pausa) [tutte] [le] [mie] notifiche "

130 "(di|del|della|dello|dei|delle) {tag} per (un'ora|un ora|una ora)"

131 ),

132 ],

133 "silence": ["(silenzia|zittisci|disattiva) [tutte] [le] [mie] notifiche [finché non lo dico|fino a nuovo ordine]"],

134 "silence_everything": [

135 "(silenzia|zittisci|disattiva) [tutte] [le] [mie] notifiche (anche|comprese) [quelle] critiche",

136 ],

137 "resume": [

138 "(riattiva|ripristina|riprendi) [tutte] [le] [mie] notifiche",

139 "(riaccendi|rimetti) [tutte] [le] [mie] notifiche",

140 "(riattiva|ripristina|riprendi) [tutte] [le] [mie] notifiche (di|del|della|dello|dei|delle) {tag}",

141 ],

142 "last": [

143 "(qual è|qual era|quale è|quale era) [stata] (l'ultima|l ultima) notifica",

144 "(dimmi|leggi|ripeti) (l'ultima|l ultima) notifica",

145 ],

146 "help": [

148 "aiuto (per|su|con) (supernotify|super notify)",

149 ],

150 },

151}

152

153

154class Voice(NamedTuple):

155 """What to say back, in the language of the sentence that matched

156

157 The responses are in the translations, as `common.voice_*`. A language with sentences needs its

158 responses there too, while English fills in for any language without them. Translations can't

159 start or end with a space, so the space before a part joined into a response is added here.

160 """

161

162 language: str

163 responses: dict[str, str]

164

165 def say(self, key: str, **values: Any) -> str:

166 return self.responses[f"component.{DOMAIN}.common.voice_{key}"].format(**values)

167

168 def joined(self, choices: list[str]) -> str:

169 return f" {self.say('or')} ".join(choices)

170

171 def can_say(self, key: str) -> bool:

172 return f"component.{DOMAIN}.common.voice_{key}" in self.responses

173

174

175async def async_register_sentences(hass: HomeAssistant, engine: SupernotifyEngine) -> CALLBACK_TYPE | None:

176 """Register the sentences with the built-in conversation agent, returning how to remove them"""

177

178 async def action(run_variables: dict[str, Any], _context: HAContext | None = None) -> ScriptRunResult:

179 trigger: dict[str, Any] = run_variables["trigger"]

180 language, _, command = str(trigger["id"]).rpartition(".")

181 response = await async_respond(

182 engine, command, trigger.get("slots", {}), _requester(trigger), language or DEFAULT_LANGUAGE

183 )

184 return ScriptRunResult(conversation_response=response, service_response=None, variables={})

185

186 try:

187 config = await async_validate_trigger_config(

188 hass,

189 [

190 {"platform": "conversation", "id": f"{language}.{command}", "command": sentences}

191 for language, commands in SENTENCES.items()

192 for command, sentences in commands.items()

193 ],

194 )

195 except Exception as e:

196 _LOGGER.warning("SUPERNOTIFY Unable to register sentences with the built-in conversation agent: %s", e)

197 return None

198 return await async_initialize_triggers(hass, config, action, DOMAIN, "Supernotify sentences", _log)

199

200

201def _log(level: int, msg: str, **kwargs: Any) -> None:

202 _LOGGER.log(level, "SUPERNOTIFY Sentences: %s", msg, **kwargs)

203

204

205def _requester(trigger: dict[str, Any]) -> HAContext:

206 """The context of the conversation, so notifications and snoozes are linked to who asked"""

207 context: dict[str, Any] = (trigger.get("user_input") or {}).get("context") or {}

208 return HAContext(user_id=context.get("user_id"), parent_id=context.get("id"))

209

210

211async def async_respond(

212 engine: SupernotifyEngine, command: str, slots: dict[str, Any], context: HAContext, language: str = DEFAULT_LANGUAGE

213) -> str:

214 """Carry out a command, returning what to say back in the language of the sentence"""

215 voice = Voice(language, await engine.context.hass_api.translations("common", language))

216 if command == "notify":

217 return await _notify(engine, str(slots.get("name", "")), str(slots.get("message", "")), context, voice)

218 if command == "last":

219 return _last(engine, voice)

220 if command == "help":

221 return voice.say("help", url=HELP_URL)

222 tag: str | None = _tag(str(slots["tag"])) if slots.get("tag") else None

223 subject: Subject | None = None

224 if tag is not None:

225 subject_or_reply: Subject | str = _subject(engine, tag, command, voice)

226 if isinstance(subject_or_reply, str):

227 return subject_or_reply

228 subject = subject_or_reply

229 if command == "snooze_minutes":

230 minutes = _minutes(str(slots.get("minutes", "")))

231 if not minutes.isdigit() or int(minutes) < 1:

232 return voice.say("minutes_as_number")

233 return _snooze(engine, CommandType.SNOOZE, context, voice, dt.timedelta(minutes=int(minutes)), subject)

234 if command == "snooze_until":

235 spoken = str(slots.get("time", ""))

236 if spoken.strip().casefold() == "i say":

237 return _snooze(engine, CommandType.SILENCE, context, voice, subject=subject)

238 until = _next_time(spoken, voice.language)

239 if until is None:

240 return voice.say("time_as_clock")

241 return _snooze(engine, CommandType.SNOOZE, context, voice, until - dt_util.now(), subject)

242 if command == "snooze_hour":

243 return _snooze(engine, CommandType.SNOOZE, context, voice, dt.timedelta(hours=1), subject)

244 if command == "silence":

245 return _snooze(engine, CommandType.SILENCE, context, voice)

246 if command == "silence_everything":

247 return _snooze(engine, CommandType.SILENCE, context, voice, everything=True)

248 if command == "resume":

249 return _snooze(engine, CommandType.NORMAL, context, voice, subject=subject)

250 return voice.say("unknown_command", command=command)

251

252

253def _subject(engine: SupernotifyEngine, tag: str, command: str, voice: Voice) -> Subject | str:

254 """A camera snooze when the tag names one camera, otherwise a tag snooze - or what to say back

255 when it names several cameras, or nothing at all"""

256 words: list[str] = tag.split()

257 camera_word: str = CAMERA_WORDS.get(voice.language, CAMERA_WORDS[DEFAULT_LANGUAGE])

258 if len(words) > 1 and camera_word in (words[0].casefold(), words[-1].casefold()):

259 name: str = " ".join(words[1:] if words[0].casefold() == camera_word else words[:-1])

260 hass_api = engine.context.hass_api

261 cameras: list[str] = sorted({

262 e for e in hass_api.entity_ids_named(name) + hass_api.entity_ids_named(tag) if e.startswith("camera.")

263 })

264 if len(cameras) > 1:

265 return voice.say("which_camera", tag=tag, cameras=voice.joined(cameras))

266 if cameras:

267 return Subject(QualifiedTargetType.CAMERA, cameras[0], tag)

268 if command != "resume" and not engine.tag_matches(tag):

269 return voice.say("unknown_tag", tag=tag)

270 return Subject(QualifiedTargetType.TAG, tag, tag)

271

272

273def _tag(spoken: str) -> str:

274 """The tag, without a leading 'the' the wildcard took in"""

275 words: list[str] = spoken.split()

276 if len(words) > 1 and words[0].casefold() == "the":

277 words.pop(0)

278 return " ".join(words)

279

280

281def _minutes(spoken: str) -> str:

282 """The number of minutes, without optional words the wildcard took in, like 'the next' or 'i prossimi'"""

283 words: list[str] = spoken.split()

284 while words and words[0].casefold() in MINUTES_PREFIXES:

285 words.pop(0)

286 return " ".join(words)

287

288

289async def _notify(engine: SupernotifyEngine, name: str, message: str, context: HAContext, voice: Voice) -> str:

290 people = engine.context.people_registry

291 target: list[str] | None = None

292 if name.strip().casefold() not in EVERYONE:

293 named = people.people_named(name)

294 if len(named) > 1:

295 choices = voice.joined(sorted(r.alias or r.name for r in named))

296 return voice.say("which", name=name, choices=choices)

297 if not named:

298 known = ", ".join(sorted(r.alias or r.name for r in people.enabled_recipients())) or voice.say("nobody_yet")

299 return voice.say("unknown_name", name=name, known=known)

300 target = [named[0].entity_id]

301 notification = await engine.async_send_message(message, target=target, context=context)

302 if notification is None or not notification.delivered:

303 return voice.say("not_sent")

304 return voice.say("sent") if target is None else voice.say("sent_to", name=name)

305

306

307def _next_time(spoken: str, language: str = DEFAULT_LANGUAGE) -> dt.datetime | None:

308 """The next time the clock shows this time, today or tomorrow

309

310 In English, 3:30 without am or pm is whichever 3:30 comes next, while 03:30 is always the morning.

311 Italian uses the 24 hour clock, so 3:30 is always the morning there.

312 """

313 spoken = spoken.strip()

314 either_half_of_day = False

315 if (match := MERIDIEM_TIME.fullmatch(spoken)) is not None:

316 hour, minute = int(match.group(1)), int(match.group(2) or 0)

317 if not 1 <= hour <= 12:

318 return None

319 hour = hour % 12 + (12 if match.group(3).lower() == "p" else 0)

320 elif (match := CLOCK_TIME.fullmatch(spoken)) is not None:

321 hour, minute = int(match.group(1)), int(match.group(2) or 0)

322 either_half_of_day = language != "it" and not match.group(1).startswith("0")

323 elif language != "it" and (match := HALF_PAST.fullmatch(spoken)) is not None:

324 if match.group(1).casefold() not in HOURS_SAID:

325 return None

326 hour, minute, either_half_of_day = HOURS_SAID[match.group(1).casefold()], 30, True

327 elif spoken.casefold() in NAMED_TIMES:

328 hour, minute = NAMED_TIMES[spoken.casefold()], 0

329 else:

330 return None

331 if hour > 23 or minute > 59:

332 return None

333 hours = [hour % 12, hour % 12 + 12] if either_half_of_day and 1 <= hour <= 12 else [hour]

334 now = dt_util.as_local(dt_util.now())

335 return min(_next_occurrence(now, candidate, minute) for candidate in hours)

336

337

338def _next_occurrence(now: dt.datetime, hour: int, minute: int) -> dt.datetime:

339 until = now.replace(hour=hour, minute=minute, second=0, microsecond=0)

340 return until if until > now else until + dt.timedelta(days=1)

341

342

343def _snooze(

344 engine: SupernotifyEngine,

345 cmd: CommandType,

346 context: HAContext,

347 voice: Voice,

348 snooze_for: dt.timedelta | None = None,

349 subject: Subject | None = None,

350 everything: bool = False,

351) -> str:

352 """Snooze, silence or resume everything, or a camera or tag, for the person asking, or for everyone

353 if they aren't known"""

354 person_id: str | None = engine.context.people_registry.person_id_for_user_id(context.user_id)

355 recipient_type = RecipientType.USER if person_id else RecipientType.EVERYONE

356 # "silence" said at bedtime with nothing named still lets critical alerts through - nobody saying it

357 # means the smoke alarm too, so critical ones are held back only when asked for ("even critical").

358 # Resuming clears both kinds, so a snooze of everything can still be undone.

359 target_type: TargetType

360 if subject:

361 target_type = subject.target_type

362 else:

363 target_type = GlobalTargetType.EVERYTHING if everything else GlobalTargetType.NONCRITICAL

364 engine.context.snoozer.register_snooze(

365 cmd, target_type, subject.target if subject else None, recipient_type, person_id, snooze_for, reason="Voice command"

366 )

367 if cmd == CommandType.NORMAL and not subject:

368 engine.context.snoozer.register_snooze(cmd, GlobalTargetType.EVERYTHING, None, recipient_type, person_id, None)

369 whose = "yours" if person_id else "all"

370 about: str = ""

371 if subject:

372 matches: list[str] = (

373 [subject.target] if subject.target_type == QualifiedTargetType.CAMERA else engine.tag_matches(subject.target)

374 )

375 about = " " + voice.say("about", tag=subject.spoken) + (f" ({', '.join(matches)})" if matches else "")

376 if cmd == CommandType.NORMAL:

377 return voice.say(f"resumed_{whose}", about=about)

378 if cmd == CommandType.SILENCE:

379 return voice.say(f"silenced_everything_{whose}" if everything else f"silenced_{whose}", about=about)

380 until = dt_util.as_local(dt_util.now() + (snooze_for or engine.context.snoozer.snooze_period))

381 return voice.say(f"snoozed_{whose}", about=about, until=_clock(until, voice.language))

382

383

384def _clock(when: dt.datetime, language: str) -> str:

385 """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"""

386 if language == "it":

387 return when.strftime("%H:%M")

388 return f"{when.hour % 12 or 12}:{when.minute:02d}{'am' if when.hour < 12 else 'pm'}"

389

390

391def _last(engine: SupernotifyEngine, voice: Voice) -> str:

392 notification = engine.last_notification

393 if notification is None:

394 return voice.say("no_last")

395 when = dt_util.as_local(notification.created).strftime("%H:%M")

396 sent = [name for name, outcomes in notification.deliveries.items() if outcomes.get(EnvelopeOutcome.SUCCESS)]

397 if sent:

398 return voice.say("last_sent", when=when, message=notification.message, sent=", ".join(sent))

399 why: str = _why_not_sent(notification, voice)

400 return voice.say("last_not_sent", when=when, message=notification.message, why=f" {why}" if why else "")

401

402

403def _why_not_sent(notification: Notification, voice: Voice) -> str:

404 """Why nothing was sent, or nothing to add when there's no plain way to say it"""

405 if notification.failed:

406 return voice.say("why_error")

407 if not notification.deliveries and not notification._skip_reasons:

408 return voice.say("why_no_delivery")

409 for reason in notification._skip_reasons:

410 key: str = f"why_{reason.lower()}"

411 if voice.can_say(key):

412 return voice.say(key)

413 return ""