235 lines
8.3 KiB
Python
235 lines
8.3 KiB
Python
"""WebSocket endpoints for vacation mode (v1.2.0)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
from typing import Any
|
|
|
|
import voluptuous as vol
|
|
from homeassistant.components import websocket_api
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.util import dt as dt_util
|
|
|
|
from ..const import (
|
|
CONF_OBJECT,
|
|
CONF_TASKS,
|
|
CONF_VACATION_BUFFER_DAYS,
|
|
CONF_VACATION_ENABLED,
|
|
CONF_VACATION_END,
|
|
CONF_VACATION_EXEMPT_TASK_IDS,
|
|
CONF_VACATION_START,
|
|
DEFAULT_WARNING_DAYS,
|
|
DOMAIN,
|
|
MAX_ID_LENGTH,
|
|
)
|
|
from ..helpers.schedule import read_legacy_fields
|
|
from ..helpers.vacation import compute_preview, get_vacation_state
|
|
from . import _get_global_entry, _get_object_entries
|
|
|
|
|
|
def _state_payload(hass: HomeAssistant) -> dict[str, Any]:
|
|
"""Serialise the current VacationState for the wire."""
|
|
return get_vacation_state(hass).as_wire_dict()
|
|
|
|
|
|
@websocket_api.websocket_command({vol.Required("type"): f"{DOMAIN}/vacation/state"})
|
|
@websocket_api.async_response
|
|
async def ws_vacation_state(
|
|
hass: HomeAssistant,
|
|
connection: websocket_api.ActiveConnection,
|
|
msg: dict[str, Any],
|
|
) -> None:
|
|
"""Return the current vacation configuration + active flag."""
|
|
connection.send_result(msg["id"], _state_payload(hass))
|
|
|
|
|
|
@websocket_api.websocket_command(
|
|
{
|
|
vol.Required("type"): f"{DOMAIN}/vacation/update",
|
|
vol.Optional("enabled"): bool,
|
|
vol.Optional("start"): vol.Any(vol.All(str, vol.Length(max=10)), None),
|
|
vol.Optional("end"): vol.Any(vol.All(str, vol.Length(max=10)), None),
|
|
vol.Optional("buffer_days"): vol.All(int, vol.Range(min=0, max=14)),
|
|
vol.Optional("exempt_task_ids"): vol.All(
|
|
[vol.All(str, vol.Length(max=MAX_ID_LENGTH))],
|
|
vol.Length(max=2000),
|
|
),
|
|
}
|
|
)
|
|
@websocket_api.require_admin
|
|
@websocket_api.async_response
|
|
async def ws_vacation_update(
|
|
hass: HomeAssistant,
|
|
connection: websocket_api.ActiveConnection,
|
|
msg: dict[str, Any],
|
|
) -> None:
|
|
"""Patch vacation config on the global entry. Partial updates allowed."""
|
|
global_entry = _get_global_entry(hass)
|
|
if global_entry is None:
|
|
connection.send_error(msg["id"], "not_found", "Global entry not found")
|
|
return
|
|
|
|
options = dict(global_entry.options or global_entry.data)
|
|
|
|
if "enabled" in msg:
|
|
options[CONF_VACATION_ENABLED] = bool(msg["enabled"])
|
|
|
|
if "start" in msg:
|
|
if msg["start"] is None:
|
|
options[CONF_VACATION_START] = None
|
|
else:
|
|
try:
|
|
date.fromisoformat(msg["start"])
|
|
except (TypeError, ValueError):
|
|
connection.send_error(msg["id"], "invalid_date", "start must be YYYY-MM-DD")
|
|
return
|
|
options[CONF_VACATION_START] = msg["start"]
|
|
|
|
if "end" in msg:
|
|
if msg["end"] is None:
|
|
options[CONF_VACATION_END] = None
|
|
else:
|
|
try:
|
|
date.fromisoformat(msg["end"])
|
|
except (TypeError, ValueError):
|
|
connection.send_error(msg["id"], "invalid_date", "end must be YYYY-MM-DD")
|
|
return
|
|
options[CONF_VACATION_END] = msg["end"]
|
|
|
|
# End-vs-start sanity (only when both are present after the patch).
|
|
sd = options.get(CONF_VACATION_START)
|
|
ed = options.get(CONF_VACATION_END)
|
|
if sd and ed:
|
|
try:
|
|
if date.fromisoformat(ed) < date.fromisoformat(sd):
|
|
connection.send_error(msg["id"], "invalid_range", "end must be on or after start")
|
|
return
|
|
except (TypeError, ValueError):
|
|
pass # Already rejected above
|
|
|
|
if "buffer_days" in msg:
|
|
options[CONF_VACATION_BUFFER_DAYS] = int(msg["buffer_days"])
|
|
|
|
if "exempt_task_ids" in msg:
|
|
# Sanitise: strip + dedupe + cap.
|
|
seen: set[str] = set()
|
|
cleaned: list[str] = []
|
|
for raw in msg["exempt_task_ids"]:
|
|
if not isinstance(raw, str):
|
|
continue
|
|
v = raw.strip()
|
|
if not v or len(v) > MAX_ID_LENGTH or v in seen:
|
|
continue
|
|
seen.add(v)
|
|
cleaned.append(v)
|
|
if len(cleaned) >= 2000:
|
|
break
|
|
options[CONF_VACATION_EXEMPT_TASK_IDS] = cleaned
|
|
|
|
hass.config_entries.async_update_entry(global_entry, options=options)
|
|
connection.send_result(msg["id"], _state_payload(hass))
|
|
|
|
|
|
@websocket_api.websocket_command({vol.Required("type"): f"{DOMAIN}/vacation/preview"})
|
|
@websocket_api.async_response
|
|
async def ws_vacation_preview(
|
|
hass: HomeAssistant,
|
|
connection: websocket_api.ActiveConnection,
|
|
msg: dict[str, Any],
|
|
) -> None:
|
|
"""Return the projected impact of the currently-configured vacation.
|
|
|
|
Even works when the toggle is off — useful for the "Preview impact"
|
|
button before the user enables it. A vacation without start/end returns
|
|
an empty list.
|
|
"""
|
|
state = get_vacation_state(hass)
|
|
|
|
# If the user is previewing without enabling, still compute against the
|
|
# currently-stored dates. Caller is responsible for passing dates via
|
|
# /update first if they want a live preview during date entry.
|
|
if state.start is None or state.end is None:
|
|
connection.send_result(msg["id"], {"rows": [], "window_end": None})
|
|
return
|
|
|
|
# Build the flat task list expected by compute_preview.
|
|
tasks: list[dict[str, Any]] = []
|
|
for entry in _get_object_entries(hass):
|
|
obj_data = entry.data.get(CONF_OBJECT, {})
|
|
obj_name = obj_data.get("name", "")
|
|
tasks_data = entry.data.get(CONF_TASKS, {})
|
|
|
|
# Merge dynamic store fields (last_performed, etc.) when available.
|
|
rd = getattr(entry, "runtime_data", None)
|
|
store = getattr(rd, "store", None) if rd else None
|
|
merged: dict[str, dict[str, Any]] = store.merge_all_tasks(tasks_data) if store is not None else dict(tasks_data)
|
|
|
|
for task_id, task_data in merged.items():
|
|
sched = read_legacy_fields(task_data)
|
|
tasks.append(
|
|
{
|
|
"task_id": task_id,
|
|
"entry_id": entry.entry_id,
|
|
"object_name": obj_name,
|
|
"task_name": task_data.get("name", ""),
|
|
"schedule_type": sched["schedule_type"],
|
|
"interval_days": sched["interval_days"],
|
|
"interval_unit": sched["interval_unit"],
|
|
# Nested schedule so the preview can project calendar kinds.
|
|
"schedule": task_data.get("schedule"),
|
|
"warning_days": task_data.get("warning_days", DEFAULT_WARNING_DAYS),
|
|
"last_performed": task_data.get("last_performed"),
|
|
"created_at": task_data.get("created_at"),
|
|
"enabled": task_data.get("enabled", True),
|
|
}
|
|
)
|
|
|
|
rows = compute_preview(state, tasks)
|
|
connection.send_result(
|
|
msg["id"],
|
|
{
|
|
"rows": rows,
|
|
"window_end": state.window_end.isoformat() if state.window_end else None,
|
|
},
|
|
)
|
|
|
|
|
|
@websocket_api.websocket_command({vol.Required("type"): f"{DOMAIN}/vacation/end_now"})
|
|
@websocket_api.require_admin
|
|
@websocket_api.async_response
|
|
async def ws_vacation_end_now(
|
|
hass: HomeAssistant,
|
|
connection: websocket_api.ActiveConnection,
|
|
msg: dict[str, Any],
|
|
) -> None:
|
|
"""Disable vacation mode immediately, preserve the date config for reuse."""
|
|
global_entry = _get_global_entry(hass)
|
|
if global_entry is None:
|
|
connection.send_error(msg["id"], "not_found", "Global entry not found")
|
|
return
|
|
|
|
options = dict(global_entry.options or global_entry.data)
|
|
options[CONF_VACATION_ENABLED] = False
|
|
# Optionally clamp end-date to today so the historical record reflects when
|
|
# the user actually returned. Use HA's configured timezone — the user's
|
|
# "today" should match what their dashboard shows, not the server's UTC.
|
|
today = dt_util.now().date()
|
|
sd = options.get(CONF_VACATION_START)
|
|
if sd:
|
|
try:
|
|
if date.fromisoformat(sd) <= today:
|
|
options[CONF_VACATION_END] = today.isoformat()
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
hass.config_entries.async_update_entry(global_entry, options=options)
|
|
connection.send_result(msg["id"], _state_payload(hass))
|
|
|
|
|
|
__all__ = [
|
|
"ws_vacation_end_now",
|
|
"ws_vacation_preview",
|
|
"ws_vacation_state",
|
|
"ws_vacation_update",
|
|
]
|