171 lines
7.5 KiB
Python
171 lines
7.5 KiB
Python
"""WebSocket commands to discover + adopt HA problem sensors as tasks.
|
|
|
|
``problem_sensors/discover`` proposes adoptable ``device_class: problem`` binary
|
|
sensors; ``problem_sensors/adopt`` turns an explicit selection into
|
|
sensor-triggered tasks (creating a maintenance object per device when needed).
|
|
Adoption is admin-gated write; discovery is read (it only lists candidates).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
import voluptuous as vol
|
|
from homeassistant.components import websocket_api
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
from ..const import CONF_OBJECT, DOMAIN, MAX_ENTITY_ID_LENGTH, MAX_ID_LENGTH, MAX_NAME_LENGTH
|
|
from ..helpers.permissions import require_write
|
|
from ..helpers.problem_sensors import (
|
|
build_problem_task,
|
|
discover_problem_sensors,
|
|
pop_stashed_config,
|
|
)
|
|
|
|
|
|
@websocket_api.websocket_command({vol.Required("type"): f"{DOMAIN}/problem_sensors/discover"})
|
|
@websocket_api.async_response
|
|
async def ws_discover_problem_sensors(
|
|
hass: HomeAssistant,
|
|
connection: websocket_api.ActiveConnection,
|
|
msg: dict[str, Any],
|
|
) -> None:
|
|
"""List adoptable problem sensors (not already watched by a task)."""
|
|
connection.send_result(msg["id"], {"sensors": discover_problem_sensors(hass)})
|
|
|
|
|
|
_SELECTION_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required("entity_id"): vol.All(str, vol.Length(max=MAX_ENTITY_ID_LENGTH)),
|
|
vol.Required("name"): vol.All(str, vol.Length(min=1, max=MAX_NAME_LENGTH)),
|
|
# Existing target object; omit to create a fresh object for this device.
|
|
vol.Optional("entry_id"): vol.All(str, vol.Length(max=MAX_ID_LENGTH)),
|
|
vol.Optional("object_name"): vol.All(str, vol.Length(min=1, max=MAX_NAME_LENGTH)),
|
|
vol.Optional("device_id"): vol.Any(vol.All(str, vol.Length(max=MAX_ID_LENGTH)), None),
|
|
# Spare part to link as consumes_parts on the adopted task (discovery's
|
|
# suggested_part_id) — completing the task then consumes/restocks it.
|
|
vol.Optional("part_id"): vol.Any(vol.All(str, vol.Length(max=MAX_ID_LENGTH)), None),
|
|
# Responsible HA user for the created task (the adopt dialog offers one
|
|
# picker applied to every selection). Wins over a stashed value.
|
|
vol.Optional("responsible_user_id"): vol.Any(vol.All(str, vol.Length(max=MAX_ID_LENGTH)), None),
|
|
}
|
|
)
|
|
|
|
|
|
@websocket_api.websocket_command(
|
|
{
|
|
vol.Required("type"): f"{DOMAIN}/problem_sensors/adopt",
|
|
vol.Required("selections"): vol.All([_SELECTION_SCHEMA], vol.Length(min=1, max=100)),
|
|
}
|
|
)
|
|
@require_write
|
|
@websocket_api.async_response
|
|
async def ws_adopt_problem_sensors(
|
|
hass: HomeAssistant,
|
|
connection: websocket_api.ActiveConnection,
|
|
msg: dict[str, Any],
|
|
) -> None:
|
|
"""Create a sensor-triggered task per selected problem sensor.
|
|
|
|
Each selection attaches to its ``entry_id`` (an existing object) or, when
|
|
omitted, to a freshly created object named ``object_name`` and bound to the
|
|
sensor's ``device_id`` — so a second adoption on the same device reuses it.
|
|
"""
|
|
from ..export import object_entries
|
|
from ..websocket.objects import async_create_object
|
|
from ..websocket.tasks_persist import async_persist_task
|
|
|
|
selections = msg["selections"]
|
|
tasks_created = 0
|
|
objects_created = 0
|
|
# Created tasks, in order — the dialog links "configure now" to the first.
|
|
created: list[dict[str, str]] = []
|
|
# Reuse an object created earlier in THIS batch for the same device, so two
|
|
# sensors on one device don't spawn two objects.
|
|
device_to_entry: dict[str, str] = {}
|
|
errors: list[dict[str, str]] = []
|
|
|
|
for sel in selections:
|
|
entity_id = sel["entity_id"]
|
|
entry_id = sel.get("entry_id")
|
|
device_id = sel.get("device_id")
|
|
created_entry_id: str | None = None # object created in THIS iteration
|
|
try:
|
|
if not entry_id and device_id and device_id in device_to_entry:
|
|
entry_id = device_to_entry[device_id]
|
|
if not entry_id:
|
|
entry_id = await async_create_object(
|
|
hass,
|
|
name=sel.get("object_name") or sel["name"],
|
|
ha_device_id=device_id or None,
|
|
)
|
|
created_entry_id = entry_id
|
|
objects_created += 1
|
|
if device_id:
|
|
device_to_entry[device_id] = entry_id
|
|
|
|
entry = hass.config_entries.async_get_entry(entry_id)
|
|
if entry is None or entry.domain != DOMAIN:
|
|
errors.append({"entity_id": entity_id, "reason": "target object not found"})
|
|
continue
|
|
|
|
task = build_problem_task(entity_id, sel["name"])
|
|
task_data = {
|
|
"id": uuid4().hex,
|
|
"object_id": entry.data.get(CONF_OBJECT, {}).get("id", ""),
|
|
"name": task["name"],
|
|
"type": task["task_type"],
|
|
"enabled": True,
|
|
"schedule": task["schedule"],
|
|
"trigger_config": task["trigger_config"],
|
|
}
|
|
# Un-adopt → re-adopt: restore (and consume) the notes and one-time
|
|
# setup the deleted predecessor task had accumulated for this
|
|
# sensor. Restored part links are re-validated below alongside the
|
|
# dialog's suggestion (the target object/parts may have changed).
|
|
from ..const import CONF_PARTS
|
|
from ..helpers.parts import sanitize_consumes_parts
|
|
|
|
stashed = pop_stashed_config(hass, entity_id) or {}
|
|
for field in ("notes", "responsible_user_id", "priority", "labels"):
|
|
if stashed.get(field):
|
|
task_data[field] = stashed[field]
|
|
# An explicit dialog pick wins over the stashed responsible user.
|
|
if sel.get("responsible_user_id"):
|
|
task_data["responsible_user_id"] = sel["responsible_user_id"]
|
|
# Link the suggested spare part — or, absent one, the stashed link —
|
|
# validated against the target object's parts (an unknown id is
|
|
# silently dropped, same as the task-CRUD path).
|
|
raw_links = (
|
|
[{"part_id": sel["part_id"], "quantity": 1}] if sel.get("part_id") else stashed.get("consumes_parts") or []
|
|
)
|
|
if raw_links:
|
|
links = sanitize_consumes_parts(raw_links, set(entry.data.get(CONF_PARTS) or {}))
|
|
if links:
|
|
task_data["consumes_parts"] = links
|
|
await async_persist_task(hass, entry, task_data)
|
|
tasks_created += 1
|
|
created.append({"entry_id": entry_id, "task_id": task_data["id"], "name": task_data["name"]})
|
|
except (ValueError, KeyError) as err:
|
|
errors.append({"entity_id": entity_id, "reason": str(err)})
|
|
# Roll back an object created in THIS iteration whose task failed —
|
|
# never leave an empty, task-less orphan object behind (and undo the
|
|
# count + device-reuse pointer so a later selection re-creates it).
|
|
if created_entry_id is not None:
|
|
objects_created -= 1
|
|
if device_id:
|
|
device_to_entry.pop(device_id, None)
|
|
if hass.config_entries.async_get_entry(created_entry_id) is not None:
|
|
await hass.config_entries.async_remove(created_entry_id)
|
|
|
|
result: dict[str, Any] = {
|
|
"tasks_created": tasks_created,
|
|
"objects_created": objects_created,
|
|
"created": created,
|
|
"total": len(object_entries(hass)),
|
|
}
|
|
if errors:
|
|
result["errors"] = errors
|
|
connection.send_result(msg["id"], result)
|