441 lines
17 KiB
Python
441 lines
17 KiB
Python
"""Persistent storage for dynamic maintenance state.
|
|
|
|
Frequently-changing data (history, last_performed, trigger runtime,
|
|
adaptive config) lives here instead of in ConfigEntry.data to avoid
|
|
excessive writes to ``core.config_entries`` which wears out SD cards.
|
|
|
|
Static task configuration (name, type, interval, trigger thresholds, etc.)
|
|
remains in ConfigEntry.data and is only written on explicit user edits.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.storage import Store
|
|
|
|
from .const import CONF_TASKS, DEFAULT_MAX_HISTORY_ENTRIES, DOMAIN
|
|
from .helpers.parts import round_qty
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
STORE_VERSION = 1
|
|
STORE_KEY_PREFIX = f"{DOMAIN}"
|
|
STORE_SAVE_DELAY = 60.0 # seconds — debounce writes to protect SD cards
|
|
|
|
# Fields that migrate from ConfigEntry.data to Store
|
|
_DYNAMIC_TASK_FIELDS = ("last_performed", "last_planned_due", "due_override", "history", "adaptive_config")
|
|
|
|
# Flat trigger_config keys that should move to Store trigger_runtime
|
|
_LEGACY_TRIGGER_RUNTIME_KEYS = (
|
|
"trigger_baseline_value",
|
|
"trigger_accumulated_seconds",
|
|
"trigger_change_count",
|
|
"trigger_on_since",
|
|
)
|
|
|
|
|
|
class MaintenanceStore:
|
|
"""Manages dynamic state storage for a single maintenance object.
|
|
|
|
Each maintenance object config entry gets its own store file at
|
|
``.storage/maintenance_supporter.<entry_id>``.
|
|
"""
|
|
|
|
def __init__(self, hass: HomeAssistant, entry_id: str) -> None:
|
|
"""Initialize the store."""
|
|
self._store: Store[dict[str, Any]] = Store(hass, STORE_VERSION, f"{STORE_KEY_PREFIX}.{entry_id}")
|
|
self._data: dict[str, Any] = {"version": STORE_VERSION, "tasks": {}}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Load / Save / Remove
|
|
# ------------------------------------------------------------------
|
|
|
|
async def async_load(self) -> dict[str, Any] | None:
|
|
"""Load store data. Returns None if no store file exists."""
|
|
raw = await self._store.async_load()
|
|
if raw is not None:
|
|
self._data = self._sanitize_loaded(raw)
|
|
return raw
|
|
|
|
@staticmethod
|
|
def _sanitize_loaded(raw: Any) -> dict[str, Any]:
|
|
"""Coerce loaded data to the expected structure (journey I2).
|
|
|
|
HA core already quarantines syntactically corrupt store files, but a
|
|
hand-edited or partially-written file can still be valid JSON with the
|
|
wrong SHAPE — and every accessor here assumes dicts and lists. Broken
|
|
parts must degrade to "never performed" instead of bricking the entry
|
|
at boot with an AttributeError.
|
|
"""
|
|
if not isinstance(raw, dict):
|
|
_LOGGER.warning(
|
|
"Store payload is %s, expected dict — resetting dynamic state",
|
|
type(raw).__name__,
|
|
)
|
|
return {"version": STORE_VERSION, "tasks": {}}
|
|
data = dict(raw)
|
|
tasks = data.get("tasks")
|
|
if not isinstance(tasks, dict):
|
|
if tasks is not None:
|
|
_LOGGER.warning(
|
|
"Store 'tasks' is %s, expected dict — resetting dynamic state",
|
|
type(tasks).__name__,
|
|
)
|
|
data["tasks"] = {}
|
|
return data
|
|
clean: dict[str, Any] = {}
|
|
for tid, state in tasks.items():
|
|
if not isinstance(state, dict):
|
|
_LOGGER.warning(
|
|
"Dropping malformed store state for task %s (%s)",
|
|
tid,
|
|
type(state).__name__,
|
|
)
|
|
continue
|
|
state = dict(state)
|
|
if "history" in state and not isinstance(state["history"], list):
|
|
_LOGGER.warning("Resetting malformed history for task %s", tid)
|
|
state["history"] = []
|
|
for key in ("trigger_runtime", "adaptive_config"):
|
|
if key in state and not isinstance(state[key], dict):
|
|
_LOGGER.warning("Dropping malformed %s for task %s", key, tid)
|
|
del state[key]
|
|
clean[tid] = state
|
|
data["tasks"] = clean
|
|
# Spare-part stock counts ({part_id: {"stock": int}}) — same shape
|
|
# discipline as tasks: a malformed section degrades to "untracked".
|
|
parts = data.get("parts")
|
|
if parts is not None and not isinstance(parts, dict):
|
|
_LOGGER.warning("Store 'parts' is %s, expected dict — resetting stock state", type(parts).__name__)
|
|
data["parts"] = {}
|
|
elif isinstance(parts, dict):
|
|
data["parts"] = {
|
|
pid: dict(state)
|
|
for pid, state in parts.items()
|
|
if isinstance(state, dict)
|
|
}
|
|
return data
|
|
|
|
def async_delay_save(self) -> None:
|
|
"""Schedule a debounced save (coalesces rapid writes)."""
|
|
self._store.async_delay_save(self._get_data, STORE_SAVE_DELAY)
|
|
|
|
async def async_save(self) -> None:
|
|
"""Immediate save (for migration or critical writes)."""
|
|
await self._store.async_save(self._data)
|
|
|
|
async def async_remove(self) -> None:
|
|
"""Delete the store file (called when config entry is removed)."""
|
|
await self._store.async_remove()
|
|
|
|
def _get_data(self) -> dict[str, Any]:
|
|
"""Return current data for save callback."""
|
|
return self._data
|
|
|
|
# ------------------------------------------------------------------
|
|
# Task State Accessors
|
|
# ------------------------------------------------------------------
|
|
|
|
def _ensure_task(self, task_id: str) -> dict[str, Any]:
|
|
"""Ensure a task state dict exists and return it."""
|
|
tasks: dict[str, Any] = self._data.setdefault("tasks", {})
|
|
if task_id not in tasks:
|
|
tasks[task_id] = {}
|
|
result: dict[str, Any] = tasks[task_id]
|
|
return result
|
|
|
|
def get_task_state(self, task_id: str) -> dict[str, Any]:
|
|
"""Return the dynamic state for a task (or empty dict)."""
|
|
tasks: dict[str, Any] = self._data.get("tasks", {})
|
|
result: dict[str, Any] = tasks.get(task_id, {})
|
|
return result
|
|
|
|
def init_task(self, task_id: str, last_performed: str | None = None) -> None:
|
|
"""Initialize state for a newly created task."""
|
|
state = self._ensure_task(task_id)
|
|
if last_performed is not None:
|
|
state["last_performed"] = last_performed
|
|
state.setdefault("history", [])
|
|
|
|
def remove_task(self, task_id: str) -> None:
|
|
"""Remove all state for a deleted task."""
|
|
self._data.get("tasks", {}).pop(task_id, None)
|
|
|
|
def prune_orphans(self, valid_task_ids: set[str]) -> int:
|
|
"""Drop task states whose ids are not in *valid_task_ids*.
|
|
|
|
Boot-time reconciliation for the entry.data <-> Store split (journey
|
|
I1): a crash between the ConfigEntry write and the Store save on a
|
|
task deletion leaves the dynamic state orphaned forever — the merge
|
|
only iterates entry.data tasks, so orphans are invisible but
|
|
accumulate. Returns the number of pruned states; the caller decides
|
|
whether to save.
|
|
"""
|
|
tasks = self._data.get("tasks", {})
|
|
orphaned = [tid for tid in tasks if tid not in valid_task_ids]
|
|
for tid in orphaned:
|
|
del tasks[tid]
|
|
return len(orphaned)
|
|
|
|
# --- spare-part stock (mutable count; static part defs live in entry.data) ---
|
|
|
|
def get_part_stock(self, part_id: str) -> float | None:
|
|
"""On-hand count for a part, or None when inventory isn't tracked."""
|
|
state = self._data.get("parts", {}).get(part_id, {})
|
|
stock = state.get("stock")
|
|
return stock if isinstance(stock, (int, float)) and not isinstance(stock, bool) else None
|
|
|
|
def all_part_stocks(self) -> dict[str, float | None]:
|
|
"""{part_id: stock} for every tracked part."""
|
|
return {pid: self.get_part_stock(pid) for pid in self._data.get("parts", {})}
|
|
|
|
def set_part_stock(self, part_id: str, stock: float | None) -> None:
|
|
"""Set (or untrack with None) a part's on-hand count, clamped to
|
|
[0, MAX_PART_STOCK]. The WS schemas cap each INPUT at MAX_PART_STOCK,
|
|
but additive paths (restock delta / buy-task completion) could push
|
|
the SUM past it — this chokepoint clamp keeps every writer honest."""
|
|
from .helpers.parts import MAX_PART_STOCK
|
|
|
|
parts: dict[str, Any] = self._data.setdefault("parts", {})
|
|
if stock is None:
|
|
parts.pop(part_id, None)
|
|
else:
|
|
parts.setdefault(part_id, {})["stock"] = min(max(0, round_qty(stock)), MAX_PART_STOCK)
|
|
|
|
def remove_part(self, part_id: str) -> None:
|
|
"""Drop a deleted part's stock state."""
|
|
self._data.get("parts", {}).pop(part_id, None)
|
|
|
|
def prune_part_orphans(self, valid_part_ids: set[str]) -> int:
|
|
"""Drop stock states whose part ids no longer exist (journey-I1 twin)."""
|
|
parts = self._data.get("parts", {})
|
|
orphaned = [pid for pid in parts if pid not in valid_part_ids]
|
|
for pid in orphaned:
|
|
del parts[pid]
|
|
return len(orphaned)
|
|
|
|
# --- last_performed ---
|
|
|
|
def get_last_performed(self, task_id: str) -> str | None:
|
|
"""Return last_performed date string (or None)."""
|
|
return self.get_task_state(task_id).get("last_performed")
|
|
|
|
def set_last_performed(self, task_id: str, date_str: str) -> None:
|
|
"""Set last_performed date string."""
|
|
self._ensure_task(task_id)["last_performed"] = date_str
|
|
|
|
# --- history ---
|
|
|
|
def get_history(self, task_id: str) -> list[dict[str, Any]]:
|
|
"""Return the history list for a task."""
|
|
result: list[dict[str, Any]] = self.get_task_state(task_id).get("history", [])
|
|
return result
|
|
|
|
def append_history(self, task_id: str, entry: dict[str, Any]) -> None:
|
|
"""Append a history entry and trim to max size."""
|
|
state = self._ensure_task(task_id)
|
|
history = state.setdefault("history", [])
|
|
history.append(entry)
|
|
if len(history) > DEFAULT_MAX_HISTORY_ENTRIES:
|
|
state["history"] = history[-DEFAULT_MAX_HISTORY_ENTRIES:]
|
|
|
|
def set_history(self, task_id: str, history: list[dict[str, Any]]) -> None:
|
|
"""Replace the entire history list."""
|
|
self._ensure_task(task_id)["history"] = history
|
|
|
|
# --- adaptive_config ---
|
|
|
|
def get_adaptive_config(self, task_id: str) -> dict[str, Any] | None:
|
|
"""Return the adaptive config (or None)."""
|
|
return self.get_task_state(task_id).get("adaptive_config")
|
|
|
|
def set_adaptive_config(self, task_id: str, config: dict[str, Any]) -> None:
|
|
"""Set the adaptive config."""
|
|
self._ensure_task(task_id)["adaptive_config"] = config
|
|
|
|
# --- trigger_runtime (per-entity) ---
|
|
|
|
def get_trigger_runtime(self, task_id: str, entity_id: str | None = None) -> dict[str, Any]:
|
|
"""Return trigger runtime data.
|
|
|
|
If *entity_id* is given, return per-entity runtime data.
|
|
Otherwise, return the whole trigger_runtime dict.
|
|
"""
|
|
state = self.get_task_state(task_id)
|
|
runtime: dict[str, Any] = state.get("trigger_runtime", {})
|
|
if entity_id is not None:
|
|
result: dict[str, Any] = runtime.get(entity_id, {})
|
|
return result
|
|
return runtime
|
|
|
|
def set_trigger_runtime(
|
|
self,
|
|
task_id: str,
|
|
entity_id: str,
|
|
data: dict[str, Any],
|
|
) -> None:
|
|
"""Set trigger runtime data for a specific entity."""
|
|
state = self._ensure_task(task_id)
|
|
runtime = state.setdefault("trigger_runtime", {})
|
|
entity_state = runtime.setdefault(entity_id, {})
|
|
entity_state.update(data)
|
|
|
|
def clear_trigger_runtime(self, task_id: str) -> None:
|
|
"""Clear all trigger runtime data for a task."""
|
|
state = self.get_task_state(task_id)
|
|
state.pop("trigger_runtime", None)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Merge / Split Helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def merge_task_data(self, task_id: str, static_data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Merge static ConfigEntry data with dynamic Store data.
|
|
|
|
Returns a unified task dict compatible with ``MaintenanceTask.from_dict``.
|
|
"""
|
|
result = dict(static_data)
|
|
state = self.get_task_state(task_id)
|
|
|
|
# Overlay dynamic fields
|
|
for field in _DYNAMIC_TASK_FIELDS:
|
|
if field in state:
|
|
result[field] = state[field]
|
|
|
|
# Merge trigger_runtime into trigger_config._trigger_state
|
|
trigger_runtime = state.get("trigger_runtime")
|
|
if not trigger_runtime:
|
|
# Fall back to legacy flat runtime keys if no per-entity data yet
|
|
trigger_runtime = state.get("trigger_runtime_legacy")
|
|
if trigger_runtime and "trigger_config" in result:
|
|
tc = dict(result["trigger_config"])
|
|
# Restructure compound flat keys (_compound_N_key) into nested
|
|
# conditions list that CompoundTrigger.async_setup() expects
|
|
compound_keys = [k for k in trigger_runtime if k.startswith("_compound_")]
|
|
if compound_keys:
|
|
non_compound = {k: v for k, v in trigger_runtime.items() if not k.startswith("_compound_")}
|
|
conditions: dict[int, dict[str, Any]] = {}
|
|
for k in compound_keys:
|
|
# Key format: _compound_<idx> or _compound_<idx>_<entity_id>
|
|
# Split into: ['', 'compound', '<idx>'] or ['', 'compound', '<idx>', '<entity_id>']
|
|
parts = k.split("_", 3)
|
|
if len(parts) < 3:
|
|
continue
|
|
try:
|
|
idx = int(parts[2])
|
|
except ValueError:
|
|
continue
|
|
if len(parts) == 3:
|
|
# No entity_id suffix — merge directly into condition dict
|
|
conditions.setdefault(idx, {}).update(trigger_runtime[k])
|
|
else:
|
|
sub_key = parts[3]
|
|
conditions.setdefault(idx, {})[sub_key] = trigger_runtime[k]
|
|
non_compound["conditions"] = [
|
|
conditions.get(i, {}) for i in range(max(conditions.keys()) + 1 if conditions else 0)
|
|
]
|
|
tc["_trigger_state"] = non_compound
|
|
else:
|
|
tc["_trigger_state"] = trigger_runtime
|
|
result["trigger_config"] = tc
|
|
|
|
return result
|
|
|
|
def merge_all_tasks(self, static_tasks: dict[str, Any]) -> dict[str, Any]:
|
|
"""Merge all tasks' static data with Store dynamic data."""
|
|
return {task_id: self.merge_task_data(task_id, task_data) for task_id, task_data in static_tasks.items()}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Migration
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def extract_dynamic_from_task(
|
|
task_data: dict[str, Any],
|
|
) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
"""Split a task dict into (static, dynamic) parts.
|
|
|
|
Modifies neither input; returns new dicts.
|
|
"""
|
|
static = dict(task_data)
|
|
dynamic: dict[str, Any] = {}
|
|
|
|
# Extract top-level dynamic fields
|
|
for field in _DYNAMIC_TASK_FIELDS:
|
|
if field in static:
|
|
dynamic[field] = static.pop(field)
|
|
|
|
# Extract trigger runtime from trigger_config
|
|
if "trigger_config" in static:
|
|
tc = dict(static["trigger_config"])
|
|
runtime = tc.pop("_trigger_state", None)
|
|
|
|
# Also extract legacy flat runtime keys into trigger_runtime
|
|
flat_runtime: dict[str, Any] = {}
|
|
for flat_key in _LEGACY_TRIGGER_RUNTIME_KEYS:
|
|
val = tc.pop(flat_key, None)
|
|
if val is not None:
|
|
flat_runtime[flat_key] = val
|
|
|
|
static["trigger_config"] = tc
|
|
|
|
if runtime:
|
|
dynamic["trigger_runtime"] = runtime
|
|
elif flat_runtime:
|
|
# Legacy: flat keys exist but no _trigger_state yet.
|
|
# Store them as-is; the trigger objects will migrate to
|
|
# per-entity format on first setup.
|
|
dynamic["trigger_runtime_legacy"] = flat_runtime
|
|
|
|
return static, dynamic
|
|
|
|
|
|
async def async_migrate_to_store(
|
|
hass: HomeAssistant,
|
|
entry_id: str,
|
|
entry_data: Mapping[str, Any],
|
|
store: MaintenanceStore,
|
|
) -> dict[str, Any] | Mapping[str, Any]:
|
|
"""Migrate dynamic state from ConfigEntry.data into a Store.
|
|
|
|
Returns the cleaned static-only entry data (for updating the ConfigEntry).
|
|
The Store is saved immediately.
|
|
|
|
This is idempotent: if the Store already has data, nothing happens
|
|
and the original *entry_data* is returned unchanged.
|
|
"""
|
|
existing = await store.async_load()
|
|
if existing is not None:
|
|
return entry_data # Already migrated
|
|
|
|
tasks_data = entry_data.get(CONF_TASKS, {})
|
|
static_tasks: dict[str, Any] = {}
|
|
|
|
for task_id, task_data in tasks_data.items():
|
|
static, dynamic = extract_dynamic_from_task(task_data)
|
|
static_tasks[task_id] = static
|
|
|
|
# Populate store
|
|
store.init_task(task_id)
|
|
state = store._ensure_task(task_id)
|
|
state.update(dynamic)
|
|
|
|
# Save store immediately (must succeed before we strip ConfigEntry)
|
|
await store.async_save()
|
|
|
|
# Build cleaned entry data
|
|
new_data = dict(entry_data)
|
|
new_data[CONF_TASKS] = static_tasks
|
|
|
|
_LOGGER.info(
|
|
"Migrated %d tasks from ConfigEntry to Store for entry %s",
|
|
len(tasks_data),
|
|
entry_id,
|
|
)
|
|
|
|
return new_data
|