803 lines
32 KiB
Python
803 lines
32 KiB
Python
"""Assist intents — query + complete maintenance tasks by voice / LLM agents.
|
|
|
|
Registered through HA's integration intent platform (``async_setup_intents``,
|
|
picked up automatically when the ``intent`` component loads — same mechanism as
|
|
shopping_list):
|
|
|
|
* ``MaintenanceSupporterListTasks`` — *"what maintenance is due?"* Speaks a
|
|
snapshot of actionable tasks, optionally filtered by status.
|
|
* ``MaintenanceSupporterCompleteTask`` — *"complete the oil change"* — fuzzy-
|
|
matches a task by its spoken name (the object name counts too, so "oil change
|
|
on the car" works) and records a REAL completion through the coordinator —
|
|
history, rotation, part consumption and on-complete actions all fire.
|
|
* ``MaintenanceSupporterTaskInstructions`` — *"how do I descale the coffee
|
|
machine?"* — answers STRICTLY from what is stored on the task (notes,
|
|
checklist, linked documents incl. per-task page hints, required spare parts,
|
|
documentation link). When nothing is stored it says so and asks whether the
|
|
user wants general, non-verified advice — grounded by design, never invented.
|
|
* ``MaintenanceSupporterTaskDue`` — *"when is the oil change due?"*
|
|
* ``MaintenanceSupporterSnoozeTask`` — *"snooze the oil change"* — suppresses
|
|
the task's reminders for the configured snooze duration.
|
|
* ``MaintenanceSupporterPartStock`` — *"how many water filters do we have?"*
|
|
|
|
LLM-based Assist pipelines expose every registered intent handler as a tool
|
|
automatically (``helpers/llm``), in any language — no setup needed. The classic
|
|
sentence-matching agent additionally needs the copy-paste sentence files shipped
|
|
under ``assist/custom_sentences/`` (see FEATURES → Voice & Assist).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
import voluptuous as vol
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
from homeassistant.helpers import config_validation as cv
|
|
from homeassistant.helpers import intent
|
|
|
|
from .const import CONF_OBJECT, CONF_TASKS, DOMAIN, GLOBAL_UNIQUE_ID
|
|
|
|
INTENT_LIST_TASKS = "MaintenanceSupporterListTasks"
|
|
INTENT_COMPLETE_TASK = "MaintenanceSupporterCompleteTask"
|
|
INTENT_TASK_INSTRUCTIONS = "MaintenanceSupporterTaskInstructions"
|
|
INTENT_TASK_DUE = "MaintenanceSupporterTaskDue"
|
|
INTENT_SNOOZE_TASK = "MaintenanceSupporterSnoozeTask"
|
|
INTENT_PART_STOCK = "MaintenanceSupporterPartStock"
|
|
INTENT_POSTPONE_TASK = "MaintenanceSupporterPostponeTask"
|
|
INTENT_SKIP_TASK = "MaintenanceSupporterSkipTask"
|
|
|
|
_ACTIONABLE = ("due_soon", "overdue", "triggered")
|
|
|
|
# Spoken responses live in assist_sentences/responses/<lang>.json — 38 keys
|
|
# across 22 languages is far too much to read past on the way to the
|
|
# handlers, and a translator should not have to edit Python. The loader and
|
|
# the English-per-key fallback live in helpers/intent_speech.
|
|
|
|
|
|
def _sp(key: str, language: str | None, **fmt: Any) -> str:
|
|
"""Spoken text for *key*, in the requesting language.
|
|
|
|
Thin seam on purpose: every handler already calls ``_sp``, so the move
|
|
to per-language files needed no changes at the call sites.
|
|
"""
|
|
from .helpers.intent_speech import speak
|
|
|
|
return speak(key, language, **fmt)
|
|
|
|
|
|
def _task_snapshot(hass: HomeAssistant) -> list[dict[str, Any]]:
|
|
"""Live snapshot of every active (non-archived) task across all objects.
|
|
|
|
Same shape/source as the ``list_tasks`` service: the coordinator's computed
|
|
payload, so status/next_due reflect the Store, not stale entry data.
|
|
"""
|
|
tasks: list[dict[str, Any]] = []
|
|
for ce in hass.config_entries.async_entries(DOMAIN):
|
|
if ce.unique_id == GLOBAL_UNIQUE_ID:
|
|
continue
|
|
rd = getattr(ce, "runtime_data", None)
|
|
coordinator = getattr(rd, "coordinator", None) if rd else None
|
|
if coordinator is None or not coordinator.data:
|
|
continue
|
|
obj = ce.data.get(CONF_OBJECT, {})
|
|
object_name = obj.get("name", ce.title)
|
|
for task_id, task in coordinator.data.get(CONF_TASKS, {}).items():
|
|
status = str(task.get("_status", ""))
|
|
if status == "archived":
|
|
continue
|
|
tasks.append(
|
|
{
|
|
"entry_id": ce.entry_id,
|
|
"task_id": task_id,
|
|
"object_name": object_name,
|
|
# The object's room, so a satellite can answer for where it
|
|
# is standing rather than for the whole house.
|
|
"area_id": obj.get("area_id") or None,
|
|
"name": str(task.get("name") or ""),
|
|
# Whose turn it is — for a rotation this is the current duty.
|
|
"responsible_user_id": task.get("responsible_user_id") or None,
|
|
"status": status,
|
|
"days_until_due": task.get("_days_until_due"),
|
|
"next_due": task.get("_next_due"),
|
|
}
|
|
)
|
|
return tasks
|
|
|
|
|
|
def _asking_area(intent_obj: intent.Intent) -> str | None:
|
|
"""Which area the request came from, or None if we cannot tell.
|
|
|
|
Home Assistant hands the handler both the device that captured the speech
|
|
and (for voice satellites) the satellite entity. Either can carry the area:
|
|
an entity's own area override wins, otherwise its device's.
|
|
"""
|
|
from homeassistant.helpers import device_registry as dr
|
|
from homeassistant.helpers import entity_registry as er
|
|
|
|
hass = intent_obj.hass
|
|
|
|
satellite_id = getattr(intent_obj, "satellite_id", None)
|
|
if satellite_id:
|
|
entry = er.async_get(hass).async_get(satellite_id)
|
|
if entry is not None:
|
|
if entry.area_id:
|
|
return entry.area_id
|
|
if entry.device_id:
|
|
device = dr.async_get(hass).async_get(entry.device_id)
|
|
if device is not None and device.area_id:
|
|
return device.area_id
|
|
|
|
if intent_obj.device_id:
|
|
device = dr.async_get(hass).async_get(intent_obj.device_id)
|
|
if device is not None and device.area_id:
|
|
return device.area_id
|
|
|
|
return None
|
|
|
|
|
|
def _asking_user(intent_obj: intent.Intent) -> str | None:
|
|
"""The Home Assistant user who spoke, when the pipeline knows one."""
|
|
context = getattr(intent_obj, "context", None)
|
|
return getattr(context, "user_id", None) if context else None
|
|
|
|
|
|
def _match_tasks(query: str, snapshot: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Fuzzy-match a spoken name against tasks (object name counts too).
|
|
|
|
An exact task-name match wins outright; otherwise every ≥2-char query token
|
|
must appear in "task name + object name" (case-insensitive substring).
|
|
"""
|
|
q = query.strip().lower()
|
|
if not q:
|
|
return []
|
|
exact = [t for t in snapshot if t["name"].lower() == q]
|
|
if len(exact) == 1:
|
|
return exact
|
|
tokens = [tok for tok in re.split(r"[^a-z0-9äöüß]+", q) if len(tok) >= 2]
|
|
if not tokens:
|
|
return exact
|
|
matches = []
|
|
for t in snapshot:
|
|
hay = f"{t['name']} {t['object_name']}".lower()
|
|
if all(tok in hay for tok in tokens):
|
|
matches.append(t)
|
|
return exact or matches
|
|
|
|
|
|
def _describe(task: dict[str, Any], language: str | None) -> str:
|
|
days = task.get("days_until_due")
|
|
if task["status"] == "triggered":
|
|
desc = _sp("st_triggered", language)
|
|
elif isinstance(days, int) and days < 0:
|
|
# "1 days overdue" was shipped in English and German alike; a single
|
|
# string cannot inflect, so the singular is its own key. Several
|
|
# languages need it far more than English does — Czech, Russian and
|
|
# Ukrainian govern the noun's case by the numeral.
|
|
key = "st_overdue_one" if days == -1 else "st_overdue"
|
|
desc = _sp(key, language, days=-days)
|
|
elif days == 0:
|
|
desc = _sp("st_due_today", language)
|
|
elif isinstance(days, int):
|
|
key = "st_due_in_one" if days == 1 else "st_due_in"
|
|
desc = _sp(key, language, days=days)
|
|
else:
|
|
desc = task["status"]
|
|
return f"{_sp('item_on', language, task=task['name'], object=task['object_name'])} ({desc})"
|
|
|
|
|
|
def _resolve_single(
|
|
intent_obj: intent.Intent, name: str, snapshot: list[dict[str, Any]]
|
|
) -> tuple[dict[str, Any] | None, intent.IntentResponse | None]:
|
|
"""Match a spoken name to exactly ONE snapshot row.
|
|
|
|
Returns ``(row, None)`` on success, ``(None, error_response)`` otherwise —
|
|
the shared not-found / ambiguity contract of every name-taking intent.
|
|
"""
|
|
lang = intent_obj.language
|
|
response = intent_obj.create_response()
|
|
matches = _match_tasks(name, snapshot)
|
|
if not matches:
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.NO_VALID_TARGETS,
|
|
_sp("not_found", lang, name=name),
|
|
)
|
|
return None, response
|
|
if len(matches) > 1:
|
|
# Before giving up, let the room decide. "Complete the filter change"
|
|
# spoken at the utility-room satellite means the one in the utility
|
|
# room — and getting this right matters more than convenience, because
|
|
# a voice completion writes real history through the coordinator.
|
|
area = _asking_area(intent_obj)
|
|
if area:
|
|
local = [t for t in matches if t.get("area_id") == area]
|
|
if len(local) == 1:
|
|
return local[0], None
|
|
candidates = ", ".join(
|
|
_sp("item_on", lang, task=t["name"], object=t["object_name"]) for t in matches[:4]
|
|
)
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.NO_VALID_TARGETS,
|
|
_sp("ambiguous", lang, candidates=candidates),
|
|
)
|
|
return None, response
|
|
return matches[0], None
|
|
|
|
|
|
async def async_setup_intents(hass: HomeAssistant) -> None:
|
|
"""Register the Maintenance Supporter intents."""
|
|
# Read the response texts now, in the executor: a handler answering a
|
|
# spoken question must not block the event loop on disk.
|
|
from .helpers.intent_speech import async_load
|
|
|
|
await async_load(hass)
|
|
|
|
intent.async_register(hass, ListTasksIntent())
|
|
intent.async_register(hass, CompleteTaskIntent())
|
|
intent.async_register(hass, TaskInstructionsIntent())
|
|
intent.async_register(hass, TaskDueIntent())
|
|
intent.async_register(hass, SnoozeTaskIntent())
|
|
intent.async_register(hass, PartStockIntent())
|
|
intent.async_register(hass, PostponeTaskIntent())
|
|
intent.async_register(hass, SkipTaskIntent())
|
|
|
|
|
|
class ListTasksIntent(intent.IntentHandler):
|
|
"""Speak which maintenance tasks need attention (optionally by status)."""
|
|
|
|
intent_type = INTENT_LIST_TASKS
|
|
description = (
|
|
"Lists the user's home-maintenance tasks that need attention "
|
|
"(overdue, due soon or sensor-triggered), optionally filtered by a "
|
|
"status. Use for questions like 'what maintenance is due?'. Set scope "
|
|
"to 'mine' for the tasks assigned to the person asking (including "
|
|
"whose turn it is on a rotating chore), or 'here' for the tasks "
|
|
"belonging to the room the request came from."
|
|
)
|
|
slot_schema = {
|
|
vol.Optional("status"): vol.In(["ok", "due_soon", "overdue", "triggered"]),
|
|
vol.Optional("scope"): vol.In(["all", "mine", "here"]),
|
|
}
|
|
|
|
async def async_handle(self, intent_obj: intent.Intent) -> intent.IntentResponse:
|
|
"""Handle the intent."""
|
|
slots = self.async_validate_slots(intent_obj.slots)
|
|
wanted = slots.get("status", {}).get("value")
|
|
scope = slots.get("scope", {}).get("value") or "all"
|
|
statuses = (wanted,) if wanted else _ACTIONABLE
|
|
tasks = [t for t in _task_snapshot(intent_obj.hass) if t["status"] in statuses]
|
|
|
|
response = intent_obj.create_response()
|
|
lang = intent_obj.language
|
|
|
|
# A scope we cannot resolve is answered honestly rather than silently
|
|
# widened: "everything in the house" is a plausible-sounding wrong
|
|
# answer to "what needs doing in here?".
|
|
if scope == "mine":
|
|
user_id = _asking_user(intent_obj)
|
|
if not user_id:
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.NO_VALID_TARGETS,
|
|
_sp("unknown_user", lang),
|
|
)
|
|
return response
|
|
tasks = [t for t in tasks if t.get("responsible_user_id") == user_id]
|
|
elif scope == "here":
|
|
area = _asking_area(intent_obj)
|
|
if not area:
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.NO_VALID_TARGETS,
|
|
_sp("unknown_area", lang),
|
|
)
|
|
return response
|
|
tasks = [t for t in tasks if t.get("area_id") == area]
|
|
|
|
# Most urgent first: overdue (most days) → due today → due soon.
|
|
tasks.sort(key=lambda t: (t["days_until_due"] is None, t["days_until_due"] or 0))
|
|
|
|
if not tasks:
|
|
empty = {"mine": "none_due_mine", "here": "none_due_here"}.get(scope, "none_due")
|
|
response.async_set_speech(_sp(empty, lang))
|
|
return response
|
|
items = ", ".join(_describe(t, lang) for t in tasks[:8])
|
|
key = "task_due_one" if len(tasks) == 1 else "tasks_due"
|
|
response.async_set_speech(_sp(key, lang, count=len(tasks), items=items))
|
|
return response
|
|
|
|
|
|
class CompleteTaskIntent(intent.IntentHandler):
|
|
"""Complete a maintenance task by its spoken name."""
|
|
|
|
intent_type = INTENT_COMPLETE_TASK
|
|
description = (
|
|
"Marks a home-maintenance task as completed, matched by its name "
|
|
"(the object/appliance name may be included, e.g. 'oil change on the car'). "
|
|
"Records a real completion including history."
|
|
)
|
|
slot_schema = {vol.Required("name"): cv.string}
|
|
|
|
async def async_handle(self, intent_obj: intent.Intent) -> intent.IntentResponse:
|
|
"""Handle the intent."""
|
|
hass = intent_obj.hass
|
|
slots = self.async_validate_slots(intent_obj.slots)
|
|
name = slots["name"]["value"].strip()
|
|
lang = intent_obj.language
|
|
|
|
target, err = _resolve_single(intent_obj, name, _task_snapshot(hass))
|
|
if err is not None:
|
|
return err
|
|
assert target is not None
|
|
response = intent_obj.create_response()
|
|
entry = hass.config_entries.async_get_entry(target["entry_id"])
|
|
rd = getattr(entry, "runtime_data", None)
|
|
coordinator = getattr(rd, "coordinator", None) if rd else None
|
|
if coordinator is None:
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
|
|
_sp("not_found", lang, name=name),
|
|
)
|
|
return response
|
|
|
|
# Honour the completion window (earliest_completion_days) like the WS
|
|
# and To-do paths do — voice must not bypass the contract.
|
|
from .websocket.tasks_actions import _completion_blocked
|
|
|
|
if _completion_blocked(rd, target["task_id"]):
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
|
|
_sp("too_early", lang, task=target["name"]),
|
|
)
|
|
return response
|
|
|
|
try:
|
|
await coordinator.complete_maintenance(
|
|
task_id=target["task_id"], completed_by="assist", unattended=True
|
|
)
|
|
except ServiceValidationError as err:
|
|
# The task demands details voice cannot capture (a photo, a cost).
|
|
# Say so plainly instead of failing with a generic error.
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
|
|
str(err),
|
|
)
|
|
return response
|
|
response.async_set_speech(
|
|
_sp("completed", lang, task=target["name"], object=target["object_name"])
|
|
)
|
|
return response
|
|
|
|
|
|
class TaskInstructionsIntent(intent.IntentHandler):
|
|
"""Speak the STORED guidance for a task — grounded by design (roadmap)."""
|
|
|
|
intent_type = INTENT_TASK_INSTRUCTIONS
|
|
description = (
|
|
"Returns the guidance STORED on a home-maintenance task: notes, checklist "
|
|
"steps, linked manuals/documents (with a page hint), required spare parts "
|
|
"(with storage location and stock) and whether a documentation link is on "
|
|
"file. Use for 'how do I …' questions about maintenance tasks. IMPORTANT: "
|
|
"this returns only verified, user-stored information — if it reports that "
|
|
"nothing is stored, tell the user so and ask whether they want general "
|
|
"advice before providing any; never present invented steps as the stored "
|
|
"procedure."
|
|
)
|
|
slot_schema = {vol.Required("name"): cv.string}
|
|
|
|
async def async_handle(self, intent_obj: intent.Intent) -> intent.IntentResponse:
|
|
"""Handle the intent."""
|
|
hass = intent_obj.hass
|
|
slots = self.async_validate_slots(intent_obj.slots)
|
|
name = slots["name"]["value"].strip()
|
|
lang = intent_obj.language
|
|
|
|
target, err = _resolve_single(intent_obj, name, _task_snapshot(hass))
|
|
if err is not None:
|
|
return err
|
|
assert target is not None
|
|
response = intent_obj.create_response()
|
|
|
|
entry = hass.config_entries.async_get_entry(target["entry_id"])
|
|
rd = getattr(entry, "runtime_data", None) if entry else None
|
|
# Static fields (notes/checklist/url/consumes_parts) come from the entry
|
|
# record — the system of record; the coordinator payload carries only a
|
|
# computed subset (consumes_parts, for one, is not in it).
|
|
task: dict[str, Any] = {}
|
|
if entry is not None:
|
|
task = entry.data.get(CONF_TASKS, {}).get(target["task_id"], {}) or {}
|
|
|
|
segments: list[str] = []
|
|
|
|
notes = task.get("notes")
|
|
if isinstance(notes, str) and notes.strip():
|
|
trimmed = notes.strip()
|
|
if len(trimmed) > 240:
|
|
trimmed = trimmed[:237] + "…"
|
|
segments.append(_sp("guide_notes", lang, notes=trimmed))
|
|
|
|
checklist = [s for s in (task.get("checklist") or []) if isinstance(s, str) and s.strip()]
|
|
if checklist:
|
|
segments.append(
|
|
_sp("guide_checklist", lang, count=len(checklist), steps="; ".join(checklist[:8]))
|
|
)
|
|
|
|
# Documents linked to THIS task, with the per-task page hint when set.
|
|
from . import DOCUMENT_STORE_KEY
|
|
from .const import CONF_PARTS
|
|
|
|
doc_store = hass.data.get(DOMAIN, {}).get(DOCUMENT_STORE_KEY)
|
|
if doc_store is not None and entry is not None:
|
|
object_id = entry.data.get(CONF_OBJECT, {}).get("id", "")
|
|
for doc in doc_store.for_object(object_id):
|
|
if target["task_id"] not in (doc.get("task_ids") or []):
|
|
continue
|
|
title = doc.get("title") or doc.get("filename") or doc.get("url") or "document"
|
|
page = (doc.get("task_pages") or {}).get(target["task_id"])
|
|
if page:
|
|
segments.append(_sp("guide_doc_page", lang, title=title, page=page))
|
|
else:
|
|
segments.append(_sp("guide_doc", lang, title=title))
|
|
|
|
if isinstance(task.get("documentation_url"), str) and task["documentation_url"].strip():
|
|
segments.append(_sp("guide_url", lang))
|
|
|
|
# Required spare parts with storage location + live stock.
|
|
parts = (entry.data.get(CONF_PARTS) or {}) if entry is not None else {}
|
|
store = getattr(rd, "store", None) if rd else None
|
|
for link in task.get("consumes_parts") or []:
|
|
part = parts.get(link.get("part_id")) if isinstance(link, dict) else None
|
|
if not isinstance(part, dict):
|
|
continue
|
|
extras: list[str] = []
|
|
if part.get("storage_location"):
|
|
extras.append(_sp("guide_part_loc", lang, loc=part["storage_location"]))
|
|
stock = store.get_part_stock(link["part_id"]) if store is not None else None
|
|
if stock is not None:
|
|
extras.append(_sp("guide_part_stock", lang, stock=stock))
|
|
segments.append(
|
|
_sp(
|
|
"guide_part",
|
|
lang,
|
|
qty=link.get("quantity", 1),
|
|
part=part.get("name") or "part",
|
|
extras=f" ({', '.join(extras)})" if extras else "",
|
|
)
|
|
)
|
|
|
|
if not segments:
|
|
# Grounded contract: nothing stored → say so and ASK before any
|
|
# general advice — the LLM relays the question instead of inventing.
|
|
response.async_set_speech(
|
|
_sp("guide_none", lang, task=target["name"], object=target["object_name"])
|
|
)
|
|
return response
|
|
|
|
response.async_set_speech(
|
|
_sp(
|
|
"guide_header",
|
|
lang,
|
|
task=target["name"],
|
|
object=target["object_name"],
|
|
segments="; ".join(segments),
|
|
)
|
|
)
|
|
return response
|
|
|
|
|
|
class TaskDueIntent(intent.IntentHandler):
|
|
"""Answer when a single task is due."""
|
|
|
|
intent_type = INTENT_TASK_DUE
|
|
description = (
|
|
"Tells when a single home-maintenance task is due, matched by its name "
|
|
"(the object/appliance name may be included). Use for questions like "
|
|
"'when is the oil change due?'"
|
|
)
|
|
slot_schema = {vol.Required("name"): cv.string}
|
|
|
|
async def async_handle(self, intent_obj: intent.Intent) -> intent.IntentResponse:
|
|
"""Handle the intent."""
|
|
slots = self.async_validate_slots(intent_obj.slots)
|
|
name = slots["name"]["value"].strip()
|
|
lang = intent_obj.language
|
|
|
|
target, err = _resolve_single(intent_obj, name, _task_snapshot(intent_obj.hass))
|
|
if err is not None:
|
|
return err
|
|
assert target is not None
|
|
response = intent_obj.create_response()
|
|
|
|
speech = _describe(target, lang) + "."
|
|
next_due = target.get("next_due")
|
|
if isinstance(next_due, str) and next_due:
|
|
speech += _sp("due_date_suffix", lang, date=next_due.split("T")[0])
|
|
response.async_set_speech(speech)
|
|
return response
|
|
|
|
|
|
class SnoozeTaskIntent(intent.IntentHandler):
|
|
"""Snooze a task's reminders for the configured snooze duration."""
|
|
|
|
intent_type = INTENT_SNOOZE_TASK
|
|
description = (
|
|
"Snoozes (mutes) the reminder notifications of a home-maintenance task "
|
|
"for the configured snooze duration. Does NOT change the task's schedule "
|
|
"or complete it."
|
|
)
|
|
slot_schema = {vol.Required("name"): cv.string}
|
|
|
|
async def async_handle(self, intent_obj: intent.Intent) -> intent.IntentResponse:
|
|
"""Handle the intent."""
|
|
hass = intent_obj.hass
|
|
slots = self.async_validate_slots(intent_obj.slots)
|
|
name = slots["name"]["value"].strip()
|
|
lang = intent_obj.language
|
|
|
|
target, err = _resolve_single(intent_obj, name, _task_snapshot(hass))
|
|
if err is not None:
|
|
return err
|
|
assert target is not None
|
|
response = intent_obj.create_response()
|
|
|
|
from . import NOTIFICATION_MANAGER_KEY
|
|
from .const import CONF_SNOOZE_DURATION_HOURS, DEFAULT_SNOOZE_DURATION_HOURS
|
|
from .helpers.global_options import get_global_options
|
|
|
|
nm = hass.data.get(DOMAIN, {}).get(NOTIFICATION_MANAGER_KEY)
|
|
if nm is None:
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
|
|
_sp("snooze_unavailable", lang),
|
|
)
|
|
return response
|
|
|
|
nm.snooze_task(target["entry_id"], target["task_id"])
|
|
hours = get_global_options(hass).get(CONF_SNOOZE_DURATION_HOURS, DEFAULT_SNOOZE_DURATION_HOURS)
|
|
if isinstance(hours, float) and hours.is_integer():
|
|
hours = int(hours) # "4 hours", not "4.0 hours"
|
|
response.async_set_speech(
|
|
_sp("snoozed", lang, task=target["name"], object=target["object_name"], hours=hours)
|
|
)
|
|
return response
|
|
|
|
|
|
def _part_snapshot(hass: HomeAssistant) -> list[dict[str, Any]]:
|
|
"""Every spare part across all objects, in the _match_tasks row shape
|
|
(``name`` + ``object_name``) so the same fuzzy matcher applies."""
|
|
from .const import CONF_PARTS
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
for ce in hass.config_entries.async_entries(DOMAIN):
|
|
if ce.unique_id == GLOBAL_UNIQUE_ID:
|
|
continue
|
|
object_name = ce.data.get(CONF_OBJECT, {}).get("name", ce.title)
|
|
rd = getattr(ce, "runtime_data", None)
|
|
store = getattr(rd, "store", None) if rd else None
|
|
for part_id, part in (ce.data.get(CONF_PARTS) or {}).items():
|
|
if not isinstance(part, dict):
|
|
continue
|
|
rows.append(
|
|
{
|
|
"name": str(part.get("name") or ""),
|
|
"object_name": object_name,
|
|
"storage_location": part.get("storage_location"),
|
|
"reorder_threshold": part.get("reorder_threshold"),
|
|
"stock": store.get_part_stock(part_id) if store is not None else None,
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
class PartStockIntent(intent.IntentHandler):
|
|
"""Answer how many of a spare part are in stock."""
|
|
|
|
intent_type = INTENT_PART_STOCK
|
|
description = (
|
|
"Tells how many of a spare part / consumable are in stock (with the "
|
|
"storage location), matched by the part's name. Use for questions like "
|
|
"'how many water filters do we have left?'"
|
|
)
|
|
slot_schema = {vol.Required("name"): cv.string}
|
|
|
|
async def async_handle(self, intent_obj: intent.Intent) -> intent.IntentResponse:
|
|
"""Handle the intent."""
|
|
slots = self.async_validate_slots(intent_obj.slots)
|
|
name = slots["name"]["value"].strip()
|
|
lang = intent_obj.language
|
|
response = intent_obj.create_response()
|
|
|
|
matches = _match_tasks(name, _part_snapshot(intent_obj.hass))
|
|
if not matches:
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.NO_VALID_TARGETS,
|
|
_sp("part_not_found", lang, name=name),
|
|
)
|
|
return response
|
|
if len(matches) > 1:
|
|
candidates = ", ".join(
|
|
_sp("item_on", lang, task=p["name"], object=p["object_name"]) for p in matches[:4]
|
|
)
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.NO_VALID_TARGETS,
|
|
_sp("ambiguous", lang, candidates=candidates),
|
|
)
|
|
return response
|
|
|
|
part = matches[0]
|
|
if part["stock"] is None:
|
|
response.async_set_speech(_sp("stock_untracked", lang, part=part["name"]))
|
|
return response
|
|
|
|
loc = _sp("stock_loc", lang, loc=part["storage_location"]) if part.get("storage_location") else ""
|
|
threshold = part.get("reorder_threshold")
|
|
low = (
|
|
_sp("stock_low", lang)
|
|
if isinstance(threshold, int) and part["stock"] <= threshold
|
|
else ""
|
|
)
|
|
response.async_set_speech(
|
|
_sp("stock_line", lang, stock=part["stock"], part=part["name"], loc=loc, low=low)
|
|
)
|
|
return response
|
|
|
|
|
|
def _resolve_coordinator(
|
|
intent_obj: intent.Intent, target: dict[str, Any], name: str
|
|
) -> tuple[Any, intent.IntentResponse | None]:
|
|
"""The coordinator behind a matched task, or a spoken failure."""
|
|
entry = intent_obj.hass.config_entries.async_get_entry(target["entry_id"])
|
|
rd = getattr(entry, "runtime_data", None)
|
|
coordinator = getattr(rd, "coordinator", None) if rd else None
|
|
if coordinator is None:
|
|
response = intent_obj.create_response()
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
|
|
_sp("not_found", intent_obj.language, name=name),
|
|
)
|
|
return None, response
|
|
return coordinator, None
|
|
|
|
|
|
class PostponeTaskIntent(intent.IntentHandler):
|
|
"""Defer just this occurrence of a task, spoken."""
|
|
|
|
intent_type = INTENT_POSTPONE_TASK
|
|
description = (
|
|
"Postpones the CURRENT occurrence of a home-maintenance task to a later "
|
|
"date without completing it; the recurring cadence is untouched. Give "
|
|
"either days (how many days to push it back) or date (YYYY-MM-DD). Use "
|
|
"for 'postpone the oil change by a week'. This is not the same as "
|
|
"snoozing, which only mutes reminders."
|
|
)
|
|
slot_schema = {
|
|
vol.Required("name"): cv.string,
|
|
vol.Optional("days"): vol.Coerce(int),
|
|
vol.Optional("date"): cv.string,
|
|
}
|
|
|
|
async def async_handle(self, intent_obj: intent.Intent) -> intent.IntentResponse:
|
|
"""Handle the intent."""
|
|
from datetime import date as date_cls
|
|
from datetime import timedelta
|
|
|
|
from homeassistant.util import dt as dt_util
|
|
|
|
hass = intent_obj.hass
|
|
slots = self.async_validate_slots(intent_obj.slots)
|
|
name = str(slots["name"]["value"]).strip()
|
|
lang = intent_obj.language
|
|
|
|
target, err = _resolve_single(intent_obj, name, _task_snapshot(hass))
|
|
if err is not None:
|
|
return err
|
|
assert target is not None
|
|
response = intent_obj.create_response()
|
|
|
|
today = dt_util.now().date()
|
|
spoken_date = slots.get("date", {}).get("value")
|
|
days = slots.get("days", {}).get("value")
|
|
|
|
if spoken_date:
|
|
try:
|
|
until = date_cls.fromisoformat(str(spoken_date))
|
|
except ValueError:
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
|
|
_sp("postpone_needs_when", lang),
|
|
)
|
|
return response
|
|
elif days is not None:
|
|
# Counted from the due date, or from today when that has already
|
|
# passed: "postpone by three days" on a task that went overdue last
|
|
# month must not land on a date that is still in the past.
|
|
base = today
|
|
next_due = target.get("next_due")
|
|
if next_due:
|
|
try:
|
|
base = max(date_cls.fromisoformat(str(next_due)[:10]), today)
|
|
except ValueError:
|
|
base = today
|
|
until = base + timedelta(days=int(days))
|
|
else:
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
|
|
_sp("postpone_needs_when", lang),
|
|
)
|
|
return response
|
|
|
|
if until <= today:
|
|
response.async_set_error(
|
|
intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
|
|
_sp("postpone_past", lang, task=target["name"]),
|
|
)
|
|
return response
|
|
|
|
coordinator, err = _resolve_coordinator(intent_obj, target, name)
|
|
if err is not None:
|
|
return err
|
|
|
|
await coordinator.async_postpone_task(target["task_id"], until)
|
|
response.async_set_speech(
|
|
_sp(
|
|
"postponed",
|
|
lang,
|
|
task=target["name"],
|
|
object=target["object_name"],
|
|
date=until.isoformat(),
|
|
)
|
|
)
|
|
return response
|
|
|
|
|
|
class SkipTaskIntent(intent.IntentHandler):
|
|
"""Skip the current cycle of a task, spoken."""
|
|
|
|
intent_type = INTENT_SKIP_TASK
|
|
description = (
|
|
"Skips the CURRENT cycle of a home-maintenance task: the task is not "
|
|
"recorded as done, and the schedule moves on to the next occurrence. "
|
|
"Use for 'skip the lawn mowing this time'."
|
|
)
|
|
slot_schema = {vol.Required("name"): cv.string}
|
|
|
|
async def async_handle(self, intent_obj: intent.Intent) -> intent.IntentResponse:
|
|
"""Handle the intent."""
|
|
hass = intent_obj.hass
|
|
slots = self.async_validate_slots(intent_obj.slots)
|
|
name = str(slots["name"]["value"]).strip()
|
|
lang = intent_obj.language
|
|
|
|
target, err = _resolve_single(intent_obj, name, _task_snapshot(hass))
|
|
if err is not None:
|
|
return err
|
|
assert target is not None
|
|
|
|
coordinator, err = _resolve_coordinator(intent_obj, target, name)
|
|
if err is not None:
|
|
return err
|
|
|
|
await coordinator.skip_maintenance(target["task_id"])
|
|
|
|
# Read the new due date back so the answer says what actually happened
|
|
# rather than just acknowledging the command.
|
|
fresh = next(
|
|
(
|
|
t
|
|
for t in _task_snapshot(hass)
|
|
if t["entry_id"] == target["entry_id"] and t["task_id"] == target["task_id"]
|
|
),
|
|
None,
|
|
)
|
|
due = str((fresh or {}).get("next_due") or "")[:10]
|
|
response = intent_obj.create_response()
|
|
response.async_set_speech(
|
|
_sp(
|
|
"skipped",
|
|
lang,
|
|
task=target["name"],
|
|
object=target["object_name"],
|
|
date=due or "?",
|
|
)
|
|
)
|
|
return response
|