# WashData - Home Assistant integration for appliance cycle monitoring via smart plugs.
# Copyright (C) 2026 Lukas Bandura
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
"""Profile storage and matching logic for WashData."""
from __future__ import annotations
import dataclasses
import hashlib
import logging
import math
import os
import re
import threading
import uuid
from datetime import datetime, timedelta
from typing import Any, TypeAlias, cast
import numpy as np
from homeassistant.core import HomeAssistant
from homeassistant.helpers.storage import Store
from homeassistant.util import dt as dt_util
from .const import (
CLUSTER_RESAMPLE_N,
CLUSTER_SHAPE_SIMILARITY_THRESHOLD,
GROUP_MIN_COHESION,
MAINTENANCE_EVENT_TYPES,
MAINTENANCE_RECENT_SUPPRESS_DAYS,
MATCH_AMBIGUITY_MARGIN,
PHASE_CONSISTENCY_MIN_CYCLES,
PHASE_PROFILE_MIN_CYCLES,
PHASE_HEAT_CV_WARN,
PHASE_HEAT_OCC_MIXED_LO,
PHASE_HEAT_OCC_MIXED_HI,
REFERENCE_PROFILE_CURVE_POINTS,
SHAPE_DRIFT_MIN_CYCLES,
SHAPE_DRIFT_RESAMPLE_N,
SHAPE_DRIFT_THRESHOLD,
SMART_TERM_LANDSCAPE_RATIO,
SMART_TERM_LANDSCAPE_MIN_SHAPE,
STORAGE_KEY,
STORAGE_VERSION,
DEFAULT_MAX_PAST_CYCLES,
DEFAULT_MAX_FULL_TRACES_PER_PROFILE,
DEFAULT_MAX_FULL_TRACES_UNLABELED,
DEFAULT_DTW_BANDWIDTH,
)
from .features import compute_signature
from .signal_processing import resample_uniform, resample_adaptive, Segment, integrate_wh, energy_gap_threshold_s
from . import analysis
from .time_utils import (
migrate_power_data_to_offsets,
power_data_to_offsets,
)
from .phase_catalog import (
DEFAULT_PHASES_BY_DEVICE,
_builtin_phase_id,
get_builtin_phase_by_id,
merge_phase_catalog,
normalize_phase_name,
)
from .phase_segmenter import phase_matching_live_supported, phase_model_for, segment_cycle
from .phase_match import (
build_phase_profile,
match_phase_profiles,
phase_eta,
phase_profile_from_dict,
phase_profile_to_dict,
)
from .log_utils import DeviceLoggerAdapter
_LOGGER = logging.getLogger(__name__)
# Absolute peak-power floor below which a cycle trace is considered a
# mis-capture (e.g. the power sensor wasn't reporting), regardless of profile.
# Used together with a relative (10% of median peak) test in envelope rebuild
# and reference selection so degenerate cycles never become the matching template.
_DEGENERATE_POWER_FLOOR = 15.0 # watts
JSONDict: TypeAlias = dict[str, Any]
CycleDict: TypeAlias = dict[str, Any]
def _is_recorded_cycle(cycle: dict[str, Any]) -> bool:
"""True when a cycle was produced by the manual recorder.
Two independent signals, because the explicit marker did not always exist:
1. **Explicit meta marker** (recent recorder builds): ``meta.source ==
"recorder"`` and, equivalently, ``meta.original_samples`` (the pre-trim
sample count). (In exported diagnostics ``meta.source`` is redacted, but
the real stored value is ``"recorder"`` — the migration runs against the
real store, not the export.)
2. **Structural signature** (old recordings, ``meta`` is ``None``): the
recorder builds the cycle dict directly (``ws_process_recording``) and
never sets ``max_power`` or ``termination_reason``, whereas *every*
auto-detected cycle goes through ``CycleDetector._finish_cycle`` which
stamps ``max_power`` unconditionally (in every version — ``termination_reason``
was added later, so old *auto* cycles can lack it but never lack
``max_power``). So a **completed** cycle missing *both* fields is a
recording that predates the meta marker. Verified against real exports:
this recovers old recordings that carry only ``meta: None`` — which the
marker-only check silently missed.
"""
meta = cycle.get("meta")
if isinstance(meta, dict) and (
meta.get("source") == "recorder" or "original_samples" in meta
):
return True
return (
cycle.get("status") == "completed"
and "max_power" not in cycle
and "termination_reason" not in cycle
)
def _flag_recorded_cycles_golden(cycles: list[dict[str, Any]]) -> int:
"""Mark manually-recorded cycles as golden references (recorded == golden).
A recorded cycle (see :func:`_is_recorded_cycle`) is, by definition, a
hand-picked clean example, so it becomes the profile's golden reference. The
flag lives in the single ``ml_review`` dict (no duplicate ``recorded`` field).
Idempotent: already-golden cycles are skipped, so it is safe to re-run from the
migration and the diagnostics processing trigger.
Returns the number of cycles newly flagged.
"""
flagged = 0
for cycle in cycles:
if not isinstance(cycle, dict):
continue
review = cycle.get("ml_review")
if isinstance(review, dict) and review.get("golden"):
continue
if not _is_recorded_cycle(cycle):
continue
review = dict(review) if isinstance(review, dict) else {}
review["golden"] = True
if not review.get("quality"):
review["quality"] = "good" # recorded cycles are clean by definition
if not review.get("reviewed_at"):
review["reviewed_at"] = dt_util.now().isoformat()
cycle["ml_review"] = review
flagged += 1
return flagged
def _empty_ranking() -> list[dict[str, Any]]:
"""Typed default factory for ranking entries."""
return []
def _parse_start_dt(value: Any) -> datetime | None:
"""Parse a start_time value (ISO string or numeric timestamp) into a datetime.
Handles the case where legacy cycles stored numeric unix timestamps instead
of ISO-formatted strings, which dt_util.parse_datetime cannot handle.
"""
if isinstance(value, datetime):
return value
if isinstance(value, (int, float)) and not isinstance(value, bool):
try:
return datetime.fromtimestamp(float(value), tz=dt_util.UTC)
except (OSError, OverflowError, ValueError):
return None
if isinstance(value, str) and value:
parsed = dt_util.parse_datetime(value)
if parsed is not None:
return parsed
try:
return datetime.fromtimestamp(float(value), tz=dt_util.UTC)
except (TypeError, ValueError):
return None
return None
def _parse_maintenance_dt(value: Any) -> datetime | None:
"""Parse a maintenance-log date (ISO date or datetime) into an aware datetime.
Accepts both full ISO datetimes (as written by ``dt_util.now().isoformat()``)
and bare ISO dates (``"2026-07-01"``) that the user may enter. Naive results are
localised so recency/comparison math stays timezone-consistent. Never raises.
"""
dt = _parse_start_dt(value)
if dt is None and isinstance(value, str) and value:
try:
dt = datetime.fromisoformat(value)
except ValueError:
dt = None
if dt is not None and dt.tzinfo is None:
dt = dt.replace(tzinfo=dt_util.DEFAULT_TIME_ZONE or dt_util.UTC)
return dt
def _empty_debug_details() -> dict[str, Any]:
"""Typed default factory for debug details."""
return {}
def _value_to_timestamp(value: Any) -> float | None:
"""Parse supported datetime-like values into unix seconds."""
if isinstance(value, datetime):
return value.timestamp()
if isinstance(value, (int, float)):
return float(value)
if isinstance(value, str) and value:
parsed = dt_util.parse_datetime(value)
if parsed is not None:
return parsed.timestamp()
try:
return float(value)
except (TypeError, ValueError):
return None
return None
def profile_sort_key(name: str) -> tuple[int, int, str]:
"""Sort key for profile names: numeric-prefixed first (by number), then alphabetically."""
match = re.match(r'^(\d+)', name)
if match:
return (0, int(match.group(1)), name)
return (1, 0, name)
def trim_zero_power_data(
data: list[list[float]],
threshold: float = 0.5
) -> list[list[float]]:
"""Trim leading/trailing zero/near-zero power readings from stored data.
Args:
data: List of [offset, power] pairs
threshold: Power values <= this are considered "zero"
Returns:
Trimmed list with leading/trailing zeros removed
"""
if not data:
return data
# Find first non-zero reading
start_idx = 0
for i, point in enumerate(data):
if point[1] > threshold:
start_idx = i
break
else:
# All readings are zero - keep at least one
return data[:1] if data else []
# Find last non-zero reading
end_idx = len(data) - 1
for i in range(len(data) - 1, -1, -1):
if data[i][1] > threshold:
end_idx = i
break
# Return trimmed slice (inclusive of end)
return data[start_idx:end_idx + 1]
def filter_duration_outliers(durations: list[float]) -> list[float]:
"""Return a robust duration set with extreme outliers removed.
Uses Tukey IQR fences for normal spread and falls back to a MAD-based
filter when IQR collapses (common when most cycles are identical).
"""
if len(durations) < 4:
return durations
arr = np.array(durations, dtype=float)
q1 = float(np.percentile(arr, 25))
q3 = float(np.percentile(arr, 75))
iqr = q3 - q1
if iqr > 0:
lower = max(60.0, q1 - 1.5 * iqr)
upper = q3 + 1.5 * iqr
filtered = arr[(arr >= lower) & (arr <= upper)]
else:
# Degenerate spread (e.g. many identical durations); keep values
# close to median and drop only extreme anomalies.
median = float(np.median(arr))
abs_dev = np.abs(arr - median)
mad = float(np.median(abs_dev))
if mad == 0:
tol = max(300.0, median * 0.15)
filtered = arr[np.abs(arr - median) <= tol]
else:
robust_z = abs_dev / (1.4826 * mad)
filtered = arr[robust_z <= 3.5]
# Guardrail: do not over-filter sparse datasets.
if len(filtered) >= max(3, int(len(arr) * 0.6)):
return filtered.astype(float).tolist()
return durations
@dataclasses.dataclass
class MatchResult:
"""Result of a profile matching attempt."""
best_profile: str | None
confidence: float
expected_duration: float
matched_phase: str | None
candidates: list[dict[str, Any]]
is_ambiguous: bool
ambiguity_margin: float
ranking: list[dict[str, Any]] = dataclasses.field(default_factory=_empty_ranking)
debug_details: dict[str, Any] = dataclasses.field(default_factory=_empty_debug_details)
is_confident_mismatch: bool = False
mismatch_reason: str | None = None
is_prefix_ambiguous: bool = False
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary with JSON-serializable types, excluding heavy arrays."""
def _convert(obj: Any) -> Any:
if isinstance(obj, np.generic):
return cast(np.generic, obj).item()
if isinstance(obj, np.ndarray):
# Fallback for unexpected arrays: just describe shape
arr = cast(np.ndarray[Any, Any], obj)
return f""
if isinstance(obj, dict):
# Exclude huge raw data arrays from cycle candidates
obj_dict = cast(dict[str, Any], obj)
return {
k: _convert(v)
for k, v in obj_dict.items()
if k not in ("current", "sample", "metrics", "warping_path")
}
if isinstance(obj, list):
obj_list = cast(list[Any], obj)
return [_convert(v) for v in obj_list]
if dataclasses.is_dataclass(obj):
return text_type_safe_asdict(obj)
return obj
def text_type_safe_asdict(d_obj: Any) -> dict[str, Any]:
return {f.name: _convert(getattr(d_obj, f.name)) for f in dataclasses.fields(d_obj)}
return text_type_safe_asdict(self)
def _safe_file_size_kb(path: str) -> float:
"""Return the size of ``path`` in KiB, or 0.0 if it cannot be read.
Runs blocking os.path calls; intended to be offloaded to the executor.
"""
try:
return os.path.getsize(path) / 1024 if os.path.exists(path) else 0.0
except OSError:
return 0.0
def decompress_power_data(cycle: CycleDict) -> list[tuple[float, float]]:
"""Return power data as ``[(offset_seconds, power), ...]`` for a cycle.
Handles both the current canonical ``[offset_float, power]`` format and the
legacy ``(iso_str, power)`` format transparently. Returns an empty list if
data is missing or malformed.
"""
raw = cycle.get("power_data", [])
if not isinstance(raw, list) or not raw:
return []
start_time_raw = cycle.get("start_time")
start_time_iso: str | None = start_time_raw if isinstance(start_time_raw, str) and start_time_raw else None
if isinstance(start_time_raw, datetime):
start_time_iso = start_time_raw.isoformat()
offsets = power_data_to_offsets(cast(list[list[Any] | tuple[Any, ...]], raw), start_time_iso)
return [(float(o), float(p)) for o, p in offsets]
def earliest_sustained_quiet_offset(
cycles: list[CycleDict],
stop_threshold_w: float,
min_quiet_span_s: float,
min_clean_cycles: int,
) -> float | None:
"""Smallest elapsed offset (s) at which any *completed* cycle first shows a
sustained (>= ``min_quiet_span_s``) near-zero (< ``stop_threshold_w``) span.
This is the device's learned "we have never legitimately been quiet before
this many seconds into a cycle" baseline, used by the opt-in terminal-drop
detector: a hard cliff-to-~0 that begins *earlier* than this is anomalous and
almost certainly a real stop rather than a soak pause.
Only ``completed`` cycles seed the baseline - interrupted / force-stopped /
terminal-drop cycles are exactly the anomalies we are trying to catch, so
including them would poison the baseline and suppress detection. Returns
``None`` when there are fewer than ``min_clean_cycles`` completed cycles with
usable traces, so the caller keeps the proven slow end-detection.
Using the strict *minimum* (rather than a higher percentile) is deliberately
conservative: one cycle that happened to go quiet early only lowers the
baseline, making the detector fire *less* often - never more.
"""
if not cycles:
return None
earliest: float | None = None
clean = 0
for cycle in cycles:
if cycle.get("status") != "completed":
continue
points = decompress_power_data(cycle)
if not points:
continue
clean += 1
span_start: float | None = None
for offset, power in points:
if power < stop_threshold_w:
if span_start is None:
span_start = offset
elif (offset - span_start) >= min_quiet_span_s:
if earliest is None or span_start < earliest:
earliest = span_start
break
else:
span_start = None
if clean < min_clean_cycles:
return None
return earliest
def device_active_peak_range(
cycles: list[CycleDict], min_clean_cycles: int
) -> tuple[float, float] | None:
"""(min_peak, max_peak) across the device's **completed** cycles.
The peak of a cycle is its maximum power. This is the device's learned
"power levels we have produced before" band, used by the terminal-drop
familiarity gate: a running cycle peaking outside this range (widened by a
tolerance) is drawing power unlike anything in its history and may be a new
program, so an early drop on it is deferred rather than assumed terminal.
Returns ``None`` below ``min_clean_cycles`` completed cycles with usable
traces (same guard as the quiet-offset baseline)."""
peaks: list[float] = []
for cycle in cycles:
if cycle.get("status") != "completed":
continue
points = decompress_power_data(cycle)
if not points:
continue
peaks.append(max(p for _, p in points))
if len(peaks) < min_clean_cycles:
return None
return (min(peaks), max(peaks))
def terminal_drop_baseline(
cycles: list[CycleDict],
stop_threshold_w: float,
min_quiet_span_s: float,
min_clean_cycles: int,
) -> tuple[float | None, tuple[float, float] | None]:
"""Combined ``(earliest_quiet_offset, peak_range)`` baseline.
Pure (no store / no I/O beyond decompressing the passed-in cycle traces), so
the manager can offload the whole per-cycle scan to an executor thread in one
hop instead of decompressing every trace on the event loop (issue #311)."""
earliest = earliest_sustained_quiet_offset(
cycles, stop_threshold_w, min_quiet_span_s, min_clean_cycles
)
peak_range = device_active_peak_range(cycles, min_clean_cycles)
return earliest, peak_range
def is_terminal_drop(
points: list[tuple[float, float]],
earliest_quiet: float | None,
peak_range: tuple[float, float] | None,
stop_threshold_w: float,
earliness_ratio: float,
min_peak_ratio: float,
peak_familiar_tol: float,
) -> bool:
"""Whether a running cycle's current low-power event is an anomalously-early
terminal drop (plug pulled / cancelled) rather than a soak pause.
Pure decision logic (no store / no I/O) so it is unit-testable in isolation;
the manager wires the learned ``earliest_quiet`` / ``peak_range`` baselines
and gates on the ML opt-in. All three checks must hold:
* **Clearly ON** - the cycle peaked >= ``min_peak_ratio`` x stop-threshold
(a low-power idling device can't trip it).
* **Familiar** - the peak is within the device's historical peak range
widened by ``peak_familiar_tol``. A peak outside that band means the cycle
is drawing power unlike anything seen before (possibly a new program), so
we DEFER (return ``False``) rather than assume a stop.
* **Anomalous** - the trailing sub-threshold span began at an offset
``< earliest_quiet * earliness_ratio`` - earlier than this device has ever
legitimately gone quiet.
Returns ``False`` (keep the proven slow path) whenever a baseline is missing
or any check fails."""
if not points or earliest_quiet is None or peak_range is None:
return False
peak = max(p for _, p in points)
if peak < min_peak_ratio * max(stop_threshold_w, 1.0):
return False
lo, hi = peak_range
if not (lo * (1.0 - peak_familiar_tol) <= peak <= hi * (1.0 + peak_familiar_tol)):
return False
drop_start: float | None = None
for offset, power in points:
if power < stop_threshold_w:
if drop_start is None:
drop_start = offset
else:
drop_start = None
if drop_start is None:
return False
return drop_start < earliest_quiet * earliness_ratio
def compress_power_data(cycle: CycleDict) -> list[Any] | None:
"""Compress cycle power data to [offset, power] format (Module-level helper).
Returns the compressed list structure or None if compression failed/not needed.
"""
raw_data_raw = cycle.get("power_data")
if not isinstance(raw_data_raw, list) or not raw_data_raw:
return None
raw_data = cast(list[Any], raw_data_raw)
# Check if already compressed (first element is number or mixed format)
first = raw_data[0]
if isinstance(first, (int, float)):
# Already flat list (very old format?) or specific compression
return None
if isinstance(first, (list, tuple)):
first_seq = cast(list[Any] | tuple[Any, ...], first)
if len(first_seq) == 2 and isinstance(first_seq[0], (int, float)):
# Already compressed [offset, power]
return None
# Proceed with compression from [iso_string, power]
if "start_time" not in cycle:
return None
try:
start_ts = _value_to_timestamp(cycle.get("start_time"))
if start_ts is None:
return None
compressed: list[list[float]] = []
last_saved_p = -999.0
last_saved_t = -999.0
for i, entry in enumerate(raw_data):
if isinstance(entry, (list, tuple)):
entry_seq = cast(list[Any] | tuple[Any, ...], entry)
if len(entry_seq) != 2:
continue
t_str, p_val_raw = entry_seq
try:
# Handle both ISO string and potential timestamp float
t = _value_to_timestamp(t_str)
if t is None:
continue
p_val = float(p_val_raw)
offset = round(t - start_ts, 1)
if offset < 0:
offset = 0.0
# Save first and last
is_endpoint = i == 0 or i == len(raw_data) - 1
# Downsample: change > 1W or gap > 60s
if (
is_endpoint
or abs(p_val - last_saved_p) > 1.0
or (offset - last_saved_t) > 60
):
compressed.append([offset, round(p_val, 1)])
last_saved_p = p_val
last_saved_t = offset
except (ValueError, TypeError):
continue
return compressed
except Exception:
return None
class WashDataStore(Store[JSONDict]):
"""Store implementation with migration support."""
async def _async_migrate_func(
self,
old_major_version: int,
old_minor_version: int, # pylint: disable=unused-argument
old_data: JSONDict,
) -> JSONDict:
"""Migrate data to the new version."""
if old_major_version < 2:
_LOGGER.info("Migrating storage from v%s to v2", old_major_version)
# Logic moved from ProfileStore._migrate_v1_to_v2
cycles_raw = old_data.get("past_cycles", [])
cycles = cast(list[dict[str, Any]], cycles_raw) if isinstance(cycles_raw, list) else []
migrated_cycles = 0
for cycle in cycles:
if "signature" not in cycle and cycle.get("power_data"):
try:
# decompress_power_data now returns [(offset_seconds, power), ...]
tuples = decompress_power_data(cycle)
if tuples and len(tuples) > 10:
ts_arr = np.array([t for t, _ in tuples])
p_arr = np.array([p for _, p in tuples])
sig = compute_signature(ts_arr, p_arr)
cycle["signature"] = dataclasses.asdict(sig)
migrated_cycles += 1
except Exception as e: # pylint: disable=broad-exception-caught
_LOGGER.warning(
"Failed to migrate signature for cycle %s: %s", cycle.get("id"), e
)
_LOGGER.info(
"Migration v1->v2: Computed signatures for %s cycles", migrated_cycles
)
if old_major_version < 3:
_LOGGER.info("Migrating storage from v%s to v3", old_major_version)
cycles_raw = old_data.get("past_cycles", [])
profiles_raw = old_data.get("profiles", {})
cycles = cast(list[dict[str, Any]], cycles_raw) if isinstance(cycles_raw, list) else []
profiles = cast(dict[str, dict[str, Any]], profiles_raw) if isinstance(profiles_raw, dict) else {}
migrated_count = 0
# 1. Migrate Power Data to canonical offset format & ensure status
for cycle in cycles:
if "status" not in cycle:
cycle["status"] = "completed"
if cycle.get("power_data") and isinstance(cycle["power_data"], list):
try:
if migrate_power_data_to_offsets(cycle):
migrated_count += 1
except Exception as e:
_LOGGER.warning(
"Failed to migrate power data for cycle %s: %s",
cycle.get("id"),
e,
)
# 2. Ensure Device Type in Profiles
for profile in profiles.values():
if "device_type" not in profile:
profile["device_type"] = "washing_machine"
_LOGGER.info(
"Migration v2->v3: Migrated power data for %s cycles", migrated_count
)
if old_major_version < 4:
_LOGGER.info("Migrating storage from v%s to v4", old_major_version)
profiles = old_data.get("profiles", {})
if isinstance(profiles, dict):
for profile in cast(dict[str, dict[str, Any]], profiles).values():
phases = profile.get("phases")
if not isinstance(phases, list):
profile["phases"] = []
custom = old_data.get("custom_phases")
if not isinstance(custom, dict):
old_data["custom_phases"] = {}
if old_major_version < 5:
_LOGGER.info("Migrating storage from v%s to v5", old_major_version)
custom = old_data.get("custom_phases")
if isinstance(custom, list):
normalized: list[dict[str, Any]] = []
seen: set[tuple[str, str]] = set()
for item in cast(list[Any], custom):
if not isinstance(item, dict):
continue
item_dict = cast(dict[str, Any], item)
name = str(item_dict.get("name", "")).strip()
if not name:
continue
device_type = str(item_dict.get("device_type", "")).strip()
key = (name.casefold(), device_type.casefold())
if key in seen:
continue
seen.add(key)
normalized.append(
{
"name": name,
"description": str(item_dict.get("description", "")).strip(),
"device_type": device_type,
"created_at": item_dict.get("created_at") or dt_util.now().isoformat(),
}
)
old_data["custom_phases"] = normalized
elif isinstance(custom, dict):
normalized: list[dict[str, Any]] = []
seen: set[tuple[str, str]] = set()
custom_dict = cast(dict[str, Any], custom)
for legacy_device_type, phase_list in custom_dict.items():
if not isinstance(phase_list, list):
continue
for item in cast(list[Any], phase_list):
if not isinstance(item, dict):
continue
item_dict = cast(dict[str, Any], item)
name = str(item_dict.get("name", "")).strip()
if not name:
continue
device_type = str(legacy_device_type or "").strip()
key = (name.casefold(), device_type.casefold())
if key in seen:
continue
seen.add(key)
normalized.append(
{
"name": name,
"description": str(item_dict.get("description", "")).strip(),
"device_type": device_type,
"created_at": item_dict.get("created_at") or dt_util.now().isoformat(),
}
)
old_data["custom_phases"] = normalized
else:
old_data["custom_phases"] = []
if old_major_version < 6:
_LOGGER.info("Migrating storage from v%s to v6", old_major_version)
cycles_raw = old_data.get("past_cycles", [])
cycles = cast(list[dict[str, Any]], cycles_raw) if isinstance(cycles_raw, list) else []
flagged = _flag_recorded_cycles_golden(cycles)
_LOGGER.info(
"Migration v5->v6: flagged %s recorded cycle(s) as golden references", flagged
)
if old_major_version < 7:
# The v6 step only ran when upgrading from below v6, so recorded
# cycles on installs already at v6 (e.g. recorded by an older build
# that didn't set the golden flag) were never tagged. Re-run the now
# broadened backfill once so every historical recorded cycle is
# recognised. Idempotent: already-golden cycles are skipped.
_LOGGER.info("Migrating storage from v%s to v7", old_major_version)
cycles_raw = old_data.get("past_cycles", [])
cycles = cast(list[dict[str, Any]], cycles_raw) if isinstance(cycles_raw, list) else []
flagged = _flag_recorded_cycles_golden(cycles)
_LOGGER.info(
"Migration v6->v7: tagged %s previously-unmarked recorded cycle(s) as golden",
flagged,
)
if old_major_version < 8:
# The v6->v7 backfill identified recordings only by the explicit
# ``meta.source``/``original_samples`` marker, which OLD recordings
# (made before that marker existed, ``meta: None``) do not carry — so
# they stayed untagged. _is_recorded_cycle now also recognises them by
# their structural signature (completed + no max_power/termination_reason).
# Re-run the backfill once more so those old recordings are finally
# tagged. Idempotent: already-golden cycles are skipped.
_LOGGER.info("Migrating storage from v%s to v8", old_major_version)
cycles_raw = old_data.get("past_cycles", [])
cycles = cast(list[dict[str, Any]], cycles_raw) if isinstance(cycles_raw, list) else []
flagged = _flag_recorded_cycles_golden(cycles)
_LOGGER.info(
"Migration v7->v8: tagged %s old recorded cycle(s) (no meta marker) as golden",
flagged,
)
if old_major_version < 9:
# Pre-initialize additive top-level keys so they are present from the
# first load rather than appearing lazily on first use. Idempotent:
# setdefault leaves existing values untouched.
_LOGGER.info("Migrating storage from v%s to v9", old_major_version)
old_data.setdefault("lifetime_energy_wh", 0.0)
# Seed the monotonic lifetime cycle counter from the retained history so an
# existing install does not start at 0 and re-fire past milestones (50/100/...)
# on cycles it already completed. Matches the cycle_count fallback semantics.
old_data.setdefault(
"lifetime_cycle_count", len(old_data.get("past_cycles") or [])
)
old_data.setdefault("settings_changelog", [])
old_data.setdefault("maintenance_log", [])
if old_major_version < 10:
# Reference cycles imported from the community store live in their own list,
# never in past_cycles, so they can feed the envelope/matcher but can never
# touch usage/energy stats. Additive + idempotent.
_LOGGER.info("Migrating storage from v%s to v10", old_major_version)
old_data.setdefault("reference_cycles", [])
if old_major_version < 11:
# Marker-only bump. Per-phase profiles (envelope["phase_profile"], used
# by phase-segmented matching / phase-resolved ETA) are DERIVED CACHE
# built by async_rebuild_envelope, not stored data - so there is nothing
# to migrate. They self-populate on the next envelope rebuild (which
# runs on every cycle end / label change); until then consumers fall
# back to the existing estimator via lazy absent-key handling. No data
# is added, removed, or altered here.
_LOGGER.info("Migrating storage from v%s to v11 (phase-profile cache marker)",
old_major_version)
return old_data
def _ambiguity_from_candidates(candidates: list[dict]) -> tuple[float, bool]:
"""Top1-vs-top2 score margin and whether the match is ambiguous.
Single source for the ambiguity rule shared by both match paths:
``is_ambiguous = margin < MATCH_AMBIGUITY_MARGIN``. ``margin`` defaults to
1.0 when there is only one candidate.
"""
margin = 1.0
if len(candidates) > 1:
margin = candidates[0]["score"] - candidates[1]["score"]
return margin, margin < MATCH_AMBIGUITY_MARGIN
class ProfileStore:
"""Manages storage of washer profiles and past cycles."""
def __init__(
self,
hass: HomeAssistant,
entry_id: str,
min_duration_ratio: float = 0.50,
max_duration_ratio: float = 1.50,
save_debug_traces: bool = False,
match_threshold: float = 0.4,
unmatch_threshold: float = 0.35,
device_name: str = "",
) -> None:
"""Initialize the profile store."""
self.hass = hass
self.entry_id = entry_id
self._logger = DeviceLoggerAdapter(_LOGGER, device_name)
self._min_duration_ratio = min_duration_ratio
self._max_duration_ratio = max_duration_ratio
self._match_threshold = match_threshold
self._unmatch_threshold = unmatch_threshold
self.dtw_bandwidth: float = DEFAULT_DTW_BANDWIDTH
self._save_debug_traces = save_debug_traces
# Cache for resampled sample segments: key=(cycle_id, dt)
self._cached_sample_segments: dict[tuple[str, float], Segment] = {}
# Cache for group cohesion scores to avoid re-running DTW on the event loop
# every 5 minutes. Keyed by sorted-members tuple; invalidated when profile_groups
# content changes (tracked by a simple generation counter).
self._cohesion_cache: dict[tuple[str, ...], float] = {}
self._cohesion_cache_generation: int = 0
self._cohesion_cache_generation_checked: int = -1
# group_cohesion runs in executor threads (live matching + Playground can
# touch the same store concurrently); serialize the read-clear-compute-store
# sequence so a concurrent call can't clear a half-populated cache or
# duplicate the DTW work. A plain thread lock, never held across an await.
self._cohesion_cache_lock = threading.Lock()
# Profile duration tolerance (set by manager; reserved for duration-based heuristics)
self._duration_tolerance: float = 0.25
# Retention policy: cap total cycles and number of full-resolution traces per profile
self._max_past_cycles = DEFAULT_MAX_PAST_CYCLES
self._max_full_traces_per_profile = DEFAULT_MAX_FULL_TRACES_PER_PROFILE
self._max_full_traces_unlabeled = DEFAULT_MAX_FULL_TRACES_UNLABELED
# Separate store for each entry to avoid giant files
# Use WashDataStore to handle migration
self._store: Store[JSONDict] = WashDataStore(
hass, STORAGE_VERSION, f"{STORAGE_KEY}.{entry_id}"
)
self._data: JSONDict = {
"profiles": {},
"past_cycles": [],
"reference_cycles": [], # Imported store cycles: envelope/matcher only, never usage stats
"envelopes": {}, # Cached statistical envelopes per profile
"auto_adjustments": [], # Log of automatic setting changes
"suggestions": {}, # Suggested settings (do NOT change user options)
"feedback_history": {}, # Persisted user feedback (cycle_id -> record)
"pending_feedback": {}, # Persisted pending feedback requests
"custom_phases": [], # Shared custom phase catalog
"ml_model_versions": {}, # On-device trained model specs (Stage 4)
"profile_groups": {}, # Named groups of near-duplicate profiles (Stage 5)
"maintenance_log": [], # User-logged maintenance events (Group E)
}
def set_suggestion(
self,
key: str,
value: Any,
reason: str | None = None,
reason_key: str | None = None,
reason_params: dict[str, Any] | None = None,
) -> None:
"""Store a suggested setting value without changing config entry options.
``reason`` is the English fallback text; ``reason_key`` + ``reason_params``
are the localization key + interpolation values the panel resolves via
``_t(reason_key, reason_params, reason)``. Both localization fields are
optional so old callers / persisted entries keep working.
"""
suggestions: JSONDict = self._data.setdefault("suggestions", {})
entry: dict[str, Any] = {
"value": value,
"reason": reason,
"updated": dt_util.now().isoformat(),
}
# Store localization sidecars only when present so we never overwrite an
# existing key with a stale ``None`` (keeps the shape lean for old data).
if reason_key is not None:
entry["reason_key"] = reason_key
if reason_params is not None:
entry["reason_params"] = reason_params
suggestions[key] = entry
def get_suggestions(self) -> dict[str, Any]:
"""Return current suggestion map."""
raw = self._data.get("suggestions")
if isinstance(raw, dict):
suggestions = cast(JSONDict, raw)
return suggestions.copy()
return {}
def delete_suggestion(self, key: str) -> None:
"""Remove a single suggestion entry by key."""
suggestions: JSONDict = self._data.setdefault("suggestions", {})
suggestions.pop(key, None)
async def clear_suggestions(self) -> None:
"""Clear all pending suggestions and persist."""
self._data["suggestions"] = {}
await self.async_save()
def get_suggestion_apply_cycle_count(self) -> int:
"""Return the total cycle count recorded at the last user suggestion apply."""
return int(self._data.get("suggestion_apply_cycle_count", 0))
def set_suggestion_apply_cycle_count(self, count: int) -> None:
"""Record the total cycle count at the moment the user applies suggestions."""
self._data["suggestion_apply_cycle_count"] = count
# ─── Settings changelog (Group D7) ─────────────────────────────────────────
#: Maximum settings-changelog entries retained per device (newest kept).
SETTINGS_CHANGELOG_MAX = 50
def get_settings_changelog(self) -> list[dict[str, Any]]:
"""Return the recorded settings-change history (most-recent-first).
Never raises: returns an empty list when no changelog exists yet.
"""
try:
raw = self._data.get("settings_changelog", [])
if isinstance(raw, list):
return raw
except Exception: # pylint: disable=broad-exception-caught
pass
return []
async def async_record_settings_changes(self, changes: list[dict]) -> None:
"""Append settings-change entries and persist (capped at the newest 50).
Each entry is normalized to ``{"key", "old", "new", "timestamp"}`` and
prepended so the list stays most-recent-first. No-op on empty/invalid
input.
"""
if not changes:
return
log = self._data.setdefault("settings_changelog", [])
if not isinstance(log, list):
log = []
self._data["settings_changelog"] = log
now_iso = dt_util.now().isoformat()
added = False
for ch in changes:
if not isinstance(ch, dict) or "key" not in ch:
continue
log.insert(
0,
{
"key": str(ch.get("key")),
"old": ch.get("old"),
"new": ch.get("new"),
"timestamp": ch.get("timestamp") or now_iso,
},
)
added = True
if not added:
return
# Keep only the newest entries (list is most-recent-first).
if len(log) > self.SETTINGS_CHANGELOG_MAX:
del log[self.SETTINGS_CHANGELOG_MAX:]
self._data["settings_changelog"] = log
await self.async_save()
# ─── On-device ML model versions (Stage 4) ────────────────────────────────
def get_ml_model_versions(self) -> dict[str, Any]:
"""Return the map of on-device trained model specs (capability -> record).
Each record is ``{spec, trained_at, cycle_count, metrics, baseline_auc}``
where ``spec`` is a NumPy-only standardized-logistic model produced by
:mod:`.ml.trainer`. Empty when nothing has been trained on-device.
"""
raw = self._data.get("ml_model_versions")
if isinstance(raw, dict):
return cast(JSONDict, raw).copy()
return {}
async def set_ml_model_version(self, capability: str, record: dict[str, Any]) -> None:
"""Persist a trained model record for a capability (quality/live_match/end)."""
versions: JSONDict = self._data.setdefault("ml_model_versions", {})
versions[capability] = record
await self.async_save()
async def clear_ml_model_versions(self) -> None:
"""Drop all on-device trained models (reverts to embedded baselines)."""
self._data["ml_model_versions"] = {}
await self.async_save()
def get_ml_last_training_run(self) -> str | None:
"""ISO timestamp of the last *completed* on-device training run, or None.
Distinct from each model's ``trained_at`` (the promotion time): this
advances on every run even when no model beat the baseline, so the panel's
"Last trained" reflects that training actually ran.
"""
ts = self._data.get("ml_last_training_run")
return ts if isinstance(ts, str) else None
async def set_ml_last_training_run(self, iso: str) -> None:
"""Record when on-device training last completed (regardless of promotion)."""
self._data["ml_last_training_run"] = iso
await self.async_save()
def get_ml_training_history(self) -> dict[str, list[dict[str, Any]]]:
"""Per-capability held-out-score history across training runs.
``{capability: [{"ts": iso, "score": float, "higher_better": bool}, ...]}``
oldest-first. Lets the panel show whether a model's fit is improving,
steady, or declining over time. Empty until training has run.
"""
raw = self._data.get("ml_training_history")
return cast(dict[str, list[dict[str, Any]]], raw) if isinstance(raw, dict) else {}
async def append_ml_training_history(self, run_iso: str, results: list[dict[str, Any]]) -> None:
"""Append each capability's held-out score from a training run.
``results`` is ``train_from_cycles``'s per-capability records; classifiers
report ``new_auc`` (higher is better), regressors ``model_mae`` (lower is
better). Capabilities without a held-out metric this run (insufficient
data) are skipped. Retains at most ``ML_TRAINING_HISTORY_MAX`` runs each.
"""
from .const import ML_TRAINING_HISTORY_MAX # noqa: PLC0415
hist: JSONDict = self._data.setdefault("ml_training_history", {})
changed = False
for rec in results or []:
if not isinstance(rec, dict):
continue
cap = rec.get("capability")
if not isinstance(cap, str) or not cap:
continue
if rec.get("new_auc") is not None:
score, higher_better = float(rec["new_auc"]), True
elif rec.get("model_mae") is not None:
score, higher_better = float(rec["model_mae"]), False
else:
continue
series = hist.setdefault(cap, [])
if not isinstance(series, list):
series = hist[cap] = []
series.append({"ts": run_iso, "score": round(score, 5), "higher_better": higher_better})
del series[:-ML_TRAINING_HISTORY_MAX]
changed = True
if changed:
await self.async_save()
# ─── On-device matching-config tuning (Stage 4/5, opt-in) ──────────────────
#: Only these bounded scoring weights (all in [0, 1]) may be overridden
#: on-device, so a stored record can never change structural matching
#: behaviour. Mirrors ml.matching_tuner.OVERRIDE_KEYS.
_MATCHING_OVERRIDE_KEYS = ("corr_weight", "duration_weight", "energy_weight", "dtw_ensemble_w")
def get_matching_config(self) -> dict[str, Any]:
"""Return the on-device tuned matcher scoring-weight record, if any.
Record shape: ``{config, trained_at, cycle_count, baseline_test_top1,
tuned_test_top1}`` where ``config`` holds the bounded scoring weights
(``corr_weight``/``duration_weight``/``energy_weight``). Empty when the
matcher is using the shipped defaults.
"""
raw = self._data.get("matching_config")
if isinstance(raw, dict):
return cast(JSONDict, raw).copy()
return {}
async def set_matching_config(self, record: dict[str, Any]) -> None:
"""Persist the tuned matcher scoring-weight override + metadata."""
self._data["matching_config"] = record
await self.async_save()
async def clear_matching_config(self) -> None:
"""Revert the matcher to the shipped default scoring weights."""
self._data.pop("matching_config", None)
await self.async_save()
def _matching_overrides(self) -> dict[str, float]:
"""Bounded scoring-weight overrides to merge into the matcher config.
Only the whitelisted keys are honoured, each clamped to ``[0, 1]``, so a
stored record can never alter structural matching behaviour - only the
emphasis between shape, level and energy agreement.
"""
rec = self._data.get("matching_config")
cfg = rec.get("config") if isinstance(rec, dict) else None
out: dict[str, float] = {}
if isinstance(cfg, dict):
for k in self._MATCHING_OVERRIDE_KEYS:
try:
out[k] = min(1.0, max(0.0, float(cfg[k])))
except (KeyError, TypeError, ValueError):
continue
return out
async def set_cycle_review(
self,
cycle_id: str,
*,
quality: str | None = None,
golden: bool | None = None,
tags: list[str] | None = None,
notes: str | None = None,
) -> bool:
"""Attach an ML-Lab review to a cycle (Stage 4b).
The review (``quality``/``golden``/``tags``/``notes``) is stored under the
cycle's ``ml_review`` key and becomes a strong training label for the
on-device quality model. Only the provided fields are updated. Returns
True when the cycle was found and updated.
"""
cycle = next(
(c for c in self._data.get("past_cycles", []) if c.get("id") == cycle_id),
None,
)
if not cycle:
raise ValueError(f"Cycle {cycle_id} not found")
prev_review = cycle.get("ml_review")
prev_golden = bool(prev_review.get("golden")) if isinstance(prev_review, dict) else False
review = dict(prev_review or {})
if quality is not None:
review["quality"] = quality
if golden is not None:
review["golden"] = bool(golden)
if tags is not None:
review["tags"] = list(tags)
if notes is not None:
review["notes"] = notes
review["reviewed_at"] = dt_util.now().isoformat()
cycle["ml_review"] = review
# Golden cycles seed the profile's matching reference/envelope. When the
# golden flag actually flips (either direction), rebuild the affected
# profile so its envelope + reference cycle + cohesion cache reflect the
# new golden set. Only when the value changed — an unchanged save stays
# idempotent and skips the (executor-bound) rebuild.
golden_changed = golden is not None and bool(golden) != prev_golden
profile_name = cycle.get("profile_name")
if golden_changed and isinstance(profile_name, str) and profile_name:
await self.async_rebuild_envelope(profile_name)
await self.async_save()
# Don't log the full review (notes/tags are user-entered) — only field names.
_changed = [
n for n, v in (("quality", quality), ("golden", golden), ("tags", tags), ("notes", notes))
if v is not None
]
self._logger.info(
"Recorded ML review for cycle %s (updated: %s)", cycle_id, ", ".join(_changed) or "none"
)
return True
async def async_backfill_recorded_golden(self) -> int:
"""Flag manually-recorded cycles (meta.source == 'recorder') as golden
references. Recorded == golden: a single stored flag, no duplicate field.
Idempotent; saves only when something changed. Returns the count flagged.
"""
cycles = self._data.get("past_cycles", [])
if not isinstance(cycles, list):
return 0
# Snapshot each cycle's golden state (aligned by list position, which
# _flag_recorded_cycles_golden never reorders) so we can rebuild exactly
# the profiles whose cycles get newly flagged — golden cycles seed the
# matching reference/envelope.
before_golden = [
bool(c.get("ml_review", {}).get("golden"))
if isinstance(c, dict) and isinstance(c.get("ml_review"), dict)
else False
for c in cycles
]
flagged = _flag_recorded_cycles_golden(cycles)
if flagged:
affected: set[str] = set()
for was_golden, cycle in zip(before_golden, cycles):
if was_golden or not isinstance(cycle, dict):
continue
review = cycle.get("ml_review")
if isinstance(review, dict) and review.get("golden"):
pname = cycle.get("profile_name")
if isinstance(pname, str) and pname:
affected.add(pname)
for pname in affected:
await self.async_rebuild_envelope(pname)
await self.async_save()
self._logger.info("Backfilled golden flag on %s recorded cycle(s)", flagged)
return flagged
def get_feedback_history(self) -> dict[str, dict[str, Any]]:
"""Return mutable feedback history mapping (cycle_id -> record)."""
raw = self._data.setdefault("feedback_history", {})
if isinstance(raw, dict):
return cast(dict[str, dict[str, Any]], raw)
return {}
def get_pending_feedback(self) -> dict[str, dict[str, Any]]:
"""Return mutable pending feedback mapping (cycle_id -> request)."""
raw = self._data.setdefault("pending_feedback", {})
if isinstance(raw, dict):
return cast(dict[str, dict[str, Any]], raw)
return {}
def add_pending_feedback(self, cycle_id: str, request_data: dict[str, Any]) -> None:
"""Add a pending feedback request (sync wrapper, does not save immediately)."""
feedbacks = self.get_pending_feedback()
feedbacks[cycle_id] = request_data
# Caller must ensure save is called eventually
def prune_orphaned_feedback(self) -> int:
"""Drop pending-feedback entries whose cycle no longer exists.
A cycle that was deleted/merged/trimmed can leave its pending feedback
behind, so the device's "needs review" count stays >0 while the Cycles
review filter (which matches feedback ids against real cycles) shows
nothing. Returns the number pruned; never raises. Caller saves.
"""
try:
pending = self._data.get("pending_feedback")
if not isinstance(pending, dict) or not pending:
return 0
live_ids = {c.get("id") for c in self.get_past_cycles()}
orphans = [cid for cid in list(pending) if cid not in live_ids]
for cid in orphans:
pending.pop(cid, None)
return len(orphans)
except Exception: # noqa: BLE001
return 0
def get_profile(self, name: str) -> JSONDict | None:
"""Return a single profile by name with calculated stats (via list_profiles)."""
# Reuse list_profiles logic to ensure consistency and avoid duplication
all_profiles = self.list_profiles()
return next((p for p in all_profiles if p["name"] == name), None)
def get_profiles(self) -> dict[str, JSONDict]:
"""Return mutable profiles mapping (profile_name -> profile data)."""
raw = self._data.setdefault("profiles", {})
if isinstance(raw, dict):
return cast(dict[str, JSONDict], raw)
return {}
def get_past_cycles(self) -> list[CycleDict]:
"""Return mutable list of stored cycles."""
raw = self._data.setdefault("past_cycles", [])
if isinstance(raw, list):
return cast(list[CycleDict], raw)
return []
# ── Community-store account (connect handoff) ─────────────────────────────
def get_store_account(self) -> dict[str, Any]:
"""Full persisted store account incl. the refresh token (credential)."""
raw = self._data.get("store_account")
return dict(raw) if isinstance(raw, dict) else {}
def get_store_identity(self) -> dict[str, Any]:
"""Safe account view for status/UI - never includes the refresh token."""
a = self.get_store_account()
return {
"connected": bool(a.get("refresh_token")),
"uid": a.get("uid"),
"name": a.get("name"),
"brand": a.get("brand"),
"model": a.get("model"),
}
async def set_store_account(self, account: dict[str, Any]) -> None:
"""Persist/merge the store account (refresh_token, uid, name, brand, model)."""
cur = self.get_store_account()
cur.update({k: v for k, v in account.items() if v is not None})
self._data["store_account"] = cur
await self.async_save()
async def clear_store_account(self) -> None:
self._data.pop("store_account", None)
await self.async_save()
def get_reference_cycles(self) -> list[CycleDict]:
"""Return the imported-store reference cycles.
These are NOT in ``past_cycles``: they feed only the envelope shape and the
matcher template, never usage/energy/count/trend stats.
"""
raw = self._data.setdefault("reference_cycles", [])
if isinstance(raw, list):
return cast(list[CycleDict], raw)
return []
def get_shareable_cycles(self) -> list[dict[str, Any]]:
"""Recorded/golden reference cycles eligible to share to the community store.
A cycle qualifies when it is hand-flagged golden (``ml_review.golden``) or a
manual recording (``meta.source == "recorder"``) and carries a program label.
Imported ``reference_cycles`` are intentionally excluded (you do not re-share
what you downloaded). Returns light summaries (no traces), most-recent-first;
the whole set (not a page) so the panel's share tree can offer all of them.
Never raises.
"""
out: list[dict[str, Any]] = []
for c in self.get_past_cycles():
if not isinstance(c, dict):
continue
program = str(c.get("profile_name") or "").strip()
if not program:
continue
review = c.get("ml_review") if isinstance(c.get("ml_review"), dict) else {}
meta = c.get("meta") if isinstance(c.get("meta"), dict) else {}
if not (review.get("golden") or meta.get("source") == "recorder"):
continue
out.append({
"id": c.get("id"),
"profile_name": program,
"start_time": c.get("start_time"),
"duration": c.get("duration"),
"source": meta.get("source"),
})
out.sort(key=lambda r: r.get("start_time") or "", reverse=True)
return out
async def add_reference_cycle(
self, profile_name: str, points: list[list[float]], meta: dict[str, Any]
) -> str:
"""Import a reference cycle downloaded from the store into ``reference_cycles``.
``points`` is a raw trace of ``[offset_seconds, watts]`` pairs. ``meta`` may carry
``store_cycle_id`` (-> ``meta.source = "store:"``), ``store_uploaded_at`` and
``sampling_interval``. The cycle is stamped with import-time timestamps (its real
run time is meaningless locally), forced ``status="completed"`` and
``ml_review.golden=True`` so it seeds the envelope shape, then the envelope is
rebuilt. Never accumulates lifetime energy or touches ``past_cycles``.
"""
# Validate the trace BEFORE creating any persistent state (profile entry /
# reference cycle): drop non-finite/non-numeric samples, require >= 2 points
# and a positive time span. A garbage trace returns "" and mutates nothing.
pairs: list[list[float]] = []
for p in (points or []):
if len(p) < 2:
continue
try:
x, y = float(p[0]), float(p[1])
except (TypeError, ValueError):
continue
if math.isfinite(x) and math.isfinite(y):
pairs.append([x, y])
if len(pairs) < 2:
return ""
pairs.sort(key=lambda q: q[0]) # defensive: normalize any out-of-order offsets
duration = float(pairs[-1][0] - pairs[0][0])
if duration <= 0:
return ""
# Re-base to offset 0 so envelope reconstruction and DTW work correctly.
if pairs[0][0] != 0.0:
origin = pairs[0][0]
pairs = [[p[0] - origin, p[1]] for p in pairs]
now = dt_util.now()
store_id = str(meta.get("store_cycle_id") or "")
cycle: CycleDict = {
"profile_name": profile_name,
"power_data": pairs,
"start_time": now.isoformat(),
# Keep the timestamp interval consistent with the imported duration
# (both were previously "now", giving a zero-length interval).
"end_time": (now + timedelta(seconds=duration)).isoformat(),
"duration": duration,
"status": "completed",
"ml_review": {"golden": True},
"meta": {
"source": f"store:{store_id}" if store_id else "store",
"store_uploaded_at": meta.get("store_uploaded_at"),
},
}
if meta.get("sampling_interval"):
cycle["sampling_interval"] = float(meta["sampling_interval"])
# A reference cycle implies its program exists locally; create a minimal profile
# entry if absent so the matcher iterates it and the rebuild can set its template.
profiles = self._data.setdefault("profiles", {})
if profile_name not in profiles:
profiles[profile_name] = {"avg_duration": duration}
self._add_cycle_data(cycle, target=self._data.setdefault("reference_cycles", []))
await self.async_rebuild_envelope(profile_name)
await self.async_save()
return str(cycle.get("id", ""))
# ── Profile groups (Stage 5: near-duplicate variants) ──────────────────────
def get_profile_groups(self) -> dict[str, JSONDict]:
"""Return mutable profile-groups mapping (group_name -> {members, created_at}).
A group bundles near-duplicate profiles (e.g. the same program at
different temperature/spin). The matcher treats the group as one
aggregate candidate and only picks the exact member once the group wins.
Groups with fewer than 2 present members are ignored by the matcher.
"""
raw = self._data.setdefault("profile_groups", {})
if not isinstance(raw, dict):
self._data["profile_groups"] = {}
return self._data["profile_groups"]
# Drop members that no longer exist as profiles (kept consistent lazily).
profiles = self.get_profiles()
for g in raw.values():
if isinstance(g, dict) and isinstance(g.get("members"), list):
g["members"] = [m for m in g["members"] if m in profiles]
return cast(dict[str, JSONDict], raw)
def _members_in_other_groups(self, members: list[str], exclude: str | None) -> dict[str, str]:
"""Map any member already assigned to a DIFFERENT group -> that group name."""
owner: dict[str, str] = {}
for gname, g in self.get_profile_groups().items():
if gname == exclude or not isinstance(g, dict):
continue
for m in (g.get("members") or []):
owner.setdefault(m, gname)
return {m: owner[m] for m in members if m in owner}
async def create_profile_group(self, name: str, members: list[str]) -> bool:
"""Create (or overwrite) a group with the given member profile names."""
name = (name or "").strip()
if not name:
raise ValueError("Group name is required")
profiles = self.get_profiles()
members = [m for m in dict.fromkeys(members or []) if m in profiles]
# A profile may belong to at most one group (else _grouped_snapshots would
# collapse it inconsistently). Reject up front — no partial mutation.
conflicts = self._members_in_other_groups(members, exclude=name)
if conflicts:
raise ValueError(
"Already in another group: "
+ ", ".join(f"{m} ({g})" for m, g in conflicts.items())
)
groups = self.get_profile_groups()
groups[name] = {"members": members, "created_at": dt_util.now().isoformat()}
self._cohesion_cache_generation += 1
await self.async_save()
self._logger.info("Created profile group %r with %d members", name, len(members))
return True
async def set_profile_group_members(self, name: str, members: list[str]) -> bool:
"""Replace a group's member list. Deletes the group if left empty."""
groups = self.get_profile_groups()
if name not in groups:
raise ValueError(f"Group {name} not found")
profiles = self.get_profiles()
members = [m for m in dict.fromkeys(members or []) if m in profiles]
conflicts = self._members_in_other_groups(members, exclude=name)
if conflicts:
raise ValueError(
"Already in another group: "
+ ", ".join(f"{m} ({g})" for m, g in conflicts.items())
)
if members:
groups[name]["members"] = members
else:
groups.pop(name, None)
self._cohesion_cache_generation += 1
await self.async_save()
return True
async def rename_profile_group(self, name: str, new_name: str) -> bool:
groups = self.get_profile_groups()
new_name = (new_name or "").strip()
if name not in groups or not new_name:
raise ValueError("Group not found or new name empty")
if new_name != name:
if new_name in groups:
raise ValueError(f"A group named {new_name!r} already exists")
groups[new_name] = groups.pop(name)
self._cohesion_cache_generation += 1
await self.async_save()
return True
async def delete_profile_group(self, name: str) -> bool:
groups = self.get_profile_groups()
if groups.pop(name, None) is None:
return False
self._cohesion_cache_generation += 1
await self.async_save()
self._logger.info("Deleted profile group %r", name)
return True
def _profile_curve(self, name: str, n: int = 150) -> np.ndarray | None:
"""A profile's envelope average resampled to n points, or None."""
env = self._data.get("envelopes", {}).get(name) if isinstance(self._data.get("envelopes"), dict) else None
avg = env.get("avg") if isinstance(env, dict) else None
if not avg or not isinstance(avg[0], (list, tuple)):
return None
ys = np.asarray([float(p[1]) for p in avg], dtype=float)
if ys.size < 4:
return None
return np.interp(np.linspace(0, 1, n), np.linspace(0, 1, ys.size), ys)
@staticmethod
def _shape_similarity(a: np.ndarray, b: np.ndarray) -> float:
"""Duration- and amplitude-tolerant shape similarity in (0,1].
Peak-normalises both curves (so temperature/spin amplitude differences
don't count against membership) then aligns them with a Sakoe-Chiba-band
DTW (so a longer heating or draining phase at a different setting warps
into alignment instead of being penalised as a different shape). 1.0 =
identical shape; genuinely different programs score low even after warping.
"""
na = a / (float(a.max()) or 1.0)
nb = b / (float(b.max()) or 1.0)
dist = analysis.compute_dtw_lite(na, nb, band_width_ratio=0.2)
norm = dist / max(len(na), 1) # mean per-sample distance on [0,1] curves
return 1.0 / (1.0 + norm / 0.15)
def group_cohesion(self, members: list[str]) -> float:
"""Minimum pairwise shape similarity among a group's members (1.0 =
identical shapes; see _shape_similarity for the DTW/peak-normalised
metric). Low cohesion means the members are not really the same program,
so their averaged aggregate would be too generic and could out-match
unrelated profiles - the matcher refuses to aggregate below
GROUP_MIN_COHESION and the UI warns the user.
Results are cached per member-set and invalidated via
``_cohesion_cache_generation`` (incremented by group mutation methods) so
the DTW pairwise comparison does not run on the event loop every 5 minutes.
"""
key = tuple(sorted(members))
with self._cohesion_cache_lock:
if self._cohesion_cache_generation != self._cohesion_cache_generation_checked:
self._cohesion_cache.clear()
self._cohesion_cache_generation_checked = self._cohesion_cache_generation
if key in self._cohesion_cache:
return self._cohesion_cache[key]
curves = [c for c in (self._profile_curve(m) for m in members) if c is not None]
if len(members) < 2:
# A genuinely single-member group is trivially cohesive (and is never
# collapsed anyway — nothing to aggregate).
result = 1.0
elif len(curves) < 2:
# Multi-member but too few built curves -> insufficient evidence, treat as
# NOT cohesive so the group isn't collapsed into a blurry aggregate yet.
result = 0.0
else:
result = 1.0
for i in range(len(curves)):
for j in range(i + 1, len(curves)):
result = min(result, self._shape_similarity(curves[i], curves[j]))
self._cohesion_cache[key] = result
return result
def _grouped_snapshots(
self, snapshots: list[dict[str, Any]]
) -> tuple[list[dict[str, Any]], dict[str, list[str]], dict[str, dict[str, Any]]]:
"""Collapse each *cohesive* group into one aggregate candidate snapshot.
Returns (snapshots, group_members, member_snaps). Groups with <2 present
members, or cohesion below GROUP_MIN_COHESION (too generic an aggregate),
are left as individual member snapshots. Aggregate name is prefixed
``__group__`` and mapped to its member profile names in group_members.
"""
groups = self.get_profile_groups()
if not groups:
return snapshots, {}, {}
by_name = {s["name"]: s for s in snapshots}
member_to_group: dict[str, str] = {}
for gname, g in groups.items():
present = [m for m in (g.get("members") or []) if m in by_name]
if len(present) < 2 or self.group_cohesion(present) < GROUP_MIN_COHESION:
continue # keep members individual
for m in present:
member_to_group[m] = gname
if not member_to_group:
return snapshots, {}, {}
n = 200
agg_curves: dict[str, list[np.ndarray]] = {}
agg_durs: dict[str, list[float]] = {}
member_snaps: dict[str, dict[str, Any]] = {}
out: list[dict[str, Any]] = []
for s in snapshots:
g = member_to_group.get(s["name"])
if g is None:
out.append(s)
continue
member_snaps[s["name"]] = s
sp = s.get("sample_power") or []
if len(sp) >= 2:
arr = np.asarray(sp, dtype=float)
agg_curves.setdefault(g, []).append(
np.interp(np.linspace(0, 1, n), np.linspace(0, 1, arr.size), arr)
)
agg_durs.setdefault(g, []).append(float(s.get("avg_duration") or 0.0))
group_members: dict[str, list[str]] = {}
for g, curves in agg_curves.items():
key = f"__group__{g}"
durs = [d for d in agg_durs[g] if d > 0]
out.append({
"name": key,
"avg_duration": float(np.mean(durs)) if durs else 0.0,
"sample_power": np.mean(np.array(curves), axis=0).tolist(),
})
group_members[key] = [m for m in member_to_group if member_to_group[m] == g and m in member_snaps]
return out, group_members, member_snaps
def _stage5_pick_member(
self, current_power: list[float], current_duration: float,
members: list[str], member_snaps: dict[str, dict[str, Any]],
) -> tuple[str, float | None, float | None]:
"""Within a winning group, pick the member whose duration + mean power +
peak best match the cycle (temperature -> mean power, spin -> peak).
Returns (member_name, individual_fit_score, member_avg_duration). The fit
score is the chosen member's own alignment score, used as a sanity check."""
cur = np.asarray(current_power, dtype=float)
if cur.size == 0 or not members:
return (members[0] if members else ""), None, None
cur_mp = float(cur.mean()); cur_pk = float(cur.max())
def agree(a: float, b: float, scale: float) -> float:
if a <= 0 or b <= 0:
return 0.0
return 1.0 / (1.0 + abs(math.log(a / b)) / scale)
best_m, best_sc, best_dur = members[0], -1.0, None
for m in members:
snap = member_snaps.get(m)
if not snap:
continue
sp = np.asarray(snap.get("sample_power") or [], dtype=float)
if sp.size == 0:
continue
md = float(snap.get("avg_duration") or 0.0)
sc = (agree(current_duration, md, 0.15)
* agree(cur_mp, float(sp.mean()), 0.20)
* agree(cur_pk, float(sp.max()), 0.20))
if sc > best_sc:
best_sc, best_m, best_dur = sc, m, md
fit = None
snap = member_snaps.get(best_m)
if snap and snap.get("sample_power"):
try:
fit = float(analysis.find_best_alignment(current_power, snap["sample_power"], 1.0)[0])
except Exception: # pylint: disable=broad-exception-caught
fit = None
return best_m, fit, best_dur
def suggest_profile_groups(self, dur_tol: float = 0.60, sim_min: float = 0.85) -> list[dict[str, Any]]:
"""Detect clusters of near-duplicate profiles not already fully grouped.
Two profiles cluster when their durations are within a (loose) ``dur_tol``
sanity bound AND their DTW/peak-normalised shape similarity exceeds
``sim_min`` (same program shape; they may differ in temperature/spin and in
phase length, which the DTW alignment tolerates). The duration bound is
loose because higher-temp/lower-rpm variants legitimately run longer - it
only rules out grouping, say, a 20-min rinse with a 3-hour cotton.
Returns {"members": [...], "existing_group": name|None} suggestions the
user confirms. Never mutates state.
"""
profiles = self.get_profiles()
envelopes = self._data.get("envelopes", {})
# Build per-profile (avg curve resampled to N, avg duration).
N = 150
reps: dict[str, tuple[np.ndarray, float]] = {}
for name, prof in profiles.items():
env = envelopes.get(name) if isinstance(envelopes, dict) else None
avg = env.get("avg") if isinstance(env, dict) else None
dur = float(prof.get("avg_duration") or 0.0)
if not avg or dur <= 0 or not isinstance(avg[0], (list, tuple)):
continue
ys = np.asarray([float(p[1]) for p in avg], dtype=float)
if ys.size < 4:
continue
curve = np.interp(np.linspace(0, 1, N), np.linspace(0, 1, ys.size), ys)
reps[name] = (curve, dur)
names = list(reps)
# Union-find near-duplicates.
parent = {n: n for n in names}
def find(x: str) -> str:
while parent[x] != x:
parent[x] = parent[parent[x]]; x = parent[x]
return x
lim = math.log(1.0 + dur_tol)
for i in range(len(names)):
for j in range(i + 1, len(names)):
a, b = names[i], names[j]
(ca, da), (cb, db) = reps[a], reps[b]
if da <= 0 or db <= 0 or abs(math.log(da / db)) > lim:
continue
if self._shape_similarity(ca, cb) > sim_min:
parent[find(a)] = find(b)
clusters: dict[str, list[str]] = {}
for n in names:
clusters.setdefault(find(n), []).append(n)
# Only clusters of >=2; annotate with any existing group overlap; skip
# clusters already fully contained in one group.
out: list[dict[str, Any]] = []
groups = self.get_profile_groups()
for members in clusters.values():
if len(members) < 2:
continue
existing = None
for gname, g in groups.items():
gmembers = set(g.get("members") or [])
if gmembers & set(members):
existing = gname
if set(members) <= gmembers:
members = [] # already grouped
break
if members:
out.append({"members": sorted(members), "existing_group": existing})
return out
# ------------------------------------------------------------------
# A1: Underrun anomaly helpers
# ------------------------------------------------------------------
def get_profile_median_duration(self, profile_name: str) -> float | None:
"""Median cycle duration (s) for this profile across all labeled cycles. Never raises."""
try:
durations = [
float(c["duration"])
for c in self.get_past_cycles()
if c.get("profile_name") == profile_name and c.get("duration")
]
return float(np.median(durations)) if len(durations) >= 2 else None
except Exception: # noqa: BLE001
return None
# ------------------------------------------------------------------
# A2: Energy anomaly helpers
# ------------------------------------------------------------------
def get_profile_energy_stats(self, profile_name: str) -> dict[str, float] | None:
"""Energy stats {avg_wh, std_wh, n} for this profile. None if fewer than 3 cycles. Never raises."""
try:
energies = [
float(c["energy_wh"])
for c in self.get_past_cycles()
if c.get("profile_name") == profile_name and c.get("energy_wh")
]
if len(energies) < 3:
return None
arr = np.asarray(energies, dtype=float)
return {"avg_wh": float(np.mean(arr)), "std_wh": float(np.std(arr)), "n": len(energies)}
except Exception: # noqa: BLE001
return None
# ------------------------------------------------------------------
# A4: Profile warm-up mode helper
# ------------------------------------------------------------------
def get_profile_labeled_count(self, profile_name: str) -> int:
"""Number of labeled cycles for this profile. Never raises."""
try:
return sum(1 for c in self.get_past_cycles() if c.get("profile_name") == profile_name)
except Exception: # noqa: BLE001
return 0
def profile_has_reference_cycles(self, profile_name: str) -> bool:
"""True if this profile was seeded from imported store reference cycle(s).
Such a profile is a trusted, community-shared template that the user
downloaded to match immediately, so it is exempt from the local warm-up
gate (which exists to stabilise profiles built from a few local cycles).
"""
try:
return any(
c.get("profile_name") == profile_name
for c in self.get_reference_cycles()
)
except Exception: # noqa: BLE001
return False
# ------------------------------------------------------------------
# B1: Lifetime energy accumulator (HA Energy dashboard)
# ------------------------------------------------------------------
def get_lifetime_energy_wh(self) -> float:
"""Total accumulated lifetime energy (Wh). Pure getter — never persists. Never raises."""
try:
return float(self._data.get("lifetime_energy_wh", 0.0))
except Exception: # noqa: BLE001
return 0.0
async def async_add_lifetime_energy_wh(self, wh: float) -> None:
"""Add *wh* (clamped >= 0) to the lifetime energy total and persist.
Called exactly once per completed cycle so it never double-counts. Does
not backfill from history — the meter starts at 0 and accumulates forward.
Ignores non-numeric input.
"""
try:
add = max(0.0, float(wh))
except (ValueError, TypeError):
return
base = self.get_lifetime_energy_wh()
self._data["lifetime_energy_wh"] = round(base + add, 3)
await self.async_save()
def get_lifetime_cycle_count(self) -> int:
"""Persisted monotonic lifetime completed-cycle count.
Only ever increments (never regresses when history is trimmed/merged), so it
is the correct basis for milestone crossings. Pure getter - never persists.
Never raises; returns 0 when unset.
"""
try:
return int(self._data.get("lifetime_cycle_count", 0) or 0)
except Exception: # noqa: BLE001
return 0
def set_lifetime_cycle_count(self, count: int) -> None:
"""Set the lifetime cycle count in memory WITHOUT persisting.
The cycle-end path batches this with the immediately-following lifetime
energy save (:meth:`async_add_lifetime_energy_wh`) so the store is written
once per cycle. Encapsulates the storage key so callers need not touch
``_data`` directly. Ignores non-integer input.
"""
try:
self._data["lifetime_cycle_count"] = int(count)
except (TypeError, ValueError):
pass
# ------------------------------------------------------------------
# E1: Maintenance log & predictive-maintenance reminders (Group E)
# ------------------------------------------------------------------
def get_maintenance_log(self) -> list[dict[str, Any]]:
"""Return logged maintenance events, most-recent-first. Never raises.
Each entry is ``{"id", "date", "event_type", "notes"}``. Entries with an
unparseable date sort last.
"""
try:
raw = self._data.get("maintenance_log", [])
if not isinstance(raw, list):
return []
entries = [dict(e) for e in raw if isinstance(e, dict)]
_floor = datetime.min.replace(tzinfo=dt_util.UTC)
entries.sort(
key=lambda e: _parse_maintenance_dt(e.get("date")) or _floor,
reverse=True,
)
return entries
except Exception: # noqa: BLE001
return []
async def async_add_maintenance_event(
self, event_type: str, date: str | None = None, notes: str = ""
) -> dict[str, Any]:
"""Append a maintenance event and persist. Returns the created entry.
``event_type`` must be one of :data:`MAINTENANCE_EVENT_TYPES` (else raises
``ValueError``). ``date`` defaults to the current timestamp; a short unique
id is generated for the entry.
"""
if event_type not in MAINTENANCE_EVENT_TYPES:
raise ValueError(f"Unknown maintenance event_type: {event_type!r}")
entry: dict[str, Any] = {
"id": uuid.uuid4().hex[:12],
"date": date if isinstance(date, str) and date else dt_util.now().isoformat(),
"event_type": event_type,
"notes": str(notes or ""),
}
log = self._data.setdefault("maintenance_log", [])
if not isinstance(log, list):
log = []
self._data["maintenance_log"] = log
log.append(entry)
await self.async_save()
return entry
async def async_delete_maintenance_event(self, event_id: str) -> bool:
"""Remove a maintenance event by id, persist, and report whether removed."""
log = self._data.get("maintenance_log")
if not isinstance(log, list):
return False
remaining = [e for e in log if not (isinstance(e, dict) and e.get("id") == event_id)]
if len(remaining) == len(log):
return False
self._data["maintenance_log"] = remaining
await self.async_save()
return True
def cycles_since_maintenance(self, event_type: str) -> int:
"""Count completed cycles since the most recent maintenance event of a type.
Counts completed cycles (``status == "completed"``) whose ``start_time`` is
after the most recent maintenance event of ``event_type``. If no such event
was ever logged, returns the total completed-cycle count. Never raises.
"""
try:
completed = [
c for c in self.get_past_cycles()
if isinstance(c, dict) and c.get("status") == "completed"
]
latest_dt: datetime | None = None
for e in self._data.get("maintenance_log", []) or []:
if not isinstance(e, dict) or e.get("event_type") != event_type:
continue
dt = _parse_maintenance_dt(e.get("date"))
if dt is not None and (latest_dt is None or dt > latest_dt):
latest_dt = dt
if latest_dt is None:
return len(completed)
count = 0
for c in completed:
start = _parse_start_dt(c.get("start_time"))
if start is not None and start > latest_dt:
count += 1
return count
except Exception: # noqa: BLE001
return 0
def get_maintenance_due(self, reminder_cfg: dict[str, Any] | None) -> list[str]:
"""Return event types whose cycles-since threshold has been reached.
For each event type with a positive integer threshold in ``reminder_cfg``,
includes it when ``cycles_since_maintenance(event_type) >= threshold``.
Never raises.
"""
try:
if not isinstance(reminder_cfg, dict):
return []
due: list[str] = []
for event_type, threshold in reminder_cfg.items():
try:
thr = int(threshold)
except (TypeError, ValueError):
continue
if thr <= 0:
continue
if self.cycles_since_maintenance(str(event_type)) >= thr:
due.append(str(event_type))
return due
except Exception: # noqa: BLE001
return []
def has_recent_maintenance(
self, event_type: str, days: int = MAINTENANCE_RECENT_SUPPRESS_DAYS
) -> bool:
"""True if a matching maintenance event was logged within ``days``. Never raises."""
try:
cutoff = dt_util.now() - timedelta(days=max(0, int(days)))
for e in self._data.get("maintenance_log", []) or []:
if not isinstance(e, dict) or e.get("event_type") != event_type:
continue
dt = _parse_maintenance_dt(e.get("date"))
if dt is not None and dt >= cutoff:
return True
return False
except Exception: # noqa: BLE001
return False
def compute_profile_health(self) -> dict[str, dict[str, Any]]:
"""Compute per-profile health indicators from labeled cycle history.
For each profile that has at least 3 labeled cycles, returns a dict with:
``cycle_count`` – number of labeled cycles used in the calculation
``confidence_mean`` – mean match_confidence across those cycles (0–1)
``duration_cv`` – coefficient of variation of cycle durations (lower = consistent)
``health_score`` – composite 0–1 health score (1 = healthy, 0 = needs attention)
``health_status`` – "healthy" / "fair" / "poor" / "unknown"
``shape_drift`` – True when the early/recent envelope correlation is below
SHAPE_DRIFT_THRESHOLD (only present when >= SHAPE_DRIFT_MIN_CYCLES
labeled cycles with power_data exist)
``shape_drift_correlation`` – Pearson r between early and recent average envelopes
Profiles with fewer than 3 labeled cycles return ``health_status="unknown"``.
Never raises — returns an empty dict on any error.
"""
try:
cycles = self.get_past_cycles()
from collections import defaultdict # pylint: disable=import-outside-toplevel
by_profile: dict[str, list[dict[str, Any]]] = defaultdict(list)
for c in cycles:
name = c.get("profile_name")
if name:
by_profile[str(name)].append(c)
result: dict[str, dict[str, Any]] = {}
for name, pcy in by_profile.items():
count = len(pcy)
durations = [float(c["duration"]) for c in pcy if c.get("duration")]
confidences = [
float(c["match_confidence"])
for c in pcy
if c.get("match_confidence") is not None # keep genuine 0.0, drop only absent
]
if count < 3 or not durations:
result[name] = {"cycle_count": count, "health_status": "unknown"}
continue
dur_arr = np.asarray(durations, dtype=float)
dur_mean = float(np.mean(dur_arr))
dur_cv = float(np.std(dur_arr) / dur_mean) if dur_mean > 0 else 0.0
conf_mean = float(np.mean(confidences)) if confidences else 0.5
# consistency: 1 at CV=0, linearly decays to 0 at CV=0.5+
consistency = max(0.0, 1.0 - dur_cv / 0.5)
health_score = round(0.5 * consistency + 0.5 * conf_mean, 3)
if health_score >= 0.65:
status = "healthy"
elif health_score >= 0.40:
status = "fair"
else:
status = "poor"
# A5: Shape drift — compare early vs recent power curve envelopes.
_sd: dict[str, Any] = {}
_traced = [c for c in pcy if c.get("power_data")]
if len(_traced) >= SHAPE_DRIFT_MIN_CYCLES:
_third = len(_traced) // 3
_early = _traced[:_third]
_recent = _traced[-_third:]
try:
from .signal_processing import resample_to_n as _resamp # noqa: PLC0415
def _avg_env(cycles: list[dict[str, Any]]) -> "np.ndarray | None":
_traces = []
for _c in cycles:
_raw = decompress_power_data(_c)
if not _raw:
continue
_pwr = [float(_p) for _, _p in _raw]
if len(_pwr) < 5:
continue
_t = np.asarray(_resamp(_pwr, SHAPE_DRIFT_RESAMPLE_N), dtype=float)
_mx = _t.max()
if _mx > 0:
_t = _t / _mx
_traces.append(_t)
if len(_traces) < 2:
return None
return np.mean(np.stack(_traces), axis=0)
_early_env = _avg_env(_early)
_rec_env = _avg_env(_recent)
if _early_env is not None and _rec_env is not None:
_corr = float(np.corrcoef(_early_env, _rec_env)[0, 1])
if not np.isfinite(_corr):
_corr = 1.0
_sd = {
"shape_drift": _corr < SHAPE_DRIFT_THRESHOLD,
"shape_drift_correlation": round(_corr, 3),
}
except Exception: # noqa: BLE001
pass
result[name] = {
"cycle_count": count,
"confidence_mean": round(conf_mean, 3),
"duration_cv": round(dur_cv, 3),
"health_score": health_score,
"health_status": status,
**_sd, # merges shape_drift and shape_drift_correlation when available
}
return result
except Exception: # noqa: BLE001
return {}
# ------------------------------------------------------------------
# Coverage gap detection (unmatched cycle clustering)
# ------------------------------------------------------------------
def suggest_coverage_gaps(
self,
recent_window: int = 30,
min_unmatched: int = 5,
min_unmatched_rate: float = 0.20,
low_confidence_threshold: float = 0.40,
duration_bucket_s: float = 900.0,
) -> dict[str, Any]:
"""Detect gaps in profile coverage from recent unlabelled cycles.
Looks at the ``recent_window`` most recent cycles. When there are
enough unmatched (no ``profile_name``) or low-confidence cycles,
returns a record with:
``unmatched_count`` – number of cycles with no profile label
``low_confidence_count`` – number of labelled cycles below the confidence
threshold (potential mis-matches)
``unmatched_rate`` – fraction of recent cycles that are unmatched
``suggest_create`` – True when unmatched_count >= min_unmatched
AND unmatched_rate >= min_unmatched_rate
``duration_clusters`` – list of duration-cluster dicts for unmatched
cycles: ``{"duration_bucket_min": int,
"count": int, "avg_duration_s": float}``
sorted descending by count. Only clusters
with ≥ 2 members are returned.
Returns an empty dict when there are fewer than ``min_unmatched``
unmatched cycles or on any error (never raises).
"""
try:
cycles = self.get_past_cycles()
recent = cycles[-recent_window:] if len(cycles) > recent_window else cycles
if not recent:
return {}
unmatched: list[dict[str, Any]] = []
low_conf: list[dict[str, Any]] = []
for c in recent:
name = c.get("profile_name")
if not name:
unmatched.append(c)
else:
conf = c.get("match_confidence")
if (
isinstance(conf, (int, float))
and not isinstance(conf, bool)
and conf < low_confidence_threshold
):
low_conf.append(c)
n_unmatched = len(unmatched)
n_total = len(recent)
rate = n_unmatched / n_total if n_total > 0 else 0.0
if n_unmatched < min_unmatched:
return {}
# Cluster unmatched cycles by duration bucket (simple histogram)
bucket_dur: dict[int, list[float]] = {}
for c in unmatched:
dur = c.get("duration")
if isinstance(dur, (int, float)) and not isinstance(dur, bool) and dur > 0:
bucket = int(dur // duration_bucket_s)
bucket_dur.setdefault(bucket, []).append(float(dur))
clusters = []
for bucket, durs in sorted(bucket_dur.items(), key=lambda kv: -len(kv[1])):
if len(durs) >= 2:
clusters.append({
"duration_bucket_min": int(bucket * duration_bucket_s / 60),
"count": len(durs),
"avg_duration_s": round(float(np.mean(durs)), 1),
})
# A3: Shape-similarity clustering within duration buckets
profile_suggestions: list[dict[str, Any]] = []
for bucket, durs in sorted(bucket_dur.items(), key=lambda kv: -len(kv[1])):
if len(durs) < 2:
continue
bucket_cycles = [
c for c in unmatched
if c.get("power_data")
and c.get("duration")
and int(float(c["duration"]) // duration_bucket_s) == bucket
]
if len(bucket_cycles) < 2:
continue
try:
from .signal_processing import resample_to_n # noqa: PLC0415
traces: list[np.ndarray] = []
ids: list[str] = []
for c in bucket_cycles[:5]: # cap at 5 per bucket for performance
raw = decompress_power_data(c)
if not raw:
continue
pwr = [float(p) for _, p in raw]
if len(pwr) < 5:
continue
t = np.asarray(resample_to_n(pwr, CLUSTER_RESAMPLE_N), dtype=float)
mx = t.max()
if mx > 0:
t = t / mx
traces.append(t)
ids.append(str(c.get("id", "")))
if len(traces) < 2:
continue
corrs = [
float(np.corrcoef(traces[i], traces[j])[0, 1])
for i in range(len(traces))
for j in range(i + 1, len(traces))
]
valid_corrs = [r for r in corrs if np.isfinite(r)]
if not valid_corrs:
continue
avg_corr = float(np.mean(valid_corrs))
if avg_corr >= CLUSTER_SHAPE_SIMILARITY_THRESHOLD:
avg_dur_s = float(np.mean(durs))
profile_suggestions.append({
"suggested_name": f"~{int(avg_dur_s // 60)} min program",
"cycle_ids": [cid for cid in ids if cid],
"avg_duration_s": round(avg_dur_s, 1),
"count": len(ids),
"similarity": round(avg_corr, 3),
})
except Exception: # noqa: BLE001
continue
# Most-recent unmatched cycle id, so the setup advisor's phase-2
# "unmatched" nudge can deep-link straight to it (open_cycle:)
# instead of always falling back to the whole unlabelled list.
last_unmatched_cycle_id = None
for c in reversed(unmatched):
cid = c.get("id")
if cid:
last_unmatched_cycle_id = cid
break
return {
"unmatched_count": n_unmatched,
"low_confidence_count": len(low_conf),
"unmatched_rate": round(rate, 3),
"suggest_create": bool(n_unmatched >= min_unmatched and rate >= min_unmatched_rate),
"duration_clusters": clusters,
"profile_suggestions": profile_suggestions, # NEW (A3)
"last_unmatched_cycle_id": last_unmatched_cycle_id,
}
except Exception: # noqa: BLE001
return {}
# ------------------------------------------------------------------
# Profile trend analysis (pure stats, appliance performance drift)
# ------------------------------------------------------------------
def compute_profile_trends(
self,
min_cycles: int = 12,
recent_window: int = 8,
slope_threshold_pct: float = 0.08,
) -> dict[str, dict[str, Any]]:
"""Detect per-profile duration / energy drift using ordinary least-squares.
Uses all labeled cycles in chronological order. For each profile with at
least ``min_cycles`` labeled cycles:
* Fits a linear trend (OLS via NumPy) to the most-recent N=``len(cycles)``
values (not just the window — the window is used only for the *recent
mean* display; the slope is computed on all data so short histories don't
give noisy slopes).
* Expresses the slope as percent-change-per-cycle relative to the mean,
so the magnitude is comparable across low- and high-duration profiles.
* Classifies as ``"up"`` / ``"down"`` / ``"stable"`` per metric when the
normalized slope exceeds ``slope_threshold_pct`` (default 8%/cycle).
Returns a dict mapping profile_name → trend record:
``duration_trend`` – "up" / "down" / "stable"
``duration_slope_pct`` – normalized slope (%/cycle)
``duration_recent_mean_s`` – mean duration of the last N cycles (s)
``energy_trend`` – "up" / "down" / "stable" (if available)
``energy_slope_pct`` – normalized slope (%/cycle)
``energy_recent_mean_wh`` – mean energy of the last N cycles (Wh)
``cycle_count`` – number of labeled cycles used
``recent_window`` – how many cycles the recent means cover
Profiles with fewer than ``min_cycles`` labeled cycles are omitted.
Never raises — returns ``{}`` on any error.
"""
try:
cycles = self.get_past_cycles()
from collections import defaultdict # pylint: disable=import-outside-toplevel
by_profile: dict[str, list[dict[str, Any]]] = defaultdict(list)
for c in cycles:
name = c.get("profile_name")
if name:
by_profile[str(name)].append(c)
result: dict[str, dict[str, Any]] = {}
for name, pcy in by_profile.items():
if len(pcy) < min_cycles:
continue
# Preserve insertion order (chronological from get_past_cycles)
durations = [float(c["duration"]) for c in pcy if c.get("duration")]
energies = [float(c["energy_wh"]) for c in pcy if c.get("energy_wh")]
if len(durations) < min_cycles:
continue
def _slope_pct(values: list[float]) -> float:
"""OLS slope as % of mean per cycle."""
arr = np.asarray(values, dtype=float)
n = len(arr)
x = np.arange(n, dtype=float)
x -= x.mean()
denom = float(np.dot(x, x))
if denom < 1e-9:
return 0.0
slope = float(np.dot(x, arr)) / denom
mean_val = float(np.mean(arr))
return (slope / mean_val) if abs(mean_val) > 1e-9 else 0.0
def _classify(slope_pct: float) -> str:
if slope_pct > slope_threshold_pct:
return "up"
if slope_pct < -slope_threshold_pct:
return "down"
return "stable"
dur_slope = _slope_pct(durations)
dur_recent = float(np.mean(durations[-recent_window:]))
rec: dict[str, Any] = {
"cycle_count": len(pcy),
"recent_window": min(recent_window, len(durations)),
"duration_trend": _classify(dur_slope),
"duration_slope_pct": round(dur_slope * 100, 2),
"duration_recent_mean_s": round(dur_recent, 1),
}
if len(energies) >= min_cycles:
en_slope = _slope_pct(energies)
en_recent = float(np.mean(energies[-recent_window:]))
rec["energy_trend"] = _classify(en_slope)
rec["energy_slope_pct"] = round(en_slope * 100, 2)
rec["energy_recent_mean_wh"] = round(en_recent, 1)
result[name] = rec
return result
except Exception: # noqa: BLE001
return {}
def compute_profile_advisories(self) -> list[dict[str, Any]]:
"""Actionable per-profile maintenance advisories from existing signals.
Pure statistics (no ML): consolidates :meth:`compute_profile_health` and
:meth:`compute_profile_trends` into a small ranked list of human-readable,
actionable recommendations (e.g. "durations trending longer -> re-record").
These surface as recommendations in the Profiles tab - never a
notification. Each item is ``{profile, severity, code, message}``; severity
is ``"warning"`` or ``"info"``. Returns ``[]`` on error (never raises).
"""
try:
advisories: list[dict[str, Any]] = []
health = self.compute_profile_health() or {}
trends = self.compute_profile_trends() or {}
for name, h in health.items():
if h.get("health_status") == "poor":
advisories.append({
"profile": name, "severity": "warning", "code": "poor_health",
"message": (
f"'{name}' has a low fit score - its recent cycles vary "
"a lot or match weakly. Review its cycles or re-record "
"the profile so matching and time estimates stay accurate."
),
# Localization: panel renders _t(message_key, message_params,
# message). The English `message` above is the fallback.
"message_key": "msg.advisory_poor_health",
"message_params": {"name": name},
})
elif h.get("shape_drift"):
corr = h.get("shape_drift_correlation")
corr_str = f" (correlation {corr:.2f})" if corr is not None else ""
advisories.append({
"profile": name, "severity": "info", "code": "shape_drift",
"message": (
f"'{name}' has drifted significantly from its original "
f"power shape{corr_str}. The appliance may have changed "
"behaviour over time (e.g. limescale, wear). Consider "
"re-recording this profile with recent cycles."
),
"message_key": (
"msg.advisory_shape_drift_corr"
if corr is not None
else "msg.advisory_shape_drift"
),
"message_params": (
{"name": name, "corr": f"{corr:.2f}"}
if corr is not None
else {"name": name}
),
})
for name, t in trends.items():
# Skip profiles already flagged poor to avoid double advice.
if health.get(name, {}).get("health_status") == "poor":
continue
if t.get("duration_trend") == "up":
# duration_slope_pct is already expressed as %/cycle by
# compute_profile_trends (dur_slope * 100); do NOT re-scale.
pct = abs(float(t.get("duration_slope_pct") or 0.0))
advisories.append({
"profile": name, "severity": "info", "code": "duration_trend_up",
"message": (
f"'{name}' cycles are running progressively longer "
f"(about +{pct:.0f}% per cycle). If the appliance's "
"behaviour changed, re-record or rebuild this profile."
),
"message_key": "msg.advisory_duration_trend_up",
"message_params": {"name": name, "pct": f"{pct:.0f}"},
})
elif t.get("energy_trend") == "up":
# energy_slope_pct is already %/cycle (en_slope * 100); no re-scale.
pct = abs(float(t.get("energy_slope_pct") or 0.0))
advisories.append({
"profile": name, "severity": "info", "code": "energy_trend_up",
"message": (
f"'{name}' is drawing progressively more energy "
f"(about +{pct:.0f}% per cycle) - worth checking the "
"appliance if that is unexpected."
),
"message_key": "msg.advisory_energy_trend_up",
"message_params": {"name": name, "pct": f"{pct:.0f}"},
})
# Phase-structure consistency (phase-matching device types only). Uses
# the cached per-role phase profile: a profile whose member cycles heat
# for wildly different times, or where only some cycles heat at all,
# most likely mixes different programs/temperatures under one label -
# which hurts both matching and the phase-resolved ETA. Advisory only
# (Profiles tab); no relabeling (phase matching does not label better).
_sd = getattr(self, "_data", None)
_envs = _sd.get("envelopes") if isinstance(_sd, dict) else None
for pname, penv in (_envs if isinstance(_envs, dict) else {}).items():
if health.get(pname, {}).get("health_status") == "poor":
continue # avoid double advice
pp = penv.get("phase_profile") if isinstance(penv, dict) else None
if not isinstance(pp, dict):
continue
try:
if int(pp.get("n_cycles") or 0) < PHASE_CONSISTENCY_MIN_CYCLES:
continue
heat = (pp.get("roles") or {}).get("heating") or {}
heat_mean = float(heat.get("dur_mean") or 0.0)
heat_std = float(heat.get("dur_std") or 0.0)
heat_occ = float(heat.get("occurrence") or 0.0)
heat_cv = (heat_std / heat_mean) if heat_mean > 60.0 else 0.0
mixed_temp = heat_cv > PHASE_HEAT_CV_WARN
mixed_prog = PHASE_HEAT_OCC_MIXED_LO <= heat_occ <= PHASE_HEAT_OCC_MIXED_HI
if not (mixed_temp or mixed_prog):
continue
advisories.append({
"profile": pname, "severity": "warning",
"code": "phase_inconsistent",
"message": (
f"'{pname}' looks like it mixes different programs or "
"temperatures - its cycles heat for very different lengths "
"of time. Splitting it into separate profiles (e.g. per "
"temperature) will improve matching and time estimates."
),
"message_key": "msg.advisory_phase_inconsistent",
"message_params": {"name": pname},
})
except (TypeError, ValueError):
continue
# E1: suppress the "needs maintenance" nag (duration-trending-longer /
# shape-drift/poor-fit) when the user recently logged a descale, filter
# clean, or drum clean — any recent maintenance clears the reminder.
try:
# ``is True`` (not truthiness) so a mocked store whose method
# returns a MagicMock does not accidentally suppress advisories.
if any(
self.has_recent_maintenance(evt) is True
for evt in ("descale", "filter_clean", "drum_clean")
):
_nag_codes = {"duration_trend_up", "poor_health", "shape_drift"}
advisories = [a for a in advisories if a.get("code") not in _nag_codes]
except Exception: # noqa: BLE001
pass
order = {"warning": 0, "info": 1}
advisories.sort(key=lambda a: order.get(a["severity"], 2))
return advisories
except Exception: # noqa: BLE001
return []
# ------------------------------------------------------------------
# Match ranking history (per-cycle ML commit snapshots)
# ------------------------------------------------------------------
def record_match_ranking_snapshot(
self,
start_time_iso: str,
features: dict[str, float],
top1_profile: str,
top1_score: float,
top2_score: float | None,
candidate_count: int,
cycle_id: str = "",
) -> None:
"""Append a live-match ranking snapshot for the active cycle.
Snapshots are keyed by ``start_time_iso`` (and optionally ``cycle_id``) so
that ``confirm_match_ranking_snapshots`` can back-fill the confirmed label
when the cycle ends. Pre-computed feature scalars are stored (not raw traces)
to keep footprint small. The store is NOT persisted here — the caller must
schedule ``async_save``.
"""
from .const import MATCH_RANKING_HISTORY_MAX # pylint: disable=import-outside-toplevel
history: list[dict[str, Any]] = self._data.setdefault("match_ranking_history", [])
history.append({
"start_time_iso": start_time_iso,
"cycle_id": str(cycle_id) if cycle_id else "",
"features": dict(features),
"top1_profile": str(top1_profile),
"top1_score": round(float(top1_score), 4),
"top2_score": round(float(top2_score), 4) if top2_score is not None else None,
"candidate_count": int(candidate_count),
"confirmed_label": None,
})
# Trim to the most recent N snapshots.
if len(history) > MATCH_RANKING_HISTORY_MAX:
del history[: len(history) - MATCH_RANKING_HISTORY_MAX]
def confirm_match_ranking_snapshots(
self,
start_time_iso: str,
confirmed_label: str,
cycle_id: str = "",
) -> int:
"""Back-fill the confirmed label for all snapshots belonging to a cycle.
When ``cycle_id`` is provided (and the snapshot was recorded with one), matches
by ``cycle_id`` to avoid cross-contamination between cycles that share the same
second-resolution ``start_time_iso``. Falls back to ``start_time_iso`` matching
for snapshots recorded without a cycle_id (backward compatibility).
Returns the number of snapshots updated.
"""
history = self._data.get("match_ranking_history")
if not isinstance(history, list):
return 0
updated = 0
for snap in history:
if not isinstance(snap, dict):
continue
snap_cid = snap.get("cycle_id") or ""
if cycle_id and snap_cid:
# Both sides have an ID — match by ID for precision.
if snap_cid == cycle_id:
snap["confirmed_label"] = str(confirmed_label)
updated += 1
else:
# Legacy path: match by timestamp (no cycle_id on one or both sides).
if snap.get("start_time_iso") == start_time_iso:
snap["confirmed_label"] = str(confirmed_label)
updated += 1
return updated
def get_match_ranking_history(self) -> list[dict[str, Any]]:
"""Return all ranking snapshots (confirmed and pending).
Callers that build a training dataset should filter to snapshots where
``confirmed_label`` is not None.
"""
raw = self._data.get("match_ranking_history")
return list(raw) if isinstance(raw, list) else []
def _get_shared_custom_phases(self) -> list[dict[str, Any]]:
"""Return mutable shared custom phase list with legacy flattening."""
raw = self._data.setdefault("custom_phases", [])
if isinstance(raw, list):
return cast(list[dict[str, Any]], raw)
# Legacy format: {device_type: [phase, ...]}. Flatten to shared list.
flattened: list[dict[str, Any]] = []
seen: set[tuple[str, str]] = set()
if isinstance(raw, dict):
raw_dict = cast(dict[str, Any], raw)
for legacy_device_type, phase_list in raw_dict.items():
if not isinstance(phase_list, list):
continue
for item in cast(list[Any], phase_list):
if not isinstance(item, dict):
continue
item_dict = cast(dict[str, Any], item)
name = str(item_dict.get("name", "")).strip()
if not name:
continue
device_type = str(legacy_device_type or "").strip()
key = (name.casefold(), device_type.casefold())
if key in seen:
continue
seen.add(key)
flattened.append(
{
"name": name,
"description": str(item_dict.get("description", "")).strip(),
"device_type": device_type,
"created_at": item_dict.get("created_at") or dt_util.now().isoformat(),
}
)
self._data["custom_phases"] = flattened
return self._data["custom_phases"]
def list_custom_phases(self, device_type: str) -> list[dict[str, Any]]:
"""Return shared custom phases relevant to the requested device type."""
def applies_to_device(item_device_type: str, target_device_type: str) -> bool:
if not item_device_type:
return True
return item_device_type == target_device_type
target = str(device_type or "").strip()
phases = self._get_shared_custom_phases()
return [
{
"id": str(p.get("id", "")),
"name": str(p.get("name", "")).strip(),
"description": str(p.get("description", "")).strip(),
"device_type": str(p.get("device_type", "")).strip(),
"is_default": False,
}
for p in phases
if p.get("name")
and applies_to_device(str(p.get("device_type", "")).strip(), target)
]
def list_phase_catalog(self, device_type: str) -> list[dict[str, Any]]:
"""Return merged shared default + custom phase catalog."""
return merge_phase_catalog(device_type, self.list_custom_phases(device_type))
async def async_create_custom_phase(
self,
device_type: str,
phase_name: str,
description: str = "",
) -> None:
"""Create a custom phase in the shared catalog."""
target_device_type = str(device_type or "").strip()
name = normalize_phase_name(phase_name)
desc = str(description or "").strip()
catalog = self.list_phase_catalog(target_device_type)
if any(str(p.get("name", "")).casefold() == name.casefold() for p in catalog):
raise ValueError("duplicate_phase")
self._get_shared_custom_phases().append(
{
"id": str(uuid.uuid4()),
"name": name,
"description": desc,
"device_type": target_device_type,
"created_at": dt_util.now().isoformat(),
}
)
await self.async_save()
async def async_update_custom_phase(
self,
phase_id: str,
new_name: str,
description: str = "",
) -> None:
"""Update a phase by id, propagating rename to profile assignments.
If phase_id matches a built-in, a custom override is created using the
built-in's id so the merge can replace it in-place.
"""
target_name = normalize_phase_name(new_name)
desc = str(description or "").strip()
phases = self._get_shared_custom_phases()
# Look for an existing custom entry with this id.
found: dict[str, Any] | None = next(
(p for p in phases if str(p.get("id", "")) == phase_id), None
)
creating_new = found is None
if creating_new:
builtin = get_builtin_phase_by_id(phase_id)
if builtin is None:
raise ValueError("phase_not_found")
candidate: dict[str, Any] = {
"id": phase_id,
"name": str(builtin.get("name", "")),
"description": "",
"device_type": str(builtin.get("device_type", "")),
"created_at": dt_util.now().isoformat(),
}
else:
candidate = found # type: ignore[assignment]
old_name = str(candidate.get("name", ""))
target_device_type = str(candidate.get("device_type", "")).strip()
# Duplicate-name check before any mutation.
for p in self.list_phase_catalog(target_device_type):
pname = str(p.get("name", ""))
if pname.casefold() == target_name.casefold() and pname.casefold() != old_name.casefold():
raise ValueError("duplicate_phase")
if creating_new:
phases.append(candidate)
found = candidate
found["name"] = target_name # type: ignore[index]
found["description"] = desc # type: ignore[index]
# Propagate rename to profile assignments.
for profile in self.get_profiles().values():
profile_device_type = str(profile.get("device_type", "")).strip()
if target_device_type and profile_device_type != target_device_type:
continue
phases_assigned = profile.get("phases", [])
if not isinstance(phases_assigned, list):
continue
for assigned in cast(list[dict[str, Any]], phases_assigned):
if str(assigned.get("name", "")).casefold() == old_name.casefold():
assigned["name"] = target_name
await self.async_save()
async def async_delete_custom_phase(self, phase_id: str) -> int:
"""Delete a custom phase by id and remove matching profile assignments.
Returns number of removed assignments.
Raises ValueError('phase_not_found') if no custom phase has this id.
Raises ValueError('cannot_delete_builtin') if the id is a built-in phase.
"""
phases = self._get_shared_custom_phases()
found = next((p for p in phases if str(p.get("id", "")) == phase_id), None)
if found is None:
raise ValueError("phase_not_found")
if get_builtin_phase_by_id(phase_id) is not None:
raise ValueError("cannot_delete_builtin")
phase_name = str(found.get("name", ""))
phase_scope = str(found.get("device_type", "")).strip()
self._data["custom_phases"] = [p for p in phases if str(p.get("id", "")) != phase_id]
removed_assignments = 0
for profile in self.get_profiles().values():
profile_device_type = str(profile.get("device_type", "")).strip()
if phase_scope and profile_device_type != phase_scope:
continue
assigned = profile.get("phases", [])
if not isinstance(assigned, list):
continue
assigned_list = cast(list[dict[str, Any]], assigned)
before = len(assigned_list)
profile["phases"] = [
p for p in assigned_list
if str(p.get("name", "")).casefold() != phase_name.casefold()
]
removed_assignments += before - len(profile["phases"])
await self.async_save()
return removed_assignments
def get_profile_phase_ranges(self, profile_name: str) -> list[dict[str, Any]]:
"""Return assigned phase ranges for a profile."""
profiles = self._data.get("profiles", {})
if not isinstance(profiles, dict):
return []
profile_raw = cast(dict[str, Any], profiles).get(profile_name)
profile = cast(dict[str, Any], profile_raw) if isinstance(profile_raw, dict) else None
if not isinstance(profile, dict):
return []
phases = profile.get("phases", [])
if not isinstance(phases, list):
return []
phases_list = cast(list[dict[str, Any]], phases)
cleaned: list[dict[str, Any]] = []
for phase in phases_list:
try:
start = float(phase.get("start", 0.0))
end = float(phase.get("end", 0.0))
except (TypeError, ValueError):
continue
if end <= start:
continue
cleaned.append(
{
"name": str(phase.get("name", "")).strip(),
"start": start,
"end": end,
"description": str(phase.get("description", "")).strip(),
}
)
return sorted(cleaned, key=lambda x: (x["start"], x["end"], x["name"]))
def get_profile_phase_ranges_for_device(
self, profile_name: str, device_type: str
) -> list[dict[str, Any]]:
"""Return assigned ranges enriched with catalog descriptions."""
ranges = self.get_profile_phase_ranges(profile_name)
catalog = self.list_phase_catalog(device_type)
desc_map = {
str(p.get("name", "")).casefold(): str(p.get("description", "")).strip()
for p in catalog
}
enriched: list[dict[str, Any]] = []
for row in ranges:
name = str(row.get("name", "")).strip()
enriched.append(
{
"name": name,
"start": float(row.get("start", 0.0)),
"end": float(row.get("end", 0.0)),
"description": desc_map.get(name.casefold(), ""),
}
)
return enriched
async def async_set_profile_phase_ranges(
self, profile_name: str, ranges: list[dict[str, Any]]
) -> None:
"""Replace assigned phase ranges for a profile."""
profile = self._data.get("profiles", {}).get(profile_name)
if not isinstance(profile, dict):
raise ValueError("profile_not_found")
normalized: list[dict[str, Any]] = []
for item in ranges:
name = normalize_phase_name(str(item.get("name", "")))
try:
start = float(item.get("start", 0.0))
end = float(item.get("end", 0.0))
except (TypeError, ValueError) as e:
raise ValueError("invalid_phase_range") from e
if end <= start:
raise ValueError("invalid_phase_range")
normalized.append({"name": name, "start": start, "end": end})
normalized.sort(key=lambda x: (x["start"], x["end"]))
prev_end = None
for row in normalized:
if prev_end is not None and row["start"] < prev_end:
raise ValueError("overlapping_phase_ranges")
prev_end = row["end"]
profile["phases"] = normalized
await self.async_save()
def set_duration_tolerance(self, tolerance: float) -> None:
"""Set the profile duration tolerance used by matching heuristics."""
try:
self._duration_tolerance = float(tolerance)
except (TypeError, ValueError):
pass
def set_retention_limits(
self,
*,
max_past_cycles: int,
max_full_traces_per_profile: int,
max_full_traces_unlabeled: int,
) -> None:
"""Set retention caps for stored cycles and full-resolution traces."""
try:
self._max_past_cycles = int(max_past_cycles)
self._max_full_traces_per_profile = int(max_full_traces_per_profile)
self._max_full_traces_unlabeled = int(max_full_traces_unlabeled)
except (TypeError, ValueError):
pass
def get_duration_ratio_limits(self) -> tuple[float, float]:
"""Return (min_duration_ratio, max_duration_ratio) used for duration matching."""
return (float(self._min_duration_ratio), float(self._max_duration_ratio))
def set_duration_ratio_limits(self, *, min_ratio: float, max_ratio: float) -> None:
"""Update duration ratio bounds used for duration matching."""
try:
self._min_duration_ratio = float(min_ratio)
self._max_duration_ratio = float(max_ratio)
except (TypeError, ValueError):
pass
def _migrate_phase_ids(self) -> bool:
"""Assign ids to any custom phase missing one. Returns True if anything changed."""
phases = self._data.get("custom_phases", [])
if not isinstance(phases, list):
return False
changed = False
for phase in cast(list[dict[str, Any]], phases):
if phase.get("id"):
continue
dt = str(phase.get("device_type", "")).strip()
name = str(phase.get("name", "")).strip()
matched_id: str | None = None
for bdt, bphases in DEFAULT_PHASES_BY_DEVICE.items():
if dt and bdt != dt:
continue
for bp in bphases:
if str(bp.get("name", "")).strip().casefold() == name.casefold():
matched_id = _builtin_phase_id(bdt, str(bp.get("name", "")))
break
if matched_id:
break
phase["id"] = matched_id if matched_id else str(uuid.uuid4())
changed = True
return changed
async def async_load(self) -> None:
"""Load data from storage with migration."""
# WashDataStore handles migration internally via _async_migrate_func
data = await self._store.async_load()
if data:
self._data = data
# Ensure legacy custom phase formats are normalized in-memory.
self._get_shared_custom_phases()
# Assign ids to any custom phase missing one.
if self._migrate_phase_ids():
await self.async_save()
# Repair cycles whose power_data was corrupted by the double-subtract bug.
if self.repair_corrupted_power_data():
await self.async_save()
await self.async_rebuild_all_envelopes()
await self.async_save()
# Prune pending feedback whose cycle no longer exists, so the device's
# "needs review" count can't disagree with the Cycles review filter.
if self.prune_orphaned_feedback():
await self.async_save()
# _migrate_v1_to_v2 and _decompress_power_from_raw removed; logic moved to WashDataStore
async def async_repair_profile_samples(self) -> dict[str, int]:
"""Repair profile sample references after retention or migrations.
Ensures each profile's sample_cycle_id points to an existing cycle that still
has full-resolution power_data. If missing, picks the newest available cycle
with power_data and assigns it as the sample (and labels that cycle to the
profile if it was unlabeled).
Returns stats dict.
"""
stats = {
"profiles_checked": 0,
"profiles_repaired": 0,
"cycles_labeled_as_sample": 0,
}
profiles: dict[str, dict[str, Any]] = self._data.get("profiles", {}) or {}
cycles: list[dict[str, Any]] = self._data.get("past_cycles", []) or []
ref_cycles: list[dict[str, Any]] = self._data.get("reference_cycles", []) or []
if not profiles or not cycles:
return stats
# Sample validity must recognise imported reference cycles: an import-only
# profile legitimately points its sample at a reference cycle. Without this,
# such a sample looks "missing" and the repair below would steal an unrelated
# unlabeled real cycle into the imported profile.
by_id: dict[str, dict[str, Any]] = {
c["id"]: c for c in list(cycles) + list(ref_cycles) if c.get("id")
}
def newest_unlabeled_with_power_data() -> dict[str, Any] | None:
candidates: list[dict[str, Any]] = [
c for c in cycles if c.get("power_data") and not c.get("profile_name")
]
if not candidates:
return None
try:
return max(candidates, key=lambda c: c.get("start_time", ""))
except Exception: # pylint: disable=broad-exception-caught
return candidates[-1]
for profile_name, profile in profiles.items():
stats["profiles_checked"] += 1
sample_id = profile.get("sample_cycle_id")
sample = by_id.get(sample_id) if sample_id else None
# Sample is valid only if it exists and still has power_data
if sample and sample.get("power_data"):
continue
# Prefer newest already-labeled cycle for this profile that still has power_data
labeled_candidates = [
c
for c in cycles
if c.get("profile_name") == profile_name and c.get("power_data")
]
if labeled_candidates:
try:
chosen = max(
labeled_candidates, key=lambda c: c.get("start_time", "")
)
except Exception: # pylint: disable=broad-exception-caught
chosen = labeled_candidates[-1]
else:
# Fallback: pick newest UNLABELED cycle with power_data
chosen = newest_unlabeled_with_power_data()
if not chosen:
continue
profile["sample_cycle_id"] = chosen.get("id")
if chosen.get("duration"):
profile["avg_duration"] = chosen["duration"]
# If chosen cycle is unlabeled, label it to this profile to bootstrap matching
if not chosen.get("profile_name"):
chosen["profile_name"] = profile_name
stats["cycles_labeled_as_sample"] += 1
stats["profiles_repaired"] += 1
try:
await self.async_rebuild_envelope(profile_name)
except Exception: # pylint: disable=broad-exception-caught
pass
return stats
async def async_save(self) -> None:
"""Save data to storage."""
await self._store.async_save(self._data)
async def async_save_active_cycle(self, detector_snapshot: JSONDict) -> None:
"""Save the active cycle state to storage (throttled by Manager)."""
self._data["active_cycle"] = detector_snapshot
self._data["last_active_save"] = dt_util.now().isoformat()
await self._store.async_save(self._data)
def get_active_cycle(self) -> JSONDict | None:
"""Get the saved active cycle."""
raw = self._data.get("active_cycle")
if isinstance(raw, dict):
return cast(JSONDict, raw)
return None
def get_last_active_save(self) -> datetime | None:
"""Return the last time the active cycle snapshot was persisted."""
raw = self._data.get("last_active_save")
if not isinstance(raw, str) or not raw:
return None
try:
return dt_util.parse_datetime(raw)
except ValueError:
return None
async def async_clear_active_cycle(self) -> None:
"""Clear the active cycle snapshot from storage."""
if "active_cycle" in self._data:
del self._data["active_cycle"]
await self._store.async_save(self._data)
def add_cycle(self, cycle_data: CycleDict) -> None:
"""Add a completed cycle to history (sync wrapper, schedules async tasks)."""
self._add_cycle_data(cycle_data)
self.hass.async_create_task(self.async_enforce_retention())
async def async_add_cycle(self, cycle_data: CycleDict) -> None:
"""Add a completed cycle to history asynchronously."""
self._add_cycle_data(cycle_data)
await self.async_enforce_retention()
def _add_cycle_data(self, cycle_data: CycleDict, target: list[CycleDict] | None = None) -> None:
"""Internal logic to add cycle data to storage.
``target`` defaults to ``past_cycles``; ``add_reference_cycle`` passes the
separate ``reference_cycles`` list so imported cycles never enter usage stats.
"""
dest = self._data["past_cycles"] if target is None else target
# Generate SHA256 ID
unique_str = f"{cycle_data['start_time']}_{cycle_data['duration']}"
cycle_data["id"] = hashlib.sha256(unique_str.encode()).hexdigest()[:12]
# Preserve profile_name if already set by manager; default to None otherwise
if "profile_name" not in cycle_data:
cycle_data["profile_name"] = None # Initially unknown
# Store power data at native sampling resolution
# Format: [seconds_offset, power] preserves actual sample rate from device
# (e.g., 3s intervals from test socket, 60s intervals from real socket)
raw_data: list[Any] = cycle_data.get("power_data", []) or []
self._logger.debug("add_cycle: raw_data has %s points", len(raw_data))
if raw_data:
start_time_raw = cycle_data.get("start_time")
start_time_iso: str | None = None
if start_time_raw is not None:
parsed_dt = _parse_start_dt(start_time_raw)
if parsed_dt is not None:
start_time_iso = parsed_dt.isoformat()
# Keep original ISO string as-is if it was already a valid ISO string
if isinstance(start_time_raw, str) and dt_util.parse_datetime(start_time_raw) is not None:
start_time_iso = start_time_raw
else:
try:
ts = float(start_time_raw)
start_time_iso = dt_util.utc_from_timestamp(ts).isoformat()
except (ValueError, OSError):
self._logger.debug(
"add_cycle: unparseable string start_time %r, falling back",
start_time_raw,
)
elif isinstance(start_time_raw, datetime):
start_time_iso = start_time_raw.isoformat()
elif isinstance(start_time_raw, (int, float)):
try:
start_time_iso = dt_util.utc_from_timestamp(float(start_time_raw)).isoformat()
except (ValueError, OSError):
pass
if start_time_iso is not None:
cycle_data["start_time"] = start_time_iso
if start_time_iso is None and _value_to_timestamp(start_time_raw) is None:
self._logger.debug("add_cycle: invalid start_time %r, skipping power_data normalization", start_time_raw)
if hasattr(self, "_save_debug_traces") and not self._save_debug_traces:
cycle_data.pop("debug_data", None)
dest.append(cycle_data)
return
# Use unified normalizer: handles offset, ISO-string, and datetime formats
pairs = power_data_to_offsets(
cast(list[list[Any] | tuple[Any, ...]], raw_data), start_time_iso
)
stored: list[list[float]] = [[round(p[0], 1), round(p[1], 1)] for p in pairs]
offsets: list[float] = [p[0] for p in stored]
# Calculate average sampling interval (in seconds)
if len(offsets) > 1:
intervals = np.diff(offsets)
positive_intervals = intervals[intervals > 0]
sampling_interval = float(np.median(positive_intervals)) if positive_intervals.size > 0 else 0.0
else:
sampling_interval = 1.0 # Default fallback
# Trim leading/trailing zero readings for cleaner data
# SKIP for completed cycles to preserve end spike data
if cycle_data.get("status") in ("completed", "force_stopped"):
# Only trim leading zeros for completed cycles, keep trailing data
start_idx = 0
for i, point in enumerate(stored):
if point[1] > 1.0:
start_idx = i
break
stored = stored[start_idx:]
self._logger.debug("add_cycle: Skipping trailing trim for completed cycle")
else:
stored = trim_zero_power_data(stored, threshold=1.0)
cycle_data["power_data"] = stored
cycle_data["sampling_interval"] = round(sampling_interval, 1)
# Helper to get arrays for signature (use stored data for consistency)
ts_arr = np.array([t for t, _ in stored])
p_arr = np.array([p for _, p in stored])
# Compute and store signature
if len(ts_arr) > 1 and len(ts_arr) == len(p_arr):
sig = compute_signature(ts_arr, p_arr)
cycle_data["signature"] = dataclasses.asdict(sig)
# Compute and store energy (Wh) if not already set (e.g. by manager).
# Shared trapezoidal integrator + data-driven outage gap (single source
# with manager._on_cycle_end).
if "energy_wh" not in cycle_data and len(ts_arr) > 1:
sort_idx = np.argsort(ts_arr)
ts_s = ts_arr[sort_idx]
p_s = p_arr[sort_idx]
cycle_data["energy_wh"] = round(
integrate_wh(ts_s, p_s, max_gap_s=energy_gap_threshold_s(ts_s)), 3
)
self._logger.debug(
"add_cycle: stored %s samples at %.1fs intervals",
len(stored),
sampling_interval,
)
# 4. Handle Debug Data (Strip if not enabled)
if hasattr(self, "_save_debug_traces") and not self._save_debug_traces:
if "debug_data" in cycle_data:
del cycle_data["debug_data"]
dest.append(cycle_data)
# Apply retention after adding
async def async_enforce_retention(self) -> None:
"""Apply retention policy asynchronously."""
affected = self._enforce_retention_data()
for p in affected:
try:
# Use async rebuild task
self.hass.async_create_task(self.async_rebuild_envelope(p))
except Exception as e: # pylint: disable=broad-exception-caught
self._logger.warning("Failed to schedule envelope rebuild for %s: %s", p, e)
def _enforce_retention_data(self) -> set[str]:
"""Internal retention logic (data operations only).
Returns set of affected profile names."""
raw_cycles = self._data.get("past_cycles", [])
cycles: list[CycleDict] = (
cast(list[CycleDict], raw_cycles) if isinstance(raw_cycles, list) else []
)
if not cycles:
return set()
def _start_time(cycle: CycleDict) -> str:
return str(cycle.get("start_time", ""))
affected_profiles: set[str] = set()
# 1) Cap total cycles
if len(cycles) > self._max_past_cycles:
# Sort by start_time and drop oldest beyond cap
try:
cycles.sort(key=_start_time)
except Exception: # pylint: disable=broad-exception-caught
pass
drop_count = len(cycles) - self._max_past_cycles
to_drop = cycles[:drop_count]
# Maintain profile sample references when dropping
sample_refs = {
name: p.get("sample_cycle_id")
for name, p in self._data.get("profiles", {}).items()
}
for cy in to_drop:
# Track affected profile
p_name = cy.get("profile_name")
if p_name:
affected_profiles.add(p_name)
cy_id = cy.get("id")
# If a profile sample points here, try to move to most recent cycle of that profile
for name, ref_id in list(sample_refs.items()):
if ref_id == cy_id:
# find newest cycle for that profile
newest = next(
(
c
for c in reversed(cycles)
if c.get("profile_name") == name and c not in to_drop
),
None,
)
if newest:
self._data["profiles"][name]["sample_cycle_id"] = (
newest.get("id")
)
else:
# No replacement available
self._data["profiles"][name].pop("sample_cycle_id", None)
# Actually drop
del cycles[:drop_count]
# 2) Strip older full traces per profile
by_profile: dict[str | None, list[CycleDict]] = {}
for cy in cycles:
key_any = cy.get("profile_name") # None for unlabeled
key: str | None = key_any if isinstance(key_any, str) and key_any else None
by_profile.setdefault(key, []).append(cy)
# Collect cycle IDs that have pending feedback - never strip their power_data
pending_feedback_ids: set[str] = set(self._data.get("pending_feedback", {}).keys())
for key, group in by_profile.items():
# newest first based on start_time
try:
group.sort(key=_start_time)
except Exception: # pylint: disable=broad-exception-caught
pass
# determine cap
cap = (
self._max_full_traces_unlabeled
if key
in (
None,
"",
)
else self._max_full_traces_per_profile
)
# count existing full traces
full_indices = [i for i, c in enumerate(group) if c.get("power_data")]
if len(full_indices) > cap:
# preserve last 'cap' full traces (newest at end after sort), strip older ones
keep_set = set(full_indices[-cap:])
# Get sample cycle ID for this profile
sample_id: str | None = None
if key and key in self._data.get("profiles", {}):
sample_id = self._data["profiles"][key].get("sample_cycle_id")
for i, c in enumerate(group):
if i in keep_set:
continue
# EXEMPTION: Never strip power data from the profile's sample cycle!
if sample_id and c.get("id") == sample_id:
continue
# EXEMPTION: Never strip power data from cycles awaiting feedback review
if c.get("id") in pending_feedback_ids:
continue
# EXEMPTION: Never strip power data from user-pinned "golden" cycles.
# The matcher uses a golden trace as the sharp single-cycle template
# (has_golden in the snapshot builder), and golden_profiles membership
# is gated on the trace still being present — trimming it would flip
# has_golden false and silently drop the profile back to the smeared
# envelope average.
rev = c.get("ml_review")
if isinstance(rev, dict) and rev.get("golden"):
continue
if c.get("power_data"):
c.pop("power_data", None)
c.pop("sampling_interval", None)
if key:
affected_profiles.add(key)
return affected_profiles
def cleanup_orphaned_profiles(self) -> int:
"""Remove profiles that reference non-existent cycles.
Returns number of profiles removed."""
# Imported reference cycles are valid sample targets too (an import-only
# profile points its sample there), so include them or such profiles would
# be wrongly deleted as orphans.
cycle_ids = {c["id"] for c in self._data.get("past_cycles", [])}
cycle_ids |= {
c["id"] for c in self._data.get("reference_cycles", []) if c.get("id")
}
orphaned: list[str] = []
for name, profile in self._data["profiles"].items():
ref = profile.get("sample_cycle_id")
# Only delete if it references a non-existent cycle ID (Broken Link)
# Creating a profile without a sample (None) is allowed (Pending State)
if ref and ref not in cycle_ids:
orphaned.append(name)
for name in orphaned:
del self._data["profiles"][name]
self._logger.info(
"Cleaned up orphaned profile '%s' (cycle no longer exists)", name
)
return len(orphaned)
async def async_run_maintenance(self) -> dict[str, int]:
"""Run full maintenance: cleanup orphans, merge fragments, trim old cycles.
Also rebuilds envelopes. Returns stats dict with counts of actions taken.
"""
stats = {
"orphaned_profiles": 0,
"merged_cycles": 0,
"split_cycles": 0,
"rebuilt_envelopes": 0,
}
# 1. Clean up orphaned profiles
stats["orphaned_profiles"] = self.cleanup_orphaned_profiles()
# 2. Auto-Label missed cycles (retroactive matching)
# Use overwrite=False to respect existing manual/confident labels
label_stats = await self.auto_label_cycles(confidence_threshold=0.75, overwrite=False)
stats["labeled_cycles"] = label_stats.get("labeled", 0)
# 3. Smart Process History (Merge/Split/Rebuild)
proc_stats = await self.async_smart_process_history()
stats["merged_cycles"] = proc_stats.get("merged", 0)
stats["split_cycles"] = proc_stats.get("split", 0)
# 4. Rebuild every profile's envelope so bands stay fresh, and report the
# real count of successful rebuilds (not an approximation).
rebuilt = 0
for profile_name in list(self._data.get("profiles", {}).keys()):
try:
# Count only real rebuilds; a no-op/failed rebuild returns False.
if await self.async_rebuild_envelope(profile_name):
rebuilt += 1
except Exception: # pylint: disable=broad-exception-caught
self._logger.debug(
"Envelope rebuild failed for %s during maintenance", profile_name, exc_info=True
)
stats["rebuilt_envelopes"] = rebuilt
# 4. Save if any changes made (smart process saves internally if needed, but explicit save safe)
if any(stats.values()):
await self.async_save()
self._logger.info("Maintenance completed: %s", stats)
return stats
def _reprocess_all_data_sync(self) -> int:
"""Synchronous implementation of reprocessing logic (run in executor)."""
cycles_raw = self._data.get("past_cycles", [])
cycles = cast(list[CycleDict], cycles_raw) if isinstance(cycles_raw, list) else []
if not cycles:
return 0
processed_count = 0
# 1. Update Signatures & Optimize Data
for cycle in cycles:
# Data Optimization: Trim leading/trailing zeros (0W)
# Only apply to compressed data to avoid breaking legacy format
p_data = cycle.get("power_data")
if (
p_data
and isinstance(p_data, list)
and p_data
and isinstance(p_data[0], (list, tuple))
):
first_point = cast(list[Any] | tuple[Any, ...], p_data[0])
# Only trim offset-format data (numeric offsets). Legacy ISO-format
# cycles skip trimming but still reach the signature block below.
if len(first_point) == 2 and isinstance(first_point[0], (int, float)):
p_data_list = cast(list[list[float]], p_data)
# Apply trim helper
original_len = len(p_data_list)
# Logic: For completed cycles, only trim leading zeros.
# For others, trim both ends.
if cycle.get("status") in ("completed", "force_stopped"):
# Only trim leading
start_idx = 0
for i, point in enumerate(p_data_list):
if point[1] > 1.0: # Match threshold below
start_idx = i
break
trimmed: list[list[float]] = p_data_list[start_idx:]
else:
trimmed = trim_zero_power_data(p_data_list, threshold=1.0) # Conservative 1W threshold
if trimmed and len(trimmed) < original_len:
# Data was trimmed - check for start time shift
first_offset = trimmed[0][0]
if first_offset > 0:
# Leading zeros removed - Must shift start_time forward
try:
start_dt = datetime.fromisoformat(cycle["start_time"])
new_start = start_dt + timedelta(seconds=first_offset)
cycle["start_time"] = new_start.isoformat()
# Re-normalize offsets to 0
shifted_data: list[list[float]] = []
for row in trimmed:
# row is [offset, power]
shifted_data.append([round(row[0] - first_offset, 1), row[1]])
cycle["power_data"] = shifted_data
processed_count += 1
except (ValueError, TypeError) as e:
self._logger.warning("Failed to shift start_time for trimmed cycle: %s", e)
else:
# Only trailing trimmed or no shift needed
cycle["power_data"] = trimmed
processed_count += 1
# Update duration to match new data length
# If we only trimmed the head, the new duration is old_duration - first_offset
# This preserves trailing silence.
if cycle.get("power_data"):
old_dur = float(cycle.get("duration", 0.0) or 0.0)
# If we shifted (first_offset > 0), new duration is old_dur - first_offset
# Otherwise if we only trimmed tail, we might want to snap,
# but for completed cycles we don't trim tail in this loop.
if first_offset > 0:
cycle["duration"] = max(0.0, old_dur - first_offset)
else:
# Only trailing was trimmed (not expected for completed cycles here)
# or no trim happened.
# If trailing was trimmed, we SHOULD snap.
if len(trimmed) < original_len:
cycle["duration"] = cycle["power_data"][-1][0]
# Self-heal duration/end_time drift. A freshly-finalized cycle
# always has last_offset == duration (the finalizer appends a
# terminal point at end_time, and add_cycle keeps the tail for
# completed cycles), so this is a no-op for healthy records. It
# repairs older / manually-edited cycles whose trace was trimmed
# without updating duration + end_time - e.g. a legacy record
# whose drying tail was dropped while duration kept the pre-trim
# value (duration says 8820s but the trace and end_time end at
# 6845s). Snap both to the trace so the three always agree.
pd_now = cycle.get("power_data")
if (
isinstance(pd_now, list)
and pd_now
and isinstance(pd_now[-1], (list, tuple))
and len(pd_now[-1]) == 2
):
trace_end = float(pd_now[-1][0])
stored_dur = float(cycle.get("duration", 0.0) or 0.0)
si = float(cycle.get("sampling_interval", 0.0) or 0.0)
# Tolerance above rounding + one sampling gap; the drifts we
# repair are large (minutes), so this never churns healthy data.
tol = max(5.0, 2.0 * si)
if trace_end > 0 and abs(stored_dur - trace_end) > tol:
cycle["duration"] = round(trace_end, 1)
start_ts = _value_to_timestamp(cycle.get("start_time"))
if start_ts is not None:
cycle["end_time"] = dt_util.utc_from_timestamp(
start_ts + trace_end
).isoformat()
processed_count += 1
self._logger.info(
"Reprocess self-heal: cycle %s duration %.0fs -> %.0fs "
"(snapped to trace; end_time realigned)",
cycle.get("id"),
stored_dur,
trace_end,
)
if cycle.get("power_data"):
try:
tuples = decompress_power_data(cycle)
if tuples and len(tuples) > 10:
ts_arr: list[float] = []
p_arr: list[float] = []
for offset_sec, p in tuples:
ts_arr.append(float(offset_sec))
p_arr.append(float(p))
sig = compute_signature(np.array(ts_arr, dtype=float), np.array(p_arr, dtype=float))
cycle["signature"] = dataclasses.asdict(sig)
processed_count += 1
except Exception as e: # pylint: disable=broad-exception-caught
self._logger.warning("Failed to reprocess signature: %s", e)
# 2. Rebuild Envelopes
return processed_count
async def async_reprocess_all_data(self) -> int:
"""Reprocess all historical data to update signatures and rebuild envelopes.
This is a non-destructive operation for raw cycle data. It:
1. Recalculates signatures for ALL past cycles using current logic.
2. Rebuilds all profile envelopes from scratch.
3. Updates global stats.
Returns total number of cycles processed.
"""
self._logger.info("Starting reprocessing (offloaded)...")
# Offload heavy synchronous work
processed_count = await self.hass.async_add_executor_job(
self._reprocess_all_data_sync
)
# 2. Rebuild Envelopes (Using new async infrastructure)
await self.async_rebuild_all_envelopes()
await self.async_save()
return processed_count
async def get_storage_stats(self) -> dict[str, Any]:
"""Get storage usage stats.
The on-disk size lookup uses blocking os.path calls, so it is offloaded
to the executor. We deliberately never json.dumps(self._data) as a
fallback: with power_data included the dataset can be many megabytes and
serialising it on the event loop previously stalled the loop long enough
that the diagnostics view appeared to hang on a spinner.
"""
cycles = self._data.get("past_cycles", [])
profiles = self._data.get("profiles", {})
debug_traces_count = sum(1 for c in cycles if c.get("debug_data"))
file_size_kb = 0.0
path = getattr(self._store, "path", None)
if path:
file_size_kb = await self.hass.async_add_executor_job(
_safe_file_size_kb, path
)
return {
"file_size_kb": round(file_size_kb, 1),
"total_cycles": len(cycles),
"total_profiles": len(profiles),
"debug_traces_count": debug_traces_count,
}
async def async_clear_debug_data(self) -> int:
"""Clear debug data from all cycles."""
cycles = self._data.get("past_cycles", [])
count = 0
for cycle in cycles:
if "debug_data" in cycle:
del cycle["debug_data"]
count += 1
if count > 0:
await self.async_save()
self._logger.info("Cleared debug data from %s cycles", count)
return count
def _rebuild_envelope_sync(
self, labeled_cycles: list[CycleDict]
) -> tuple[Any, list[float]] | None:
"""Sync worker to parse data and build envelope (run in executor).
Degenerate cycles (a near-flat trace whose peak is a tiny fraction of the
profile's typical peak - e.g. a run where the power sensor wasn't
reporting) are excluded so they cannot pollute the envelope average that
the live matcher scores against, or drag ``avg_duration`` around.
User-pinned golden cycles are always kept.
"""
# First pass: decompress everything and record each cycle's peak so we
# can judge degeneracy relative to the profile (works for both a 2000W
# dishwasher and a low-power pump).
parsed: list[tuple[list[float], list[float], float, bool, float]] = []
for cycle in labeled_cycles:
pairs = decompress_power_data(cycle)
if len(pairs) < 3:
continue
offsets = [p[0] for p in pairs]
values = [p[1] for p in pairs]
stored_dur = float(cycle.get("duration", 0.0) or 0.0)
authoritative_dur = float(max(offsets[-1], stored_dur))
man_dur = cycle.get("manual_duration")
final_dur = float(man_dur) if man_dur else authoritative_dur
review = cycle.get("ml_review")
is_golden = bool(review.get("golden")) if isinstance(review, dict) else False
peak = max(values) if values else 0.0
parsed.append((offsets, values, final_dur, is_golden, peak))
if not parsed:
return None
# Degeneracy floor: below max(_DEGENERATE_POWER_FLOOR, 10% of the median
# peak) a cycle is treated as a mis-capture and dropped (unless golden).
peaks = sorted(p[4] for p in parsed)
median_peak = peaks[len(peaks) // 2] if peaks else 0.0
degen_floor = max(_DEGENERATE_POWER_FLOOR, 0.10 * median_peak)
raw_cycles_data: list[tuple[list[float], list[float], float]] = []
durations: list[float] = []
golden_mask: list[bool] = []
dropped = 0
for offsets, values, final_dur, is_golden, peak in parsed:
if peak < degen_floor and not is_golden:
dropped += 1
continue
raw_cycles_data.append((offsets, values, final_dur))
durations.append(final_dur)
golden_mask.append(is_golden)
# Never drop everything: if the filter removed all cycles (e.g. a truly
# low-power profile misjudged), fall back to using them all.
if not raw_cycles_data:
for offsets, values, final_dur, is_golden, _peak in parsed:
raw_cycles_data.append((offsets, values, final_dur))
durations.append(final_dur)
golden_mask.append(is_golden)
elif dropped:
self._logger.debug(
"Envelope rebuild: excluded %d degenerate cycle(s) below %.0fW "
"(median peak %.0fW)", dropped, degen_floor, median_peak,
)
if not raw_cycles_data:
return None
# Run Heavy Computation. When the profile has user-verified "golden"
# cycles, they define the reference shape (see compute_envelope_worker).
result = analysis.compute_envelope_worker(
cast(Any, raw_cycles_data),
self.dtw_bandwidth,
reference_mask=golden_mask if any(golden_mask) else None,
)
if not result:
return None
return result, durations
def _cycle_peak(self, cycle: CycleDict) -> float:
"""Peak power of a cycle's trace (0.0 if it has none)."""
pairs = decompress_power_data(cycle)
return max((p[1] for p in pairs), default=0.0) if pairs else 0.0
def _select_reference_cycle_id(
self, profile_name: str, target_duration: float | None = None
) -> str | None:
"""Choose the best reference cycle for a profile's matching template.
Preference order: user-pinned golden cycles first, then non-degenerate
cycles (peak not a tiny fraction of the profile's median peak - this
rejects mis-captured near-flat traces), and among those the one closest
to the profile's representative duration (so a truncated half-cycle is
not chosen). Returns a cycle id, or None if there are no usable cycles.
"""
cands = [
c for c in list(self._data.get("past_cycles", [])) + list(self._data.get("reference_cycles", []))
if c.get("profile_name") == profile_name
and c.get("status") in ("completed", "force_stopped")
and isinstance(c.get("power_data"), list) and len(c["power_data"]) >= 3
]
if not cands:
return None
peaks = {c["id"]: self._cycle_peak(c) for c in cands}
med_peak = sorted(peaks.values())[len(peaks) // 2] if peaks else 0.0
degen_floor = max(_DEGENERATE_POWER_FLOOR, 0.10 * med_peak)
golden = [
c for c in cands
if isinstance(c.get("ml_review"), dict) and c["ml_review"].get("golden")
]
if golden:
pool = golden
else:
pool = [c for c in cands if peaks.get(c["id"], 0.0) >= degen_floor] or cands
if target_duration and target_duration > 0:
best = min(pool, key=lambda c: abs(float(c.get("duration") or 0.0) - target_duration))
else:
# No duration hint: prefer the longest (avoids truncated half-cycles).
best = max(pool, key=lambda c: float(c.get("duration") or 0.0))
return best.get("id")
async def async_rebuild_all_envelopes(self) -> int:
"""Rebuild envelopes for all profiles. Returns count of envelopes rebuilt."""
count = 0
for profile_name in list(self._data["profiles"].keys()):
if await self.async_rebuild_envelope(profile_name):
count += 1
return count
def repair_corrupted_power_data(self) -> int:
"""Fix cycles whose power_data offsets were corrupted by the double-subtract bug.
The bug caused ``offset = small_float - unix_timestamp`` to be stored instead of
just ``small_float``. Corrupted cycles have a first-offset < -1e8 (a value that
can never occur for a real appliance cycle offset). Recovery: add ``start_ts``
back to every offset in the affected cycle.
Returns the number of cycles repaired.
"""
repaired = 0
for cycle in self._data.get("past_cycles", []):
power_data = cycle.get("power_data")
if not isinstance(power_data, list) or not power_data:
continue
first = power_data[0]
if not isinstance(first, (list, tuple)) or len(first) < 2:
continue
first_offset = first[0]
if not isinstance(first_offset, (int, float)) or first_offset > -1e8:
continue # Not corrupted
start_ts = _value_to_timestamp(cycle.get("start_time"))
if start_ts is None:
continue
repaired_rows: list[list[float]] = []
for pt in power_data:
if not isinstance(pt, (list, tuple)) or len(pt) < 2:
continue
try:
repaired_rows.append([round(float(pt[0]) + start_ts, 1), round(float(pt[1]), 1)])
except (TypeError, ValueError):
continue
if not repaired_rows:
continue # all rows malformed - leave original trace untouched
cycle["power_data"] = repaired_rows
repaired += 1
repaired_data = cycle["power_data"]
if len(repaired_data) > 1:
r_offsets = [pt[0] for pt in repaired_data]
r_intervals = np.diff(r_offsets)
r_pos = r_intervals[r_intervals > 0]
r_si = float(np.median(r_pos)) if len(r_pos) > 0 else 1.0
cycle["sampling_interval"] = round(r_si, 1)
# duration = last sample offset from cycle start (not span between
# first and last sample, which would be wrong when leading zeros
# were trimmed before storage)
r_duration = round(r_offsets[-1], 1)
cycle["duration"] = r_duration
cycle["end_time"] = dt_util.utc_from_timestamp(
start_ts + r_duration
).isoformat()
r_ts = np.array(r_offsets, dtype=float)
r_p = np.array([pt[1] for pt in repaired_data], dtype=float)
r_sig = compute_signature(r_ts, r_p)
cycle["signature"] = dataclasses.asdict(r_sig)
elif len(repaired_data) == 1:
cycle["sampling_interval"] = 1.0
cycle["duration"] = 0.0
cycle["end_time"] = dt_util.utc_from_timestamp(start_ts).isoformat()
cycle["signature"] = None
if repaired:
self._logger.warning(
"Repaired corrupted power_data offsets in %d cycle(s)", repaired
)
return repaired
async def async_rebuild_envelope(self, profile_name: str) -> bool:
"""
Build/rebuild statistical envelope for a profile asynchronously.
Offloads heavy DTW/normalization to executor.
"""
# A rebuild changes this profile's curve, which feeds group cohesion, so
# invalidate the cohesion cache (not only on group mutations) to avoid stale
# cohesion approving/rejecting a collapse against outdated shapes.
self._cohesion_cache_generation += 1
# 1. Gather Data (Main Thread)
def _eligible(seq: list[CycleDict]) -> list[CycleDict]:
return [
c
for c in seq
if c.get("profile_name") == profile_name
and c.get("status") in ("completed", "force_stopped")
and c.get("duration", 0) > 60
]
# Real cycles drive usage stats (energy/count). Imported reference cycles
# additionally shape the curves + matching duration, but never usage stats.
real_cycles = _eligible(self._data["past_cycles"])
ref_cycles = _eligible(self._data.get("reference_cycles", []))
shape_cycles = real_cycles + ref_cycles
if not shape_cycles:
if profile_name in self._data.get("envelopes", {}):
del self._data["envelopes"][profile_name]
return False
# Kept for the fallback path + duration-stat code below (behaviour is
# byte-identical to before when there are no reference cycles).
labeled_cycles = shape_cycles
# 2. Run Heavy Computation in Executor (Parsing + DTW)
result_pkg = await self.hass.async_add_executor_job(
self._rebuild_envelope_sync,
shape_cycles
)
if not result_pkg:
# Envelope shape couldn't be built (no power data / too few points).
# Still update profile min/max/avg from raw cycle durations so that
# a duration correction via feedback is immediately reflected in stats.
if labeled_cycles and profile_name in self._data.get("profiles", {}):
raw_durs = [
float(c.get("manual_duration") or c.get("duration", 0))
for c in labeled_cycles
]
raw_durs = [d for d in raw_durs if d > 60]
if raw_durs:
raw_arr_fallback = np.array(raw_durs, dtype=float)
self._data["profiles"][profile_name]["min_duration"] = float(np.min(raw_arr_fallback))
self._data["profiles"][profile_name]["max_duration"] = float(np.max(raw_arr_fallback))
self._data["profiles"][profile_name]["avg_duration"] = float(np.mean(raw_arr_fallback))
if profile_name in self._data.get("envelopes", {}):
del self._data["envelopes"][profile_name]
return False
result, durations = result_pkg
# Update profile stats in storage (Fast metadata update)
if durations and profile_name in self._data.get("profiles", {}):
stats_durations = filter_duration_outliers(durations)
raw_arr = np.array(durations, dtype=float)
# min/max reflect the actual observed range (including outliers)
# avg uses the outlier-filtered set for a robust representative value
min_duration = float(np.min(raw_arr))
max_duration = float(np.max(raw_arr))
avg_duration = float(np.mean(stats_durations))
self._data["profiles"][profile_name]["min_duration"] = min_duration
self._data["profiles"][profile_name]["max_duration"] = max_duration
self._data["profiles"][profile_name]["avg_duration"] = avg_duration
# Re-point the matching reference to a golden / non-degenerate cycle
# near the representative duration, so a mis-captured (flat) or
# truncated cycle can never remain the profile's template.
ref_id = self._select_reference_cycle_id(profile_name, avg_duration)
if ref_id and self._data["profiles"][profile_name].get("sample_cycle_id") != ref_id:
old = self._data["profiles"][profile_name].get("sample_cycle_id")
self._data["profiles"][profile_name]["sample_cycle_id"] = ref_id
self._logger.debug(
"Re-selected reference cycle for %s: %s -> %s",
profile_name, old, ref_id,
)
if not result:
if profile_name in self._data.get("envelopes", {}):
del self._data["envelopes"][profile_name]
return False
time_grid, min_curve, max_curve, avg_curve, std_curve, target_duration = result
# 3. Update Storage
# Convert to list of points [[x, y], ...]
def to_points(y_vals: list[float]) -> list[list[float]]:
return [[round(t, 1), round(y, 1)] for t, y in zip(time_grid, y_vals)]
# Calculate scalar stats
duration_std_dev = float(np.std(durations)) if durations else 0.0
# Usage stats (avg_energy, cycle_count) come from REAL cycles only, so imported
# reference cycles can shape the curve without ever inflating energy/count.
# When there are no reference cycles this is byte-identical to the prior behaviour
# (avg_energy from the avg curve, cycle_count = len(durations)).
if ref_cycles:
real_energies = [
float(c["energy_wh"])
for c in real_cycles
if isinstance(c.get("energy_wh"), (int, float))
]
avg_energy = (sum(real_energies) / len(real_energies) / 1000.0) if real_energies else 0.0
cycle_count = len(real_cycles)
else:
# Average-curve energy (kWh) via the shared trapezoidal integrator.
# avg_curve is in Watts, time_grid in seconds; integrate_wh returns Wh.
avg_energy = integrate_wh(time_grid, avg_curve) / 1000.0
cycle_count = len(durations)
envelope_data: dict[str, Any] = {
"time_grid": time_grid, # Time grid used by manager for phase estimation
"target_duration": target_duration, # Target duration for phase estimation
"min": to_points(min_curve),
"max": to_points(max_curve),
"avg": to_points(avg_curve),
"std": to_points(std_curve),
"cycle_count": cycle_count,
"avg_energy": avg_energy,
"duration_std_dev": duration_std_dev,
"updated": dt_util.now().isoformat(),
}
# Derived cache: per-phase profile (per-role duration/energy priors) used by
# phase-segmented matching / phase-resolved ETA. Built only for device types
# phase matching is live-supported for; absent otherwise (consumers fall back
# to the whole-cycle pipeline). Pure/cheap - segmentation is O(samples).
device_type = str(
self._data.get("profiles", {}).get(profile_name, {}).get("device_type") or ""
)
# Offload the per-cycle segmentation to the executor (it can be tens of ms
# for very long traces; keep it off the event loop, like the envelope DTW).
phase_profile = await self.hass.async_add_executor_job(
self._compute_phase_profile, profile_name, shape_cycles, device_type
)
if phase_profile is not None:
envelope_data["phase_profile"] = phase_profile
if "envelopes" not in self._data:
self._data["envelopes"] = {}
self._data["envelopes"][profile_name] = envelope_data
return True
def _compute_phase_profile(
self, profile_name: str, cycles: list[CycleDict], device_type: str
) -> dict[str, Any] | None:
"""Segment each member cycle and aggregate a per-role phase profile.
Returns a JSON-safe dict for ``envelope["phase_profile"]`` or ``None`` when
phase matching is not live-supported for this device type or no cycle could
be segmented. Never raises (phase support must never break envelope rebuild).
"""
try:
if not phase_matching_live_supported(device_type):
return None
model = phase_model_for(device_type)
if model is None:
return None
segmented: list = []
for cycle in cycles:
offsets = power_data_to_offsets(cycle.get("power_data") or [])
if len(offsets) < 4:
continue
t = [float(o) for o, _ in offsets]
w = [float(p) for _, p in offsets]
segs = segment_cycle(t, w, model)
if segs:
segmented.append(segs)
if not segmented:
return None
profile = build_phase_profile(profile_name, segmented)
return phase_profile_to_dict(profile) if profile is not None else None
except Exception: # noqa: BLE001 - phase caching must never break rebuild
self._logger.debug("phase-profile build failed for %s", profile_name, exc_info=True)
return None
def _group_scope(self, program: str) -> set[str] | None:
"""Phase-narrowing scope for the matched ``program``:
* If ``program`` is in a group with >= 2 members, return that group's
members - narrow WITHIN the family (design §9). This is both coherent
(same program family as the displayed program) and accurate (picks the
right temperature/spin variant among siblings).
* Otherwise return ``None`` = no scope filter (consider ALL of the
device's phase profiles). The Phase-0 gate showed that constraining an
UNGROUPED cycle to only the whole-cycle-matched program regresses the
ETA whenever that match is wrong (common on mislabeled data): the best
ETA comes from letting the phase matcher pick the best-fitting profile,
bounded by the ambiguity gate + cold-start floor. Grouping variants is
the recommended workflow and restores full coherence.
"""
try:
for grp in self.get_profile_groups().values():
members = grp.get("members") if isinstance(grp, dict) else None
if isinstance(members, list) and program in members:
sib = {m for m in members if isinstance(m, str)}
if len(sib) >= 2:
return sib
except Exception: # noqa: BLE001
self._logger.debug("_group_scope failed for %r", program, exc_info=True)
return None
def _candidate_phase_profiles(self, scope: set[str] | None = None) -> list:
"""Cached per-profile PhaseProfiles (from envelope['phase_profile']).
Restricted to ``scope`` (profile names) when given, and always filtered to
profiles with >= ``PHASE_PROFILE_MIN_CYCLES`` member cycles so a noisy
single-cycle prior can't drive the ETA (cold-start floor).
"""
out = []
for name, env in (self._data.get("envelopes") or {}).items():
if scope is not None and name not in scope:
continue
if isinstance(env, dict):
pp = phase_profile_from_dict(env.get("phase_profile"))
if pp is not None and pp.n_cycles >= PHASE_PROFILE_MIN_CYCLES:
out.append(pp)
return out
def phase_remaining(
self,
power_data: list,
device_type: str,
program: str | None = None,
) -> dict[str, Any] | None:
"""Phase-resolved remaining-time for a running cycle. Never raises.
Segments the observed-so-far trace and matches it against the matched
``program``'s phase profile (and its group siblings, design §9), returning
the winning member's per-role budget remaining. Returns ``None`` (caller
keeps the current estimate) when: phase matching is not live-supported for
the device type; ``program`` is unknown / has no cached phase profile
(or too few cycles - cold-start floor); segmentation is degenerate; or the
top two candidates are within ``MATCH_AMBIGUITY_MARGIN`` (ambiguous - do
not commit a variant, design §7).
This is the *phase* half of the blended ETA; the blend with the current
estimator lives in ``progress.compute_progress`` (single source of truth).
Pure and cheap (segmentation + per-role agreement, no DTW) - safe to call
inline from the async matching path.
"""
try:
if not phase_matching_live_supported(device_type):
return None
model = phase_model_for(device_type)
if model is None or not program:
return None
candidates = self._candidate_phase_profiles(self._group_scope(program))
if not candidates:
return None
offsets = power_data_to_offsets(power_data or [])
if len(offsets) < 4:
return None
t = [float(o) for o, _ in offsets]
w = [float(p) for _, p in offsets]
segs = segment_cycle(t, w, model, partial=True)
if not segs:
return None
ranked = match_phase_profiles(segs, candidates, {})
if not ranked:
return None
# Ambiguity gate: a near-tie among group members is not a confident
# variant call - fall back rather than swing the ETA between budgets.
if (len(ranked) >= 2
and (ranked[0].score - ranked[1].score) < MATCH_AMBIGUITY_MARGIN):
return None
best = next((c for c in candidates if c.name == ranked[0].name), None)
remaining = phase_eta(segs, best) if best is not None else None
if remaining is None:
return None
return {
"remaining_s": float(remaining),
"matched": ranked[0].name,
"score": float(ranked[0].score),
}
except Exception: # noqa: BLE001 - phase ETA must never break the estimate
self._logger.debug("phase_remaining failed", exc_info=True)
return None
def get_envelope(self, profile_name: str) -> JSONDict | None:
"""Get cached envelope for a profile, or None if not available."""
envelopes = self._data.get("envelopes", {})
if isinstance(envelopes, dict):
envelopes_map = cast(dict[str, Any], envelopes)
env = envelopes_map.get(profile_name)
return cast(JSONDict, env) if isinstance(env, dict) else None
return None
def reference_curve(
self, profile_name: str, n: int = REFERENCE_PROFILE_CURVE_POINTS
) -> JSONDict | None:
"""Compact, read-only reference power curve for a matched profile.
Downsamples the profile envelope's average power-over-time shape to at
most ``n`` points so it can be exposed as an entity attribute for
consumers (e.g. home energy managers) that want the *forward-looking
load shape* rather than a scalar ETA - to anticipate, say, a heating
spike later in the cycle. Shape::
{
"points": [[offset_s, watts], ...], # <= n samples
"duration_s": float, # profile target duration
"cycle_count": int, # cycles behind the average
}
Offsets are absolute seconds from cycle start (0 .. ``duration_s``); a
consumer derives the *remaining* curve by slicing from the live progress
position (already exposed as the progress sensor). The curve is static
per profile - it only changes when the profile is re-learned - so it can
be surfaced as an attribute without recorder churn.
Pure statistics (no ML); never raises - returns ``None`` when the
envelope is missing or too short to be meaningful.
"""
try:
env = self.get_envelope(profile_name)
if not isinstance(env, dict):
return None
avg = env.get("avg")
if (
not isinstance(avg, list)
or len(avg) < 2
or not isinstance(avg[0], (list, tuple))
or len(avg[0]) < 2
):
return None
ts = np.asarray([float(p[0]) for p in avg], dtype=float)
ws = np.asarray([float(p[1]) for p in avg], dtype=float)
if ts.size < 2 or not (np.all(np.isfinite(ts)) and np.all(np.isfinite(ws))):
return None
if ts[-1] <= ts[0]:
return None
duration = float(env.get("target_duration") or 0.0)
if duration <= 0:
duration = float(ts[-1])
count = max(2, min(int(n), ts.size))
grid = np.linspace(float(ts[0]), float(ts[-1]), count)
vals = np.interp(grid, ts, ws)
points = [
[int(round(float(g))), round(float(v), 1)]
for g, v in zip(grid, vals)
]
return {
"points": points,
"duration_s": round(duration, 1),
"cycle_count": int(env.get("cycle_count") or 0),
}
except Exception: # pragma: no cover - defensive; never break the sensor
return None
def get_profile_power_profile(
self, profile_name: str, interval_s: float = 900.0
) -> list[float]:
"""Average power (W) per fixed interval across a profile's learned shape.
Resamples the profile envelope's average power-over-time curve into
consecutive ``interval_s`` buckets (default 15 min) and returns the mean
watts in each, e.g. ``[2200, 2200, 800, 800, 1500, 500, 400, 200]`` - the
flat per-slot array external planners such as tibber_prices'
``power_profile`` consume to pick the cheapest window to run the appliance
(issue #272). Unlike :meth:`reference_curve` (a downsampled time/watt shape
surfaced on the running-program sensor), this is a fixed-interval average
exposed per profile so it can be read for planning before a cycle starts.
The final bucket is averaged only over the part of the cycle that actually
falls inside it. Pure statistics; never raises. Returns an empty list when
the profile has no learned envelope yet.
"""
try:
if interval_s <= 0:
return []
env = self.get_envelope(profile_name)
if not isinstance(env, dict):
return []
avg = env.get("avg")
if (
not isinstance(avg, list)
or len(avg) < 2
or not isinstance(avg[0], (list, tuple))
or len(avg[0]) < 2
):
return []
ts = np.asarray([float(p[0]) for p in avg], dtype=float)
ws = np.asarray([float(p[1]) for p in avg], dtype=float)
if ts.size < 2 or not (np.all(np.isfinite(ts)) and np.all(np.isfinite(ws))):
return []
if ts[-1] <= ts[0]:
return []
total = float(env.get("target_duration") or 0.0)
if total <= 0:
total = float(ts[-1])
if total <= 0:
return []
n_buckets = int(math.ceil(total / interval_s))
out: list[float] = []
for k in range(n_buckets):
lo = k * interval_s
hi = min(lo + interval_s, total)
if hi <= lo:
break
# Time-average the curve over the half-open slot [lo, hi) on a fine
# interpolated grid, so the result is independent of the envelope's
# grid spacing. endpoint=False keeps a slot boundary from being
# double-counted into the adjacent slot.
fine = np.linspace(lo, hi, 16, endpoint=False)
out.append(round(float(np.mean(np.interp(fine, ts, ws))), 1))
return out
except Exception: # pragma: no cover - defensive; never break the sensor
return []
def compute_envelope_conformance(
self,
profile_name: str,
points: list[tuple[float, float]],
) -> dict[str, Any] | None:
"""Score how well the current power trace conforms to the profile envelope band.
Resamples ``points`` to the envelope's time grid (scaling by the ratio of
the current elapsed time to the envelope duration) and computes the fraction
of samples that land within the [lower, upper] band. Returns a dict:
``conformance`` – fraction of samples inside the envelope band (0–1)
``outside_frac`` – fraction outside the band (1 - conformance)
``samples`` – number of samples compared
``envelope_name`` – profile_name used
Returns ``None`` if the envelope or points are unavailable / too short.
Never raises.
This is a complementary signal to match confidence: confidence measures
shape correlation, while conformance measures absolute level/spread. A
high-confidence but low-conformance score may indicate the cycle was
mis-detected (merged cycles, offset start) or the appliance is running
anomalously.
"""
try:
if not points or len(points) < 4:
return None
env = self.get_envelope(profile_name)
if not env:
return None
time_grid = env.get("time_grid")
lower_raw = env.get("min")
upper_raw = env.get("max")
if not time_grid or not lower_raw or not upper_raw:
return None
tg = np.asarray(time_grid, dtype=float)
# Envelope curves are stored as [[t, y], ...]; extract the y column.
lo = np.asarray(
[p[1] if isinstance(p, (list, tuple)) else p for p in lower_raw],
dtype=float,
)
hi = np.asarray(
[p[1] if isinstance(p, (list, tuple)) else p for p in upper_raw],
dtype=float,
)
env_duration = float(tg[-1]) if len(tg) > 1 else 1.0
# Normalise observed trace to [0, env_duration] time range
t_obs = np.asarray([t for t, _ in points], dtype=float)
p_obs = np.asarray([p for _, p in points], dtype=float)
obs_duration = float(t_obs[-1]) if len(t_obs) > 1 else 1.0
t_scaled = t_obs * (env_duration / obs_duration) if obs_duration > 0 else t_obs
# Interpolate envelope bounds at scaled observed time points (clamp ends)
t_clamped = np.clip(t_scaled, tg[0], tg[-1])
lo_interp = np.interp(t_clamped, tg, lo)
hi_interp = np.interp(t_clamped, tg, hi)
inside = np.sum((p_obs >= lo_interp) & (p_obs <= hi_interp))
n = len(p_obs)
conformance = float(inside) / n if n > 0 else 0.0
return {
"conformance": round(conformance, 3),
"outside_frac": round(1.0 - conformance, 3),
"samples": n,
"envelope_name": profile_name,
}
except Exception: # noqa: BLE001
return None
def detect_cycle_artifacts(
self,
profile_name: str,
points: list[tuple[float, float]],
) -> list[dict[str, Any]]:
"""Detect transient artifacts in a cycle by comparing it to the profile band.
Walks the trace against the matched profile's envelope and flags contiguous
segments that deviate from what the program normally does:
* ``pause`` – power fell to ~0 where the profile expects activity and then
resumed (e.g. the door was opened mid-cycle to add an item, or a manual
pause); excluded at the very end (that's just the cycle finishing).
* ``dip`` – ran sustainedly below the usual power band.
* ``spike`` – ran sustainedly above the usual power band.
Returns a chronological list of ``{type, start_s, end_s, detail, severity}``
in the trace's own time offsets (so the panel can mark them on the graph),
capped to the most significant few. ``[]`` when unavailable. Pure statistics
(no ML), never raises — and the events double as candidate labels for a
future supervised anomaly model.
"""
try:
if not points or len(points) < 6:
return []
env = self.get_envelope(profile_name)
if not env:
return []
tg = np.asarray(env.get("time_grid") or [], dtype=float)
# Envelope curves are stored as [[t, y], ...]; extract the y column.
def _extract_y(raw: list) -> np.ndarray:
if not raw:
return np.array([], dtype=float)
return np.asarray(
[p[1] if isinstance(p, (list, tuple)) else p for p in raw],
dtype=float,
)
lo = _extract_y(env.get("min") or [])
hi = _extract_y(env.get("max") or [])
avg = _extract_y(env.get("avg") or [])
if tg.size < 2 or lo.size != tg.size or hi.size != tg.size:
return []
t_obs = np.asarray([t for t, _ in points], dtype=float)
p_obs = np.asarray([max(0.0, float(p)) for _, p in points], dtype=float)
obs_dur = float(t_obs[-1]) if t_obs.size > 1 else 1.0
env_dur = float(tg[-1]) if tg.size > 1 else 1.0
if obs_dur <= 0 or env_dur <= 0:
return []
# Map each observed time to the envelope grid (same scaling as
# compute_envelope_conformance) so the band is aligned to progress.
t_scaled = np.clip(t_obs * (env_dur / obs_dur), tg[0], tg[-1])
lo_i = np.interp(t_scaled, tg, lo)
hi_i = np.interp(t_scaled, tg, hi)
avg_i = np.interp(t_scaled, tg, avg) if avg.size == tg.size else (lo_i + hi_i) / 2.0
peak = max(float(np.max(hi)), 1.0)
active_thr = max(5.0, 0.05 * peak) # profile expects real power here
pause_thr = max(2.0, 0.03 * peak) # observed effectively off
margin = max(10.0, 0.12 * peak) # band slack to ignore edge noise
# Pre-check: if the cycle's linear-resampled alignment is too poor
# (e.g. duration or phase structure doesn't match the profile), the
# envelope comparison produces unreliable artifacts. Compute tight
# conformance (no margin) over the active region; bail out when more
# than 45 % of expected-active samples are outside the raw band.
active_mask = avg_i > active_thr
n_active = int(np.sum(active_mask))
if n_active > 10:
outside_tight = int(np.sum(
(p_obs[active_mask] < lo_i[active_mask]) |
(p_obs[active_mask] > hi_i[active_mask])
))
if outside_tight / n_active > 0.45:
return []
states: list[str] = []
for i in range(len(p_obs)):
expects_power = avg_i[i] > active_thr
if expects_power and p_obs[i] <= pause_thr:
states.append("pause")
elif expects_power and p_obs[i] < lo_i[i] - margin:
states.append("dip")
elif p_obs[i] > hi_i[i] + margin:
states.append("spike")
else:
states.append("ok")
min_dur = {"pause": 25.0, "dip": 45.0, "spike": 30.0}
sev_scale = {"pause": 300.0, "dip": 600.0, "spike": 300.0}
events: list[dict[str, Any]] = []
n = len(states)
i = 0
while i < n:
s = states[i]
if s == "ok":
i += 1
continue
j = i
while j + 1 < n and states[j + 1] == s:
j += 1
start_s, end_s = float(t_obs[i]), float(t_obs[j])
dur = end_s - start_s
resumes = (j + 1 < n) and float(np.max(p_obs[j + 1:])) > active_thr
if dur >= min_dur[s] and (s != "pause" or resumes):
if s == "pause":
detail = (f"Power dropped to near zero for ~{int(dur)}s then resumed — "
"likely the door was opened mid-cycle or the cycle was paused.")
detail_key = "msg.artifact_pause_detail"
elif s == "dip":
detail = f"Drew below the usual power band for ~{int(dur)}s."
detail_key = "msg.artifact_dip_detail"
else:
detail = f"Drew above the usual power band for ~{int(dur)}s."
detail_key = "msg.artifact_spike_detail"
events.append({
"type": s,
"start_s": round(start_s, 1),
"end_s": round(end_s, 1),
"detail": detail,
"detail_key": detail_key,
"detail_params": {"n": int(dur)},
"severity": round(min(1.0, dur / sev_scale[s]), 3),
})
i = j + 1
events.sort(key=lambda e: -e["severity"])
events = events[:6]
events.sort(key=lambda e: e["start_s"])
return events
except Exception: # noqa: BLE001
return []
def get_match_candidates_summary(
self, match_result: MatchResult, limit: int = 3
) -> list[dict[str, Any]]:
"""Extract top candidates from query result for UI display.
Args:
match_result: MatchResult from profile matching
limit: Number of top candidates to return
Returns:
List of dicts with keys: profile_name, confidence_pct, mae, correlation, duration_ratio
"""
candidates: list[dict[str, Any]] = []
for candidate in match_result.ranking[:limit]:
try:
confidence_pct = round(candidate.get("score", 0.0) * 100, 1)
metrics = candidate.get("metrics", {})
mae = round(metrics.get("mae", 0.0), 2)
corr = round(metrics.get("corr", 0.0), 3)
profile_duration = candidate.get("profile_duration", 0.0)
actual_duration = match_result.expected_duration
duration_ratio = (
round((actual_duration / profile_duration - 1.0) * 100, 1)
if profile_duration > 0
else 0.0
)
candidates.append({
"profile_name": candidate.get("name", "Unknown"),
"confidence_pct": confidence_pct,
"mae": mae,
"correlation": corr,
"duration_ratio": duration_ratio, # ±% from expected
})
except (TypeError, ValueError, KeyError):
continue
return candidates
def _get_cached_sample_segment(
self, sample_cycle: dict[str, Any], dt: float
) -> Segment | None:
"""Get or compute resampled segment for a sample cycle, using cache."""
cycle_id = sample_cycle.get("id")
if not cycle_id:
return None
# Round dt to avoid float cache misses
dt_key = float(round(dt, 2))
key = (cycle_id, dt_key)
if key in self._cached_sample_segments:
return self._cached_sample_segments[key]
# Miss: Compute
sample_data = sample_cycle.get("power_data")
if not sample_data:
return None
try:
if len(sample_data) > 0 and isinstance(sample_data[0], (list, tuple)):
s_ts = np.array([x[0] for x in sample_data])
s_p = np.array([x[1] for x in sample_data])
else:
return None
s_segments = resample_uniform(s_ts, s_p, dt_s=dt, gap_s=21600.0)
if not s_segments:
return None
sample_seg = max(s_segments, key=lambda s: len(s.power))
# Store
self._cached_sample_segments[key] = sample_seg
return sample_seg
except Exception as e: # pylint: disable=broad-exception-caught
self._logger.warning("Error caching sample segment %s: %s", cycle_id, e)
return None
async def async_match_profile(
self,
current_power_data: list[tuple[str, float]] | list[tuple[datetime, float]] | list[tuple[float, float]] | list[list[float]],
current_duration: float,
) -> MatchResult:
"""Run profile matching asynchronously in executor."""
# 1. Prepare data in main thread (Access ProfileStore state safely)
group_members: dict[str, list[str]] = {}
member_snaps: dict[str, dict[str, Any]] = {}
# Convert to list of floats for current power (uniform resampling)
if not current_power_data:
return MatchResult(None, 0.0, 0.0, None, [], False, 0.0)
# Pre-process current data
try:
# Normalize input format
first_elem = current_power_data[0][0]
if isinstance(first_elem, datetime):
# datetime objects: compute relative timestamps
t_start = first_elem.timestamp()
ts_arr = np.array([(x[0].timestamp() - t_start) for x in cast(list[tuple[datetime, float]], current_power_data)])
elif isinstance(first_elem, (int, float)):
# Already offset timestamps (from compressed format)
ts_arr = np.array([float(x[0]) for x in cast(list[tuple[float, float]], current_power_data)])
else:
# ISO format strings
t_start = datetime.fromisoformat(first_elem).timestamp()
ts_arr = np.array(
[
(datetime.fromisoformat(x[0]).timestamp() - t_start)
for x in cast(list[tuple[str, float]], current_power_data)
]
)
p_arr = np.array([float(x[1]) for x in current_power_data])
# Resample current
segments, used_dt = resample_adaptive(ts_arr, p_arr, min_dt=5.0, gap_s=21600.0)
if not segments:
return MatchResult(None, 0.0, 0.0, None, [], False, 0.0)
current_seg = max(segments, key=lambda s: len(s.power))
if len(current_seg.power) < 12:
return MatchResult(None, 0.0, 0.0, None, [], False, 0.0)
current_power_list = current_seg.power.tolist()
# Prepare Snapshots. Imported reference cycles are eligible as matching
# templates alongside real cycles (so an import-only profile can match).
all_cycles = list(self._data["past_cycles"]) + list(self._data.get("reference_cycles", []))
# Precompute per-profile lookups ONCE so the loop below is O(profiles),
# not O(profiles x cycles). Rescanning all_cycles with next()/any() for
# every profile made matching quadratic and stalled low-power hosts on
# auto-label (many matches x many cycles) - issue #311. Selections are
# byte-identical: cycles_by_id keeps the FIRST occurrence (== next()),
# labeled_by_profile keeps the first eligible cycle in all_cycles order,
# and golden_profiles mirrors the any(...) golden test.
cycles_by_id: dict[str, CycleDict] = {}
labeled_by_profile: dict[str, CycleDict] = {}
golden_profiles: set[str] = set()
for c in all_cycles:
cid = c.get("id")
if cid is not None and cid not in cycles_by_id:
cycles_by_id[cid] = c
pname = c.get("profile_name")
if not pname or not c.get("power_data"):
continue
if (
pname not in labeled_by_profile
and c.get("status") in ("completed", "force_stopped")
):
labeled_by_profile[pname] = c
rev = c.get("ml_review")
if isinstance(rev, dict) and rev.get("golden"):
golden_profiles.add(pname)
snapshots: list[dict[str, Any]] = []
skipped_profiles: list[str] = []
for name, profile in self._data["profiles"].items():
# Try sample_cycle_id first, fall back to any labeled cycle
sample_id = profile.get("sample_cycle_id")
sample_cycle = cycles_by_id.get(sample_id) if sample_id else None
# Fallback: find ANY completed cycle labeled with this profile
if not sample_cycle:
sample_cycle = labeled_by_profile.get(name)
# Boost user-pinned "golden" cycles: when a profile has one, use
# its sharp single-cycle trace as the matching template instead
# of the envelope average. The envelope average smears the
# wash-phase peaks (each cycle's spikes land at slightly
# different times), which hurts correlation for sharply-shaped
# programs; a trusted golden cycle preserves that shape.
has_golden = name in golden_profiles
# Prefer envelope avg curve when ≥2 labeled cycles have been
# confirmed - it gives a more representative reference signal
# than the original sample alone, so confidence improves over
# time as the user keeps confirming correct detections. Skipped
# when a golden cycle is pinned (see above).
envelope = self._data.get("envelopes", {}).get(name)
_env_avg = envelope.get("avg") if envelope else None
if (
not has_golden
and envelope
and envelope.get("cycle_count", 0) >= 2
and _env_avg
and isinstance(_env_avg[0], (list, tuple))
and len(_env_avg[0]) >= 2
):
avg_y = [float(p[1]) for p in _env_avg]
_env_ts_duration = (
float(_env_avg[-1][0]) - float(_env_avg[0][0])
if len(_env_avg) > 1 else 0.0
)
avg_duration = (
envelope.get("target_duration") or
profile.get("avg_duration") or
_env_ts_duration or
None
)
if not avg_duration:
skipped_profiles.append(
f"{name}: no valid duration (envelope has no target_duration, avg_duration, or timestamp span)"
)
continue
snapshots.append({
"name": name,
"avg_duration": float(avg_duration),
"sample_power": avg_y,
})
continue
if not sample_cycle:
skipped_profiles.append(
f"{name}: no sample cycle (sample_id={sample_id})"
)
continue
# Prepare sample segment (using cache)
sample_seg = self._get_cached_sample_segment(sample_cycle, used_dt)
if not sample_seg:
skipped_profiles.append(
f"{name}: failed to resample cycle {sample_cycle.get('id')}"
)
continue
# avg_duration preference order:
# 1. profile["avg_duration"] (rolling average, most accurate)
# 2. sample_cycle["duration"] (raw cycle field)
# 3. timestamp span of sample_seg (estimate from the resampled data)
# Profiles created before avg_duration tracking was added may have
# 0 or a missing value; falling back to the segment estimate prevents
# update_match() from always seeing expected_duration=0, which
# silences time-remaining estimates and logs a misleading warning.
_seg_ts_duration = (
float(sample_seg.timestamps[-1]) - float(sample_seg.timestamps[0])
if len(sample_seg.timestamps) > 1 else 0.0
)
avg_dur = (
profile.get("avg_duration") or
sample_cycle.get("duration") or
_seg_ts_duration
)
if not avg_dur:
skipped_profiles.append(
f"{name}: no valid duration (avg_duration, cycle duration, and timestamp span all zero/missing)"
)
continue
snapshots.append({
"name": name,
"avg_duration": float(avg_dur),
"sample_power": sample_seg.power.tolist(),
"sample_dt": used_dt
})
if skipped_profiles:
self._logger.debug(
"Profile matching skipped %d profiles: %s",
len(skipped_profiles),
"; ".join(skipped_profiles)
)
# Stage 5: collapse cohesive near-duplicate groups into one aggregate
# candidate each (loose groups stay individual). No-op without groups.
snapshots, group_members, member_snaps = self._grouped_snapshots(snapshots)
config = {
"min_duration_ratio": self._min_duration_ratio,
"max_duration_ratio": self._max_duration_ratio,
"dtw_bandwidth": self.dtw_bandwidth,
# On-device tuned scoring weights (opt-in); empty = shipped defaults.
**self._matching_overrides(),
}
except Exception as e: # pylint: disable=broad-exception-caught
self._logger.error("Preparation for async match failed: %s", e)
return MatchResult(None, 0.0, 0.0, None, [], False, 0.0)
# 2. Run Heavy Logic in Executor
candidates = await self.hass.async_add_executor_job(
analysis.compute_matches_worker,
current_power_list,
current_duration,
cast(Any, snapshots),
config
)
# 3. Process Result (Main Thread)
if not candidates:
profiles_count = len(self._data.get("profiles", {}))
snapshots_count = len(snapshots) if 'snapshots' in dir() else 0
self._logger.debug(
"No profile match candidates: profiles=%d, snapshots=%d, "
"duration=%.0fs. Possible reasons: duration ratio filter, "
"no labeled cycles, or no profiles defined.",
profiles_count,
snapshots_count,
current_duration
)
return MatchResult(None, 0.0, 0.0, None, [], False, 0.0, [], {}, is_confident_mismatch=True, mismatch_reason="all_rejected")
best = candidates[0]
# Reconstruct MatchResult. Top-level ambiguity (safeguard #1): a close
# group-vs-runner-up call is flagged so it goes to the uncertain/feedback
# path, not a confident commit.
margin, is_ambiguous = _ambiguity_from_candidates(candidates)
best_name = best["name"]
best_duration = best["profile_duration"]
# Stage 5: if a group won, pick the best-fitting member.
if best_name in group_members:
chosen, member_fit, member_dur = self._stage5_pick_member(
current_power_list, current_duration, group_members[best_name], member_snaps
)
best_name = chosen
if member_dur:
best_duration = member_dur
# Safeguard #2: the group aggregate matched but if the chosen member
# does not individually fit reasonably (vs the group score), the real
# program may be a different single profile -> treat as uncertain.
# member_fit is a Stage-2-only score; best["score"] includes DTW-blend +
# duration/energy agreement (typically 25-30% higher). Use 0.55× to avoid
# the threshold being effectively too strict for DTW-boosted groups.
if member_fit is not None and best["score"] > 0 and member_fit < 0.55 * best["score"]:
is_ambiguous = True
# Overrun guard: if the cycle has already outlasted the chosen member's
# expected duration we may be running the longer group member — downgrade
# to ambiguous so Smart Termination falls back to the power timeout.
if best_duration and current_duration > best_duration * 1.05:
is_ambiguous = True
# Relabel the winning candidate for the ranking / diagnostics.
candidates = [{**best, "name": best_name, "profile_duration": best_duration}, *candidates[1:]]
matched_phase = None
if best_name:
# Always resolve phase for the matched profile so phase sensors can
# show user-assigned phase names even when confidence is moderate.
matched_phase = self.check_phase_match(best_name, current_duration)
# Detect "prefix ambiguity": a non-winning candidate whose duration is
# significantly longer than the matched profile AND whose shape matched
# well before Stage-4 penalised its duration. When this is true the
# current trace may be a prefix of that longer program, not a complete
# short cycle. Signal cycle_detector to block Smart Termination; the
# power-based fallback timeout will decide instead.
best_dur = best_duration or 0.0
is_prefix_ambiguous = best_dur > 0 and any(
float(c.get("profile_duration") or 0) > best_dur * SMART_TERM_LANDSCAPE_RATIO
and float(c.get("shape_score", c.get("score", 0))) >= SMART_TERM_LANDSCAPE_MIN_SHAPE
for c in candidates[1:]
)
return MatchResult(
best_name,
best["score"],
best_duration,
matched_phase,
candidates[:5],
is_ambiguous,
margin,
ranking=candidates[:5], # populate ranking (consumed for training snapshots)
is_prefix_ambiguous=is_prefix_ambiguous,
)
def match_profile(
self, power_data: list[tuple[str, float]], duration: float
) -> MatchResult:
"""Synchronous wrapper for matching (for use in executor tasks)."""
# Convert to list for worker
p_list = [p[1] for p in power_data]
# Prepare snapshots safely
snapshots: list[dict[str, Any]] = []
# Accessing self._data in thread is generally safe for reads if not modifying
for name, profile in self._data["profiles"].items():
sample_id = profile.get("sample_cycle_id")
sample_cycle = next((c for c in self._data["past_cycles"] if c["id"] == sample_id), None)
if not sample_cycle:
continue
# Decompress sample data
sample_p_data = decompress_power_data(sample_cycle)
if not sample_p_data:
continue
snapshots.append({
"name": name,
"avg_duration": profile.get("avg_duration", sample_cycle.get("duration", 0)),
"sample_power": [x[1] for x in sample_p_data],
})
config = {
"min_duration_ratio": self._min_duration_ratio,
"max_duration_ratio": self._max_duration_ratio,
"dtw_bandwidth": self.dtw_bandwidth,
# On-device tuned scoring weights (opt-in); empty = shipped defaults.
**self._matching_overrides(),
}
candidates = analysis.compute_matches_worker(
p_list, duration, cast(Any, snapshots), config
)
if not candidates:
return MatchResult(None, 0.0, 0.0, None, [], False, 0.0)
best = candidates[0]
margin, is_ambiguous = _ambiguity_from_candidates(candidates)
return MatchResult(
best["name"],
best["score"],
best["profile_duration"],
None,
candidates,
is_ambiguous,
margin,
ranking=candidates,
)
async def async_verify_alignment(
self,
profile_name: str,
current_power_data: list[list[float]] | list[tuple[Any, ...]],
) -> tuple[bool, float, float]:
"""
Verify if the current power trace aligns with an expected low-power region in the envelope.
Returns: (is_confirmed_low_power, mapped_envelope_time, mapped_envelope_power)
"""
envelope = self.get_envelope(profile_name)
if not envelope or not envelope.get("avg") or not current_power_data:
return False, 0.0, 9999.0
# Extract envelope curves
# "avg" can be list of [t, p] (new) or [p, ...] (legacy)
env_avg_raw = envelope.get("avg", [])
if not env_avg_raw:
return False, 0.0, 9999.0
try:
# Handle both formats: [[t, y], ...] (new) or [y, ...] (legacy)
if isinstance(env_avg_raw[0], (list, tuple)) and len(env_avg_raw[0]) >= 2:
# New format: [[t, y], ...]
env_points = cast(list[list[Any] | tuple[Any, ...]], env_avg_raw)
env_time = [float(p[0]) for p in env_points]
env_power = [float(p[1]) for p in env_points]
else:
# Legacy format: [y, ...]
env_values = cast(list[float | int], env_avg_raw)
env_power = [float(p) for p in env_values]
# Reconstruct time grid from envelope if available, or assume 60s intervals
env_time_raw = envelope.get("time_grid")
env_time = cast(list[float], env_time_raw) if isinstance(env_time_raw, list) else None
if not env_time or len(env_time) != len(env_power):
target_dur = float(envelope.get("target_duration", 0.0) or 0.0)
if target_dur > 0:
env_time = cast(list[float], np.linspace(0, target_dur, len(env_power)).tolist())
else:
env_time = [float(i * 60) for i in range(len(env_power))]
except (TypeError, ValueError, IndexError) as e:
first_type_name = type(env_avg_raw[0]).__name__ if env_avg_raw else "None"
self._logger.error(
"Malformed envelope 'avg' data for %s. Type: %s, Length: %d, Error: %s",
profile_name, first_type_name, len(env_avg_raw), e
)
return False, 0.0, 9999.0
try:
current_power_list = [float(x[1]) for x in current_power_data]
except Exception: # pylint: disable=broad-exception-caught
return False, 0.0, 9999.0
# Offload to worker
mapped_time, mapped_power, score = await self.hass.async_add_executor_job(
analysis.verify_profile_alignment_worker,
current_power_list,
env_power,
env_time,
self.dtw_bandwidth
)
# Verify if mapped power and alignment score indicate an expected low-power region.
# Thresholds: Expected power < 15W, Alignment score > 0.4
is_confirmed = (mapped_power < 15.0) and (score > 0.4)
return is_confirmed, mapped_time, mapped_power
def check_phase_match(self, profile_name: str, duration: float) -> str | None:
"""
Check if the current duration aligns with a known phase in the profile.
Returns the phase name (e.g., 'Rinse', 'Spin') or None.
"""
profile = self._data["profiles"].get(profile_name)
if not profile:
return None
phases = profile.get("phases", [])
if not phases:
return None
phases_sorted = sorted(
phases,
key=lambda p: float(p.get("start", 0)),
)
for phase in phases_sorted:
p_start = phase.get("start", 0)
p_end = phase.get("end", 0)
if p_start <= duration <= p_end:
return str(phase.get("name", "Unknown"))
# If duration is outside explicit bounds, keep a phase label anyway so
# entities avoid falling back to generic running/starting states.
if phases_sorted:
if duration < float(phases_sorted[0].get("start", 0)):
return str(phases_sorted[0].get("name", "Unknown"))
return str(phases_sorted[-1].get("name", "Unknown"))
return None
async def create_profile(self, name: str, source_cycle_id: str) -> None:
"""Create a new profile from a past cycle."""
cycle = next(
(c for c in self._data["past_cycles"] if c["id"] == source_cycle_id), None
)
if not cycle:
raise ValueError("Cycle not found")
cycle["profile_name"] = name
self._data.setdefault("profiles", {})[name] = {
"avg_duration": cycle["duration"],
"sample_cycle_id": source_cycle_id,
}
# Save to persist the label
await self.async_save()
@property
def has_real_profiles(self) -> bool:
"""True if at least one stored profile is backed by a real cycle.
A profile counts as "real" when it has a labelled cycle in ``past_cycles``
OR an imported ``reference_cycle`` (store-adopted templates that the matcher
treats as eligible snapshots — see the snapshot builder in async_match). An
import-only install has zero past_cycles but is fully matchable, so it must
pass this gate too, otherwise matching and the setup notifications are
skipped for it entirely.
"""
profile_names = self._data.get("profiles", {}).keys()
if not profile_names:
return False
assigned = {
c.get("profile_name")
for c in self._data.get("past_cycles", [])
if c.get("profile_name")
}
if assigned.intersection(profile_names):
return True
ref_assigned = {
c.get("profile_name")
for c in self._data.get("reference_cycles", [])
if c.get("profile_name")
}
return bool(ref_assigned.intersection(profile_names))
def list_profiles(self) -> list[dict[str, Any]]:
"""List all profiles with metadata."""
profiles: list[JSONDict] = []
raw_profiles = self._data.get("profiles", {})
profiles_map = (
cast(dict[str, Any], raw_profiles) if isinstance(raw_profiles, dict) else {}
)
for name, data in profiles_map.items():
profile_meta = cast(JSONDict, data) if isinstance(data, dict) else {}
# Calculate count and last_run
p_cycles = [
c for c in self._data.get("past_cycles", [])
if c.get("profile_name") == name
]
cycle_count = len(p_cycles)
last_run = None
if p_cycles:
last_c = max(p_cycles, key=lambda x: x.get("start_time", ""))
last_run = last_c.get("start_time")
# Fetch envelope stats
envelope = self.get_envelope(name)
avg_energy = envelope.get("avg_energy") if envelope else None
duration_std_dev = envelope.get("duration_std_dev") if envelope else None
# Per-profile cost aggregates from frozen per-cycle costs. Never raises;
# both default to None when no cycle carries a cost.
avg_cost: float | None = None
total_cost: float | None = None
try:
costs = [float(c["cost"]) for c in p_cycles if c.get("cost") is not None]
if costs:
total_cost = round(sum(costs), 4)
avg_cost = round(sum(costs) / len(costs), 4)
except Exception: # noqa: BLE001
avg_cost = None
total_cost = None
# Downsampled power signature (the profile's real average power curve),
# for the profile-card mini graph. Small (<=40 pts), power values only.
sig_curve: list[float] = []
try:
avg_pts = (envelope or {}).get("avg") or []
ps = [
float(p[1]) for p in avg_pts
if isinstance(p, (list, tuple)) and len(p) >= 2
]
if not ps: # legacy flat [p, ...] envelopes
ps = [float(x) for x in avg_pts if isinstance(x, (int, float))]
if len(ps) > 40:
step = (len(ps) - 1) / 39.0
sig_curve = [ps[round(i * step)] for i in range(40)]
else:
sig_curve = ps
except Exception: # noqa: BLE001
sig_curve = []
profiles.append(
{
"name": name,
"avg_duration": profile_meta.get("avg_duration", 0),
"min_duration": profile_meta.get("min_duration", 0),
"max_duration": profile_meta.get("max_duration", 0),
"sample_cycle_id": profile_meta.get("sample_cycle_id"),
"cycle_count": cycle_count,
"last_run": last_run,
"avg_energy": avg_energy,
"duration_std_dev": duration_std_dev,
"avg_cost": avg_cost,
"total_cost": total_cost,
"signature_curve": sig_curve,
"is_imported": self.profile_has_reference_cycles(name),
}
)
return sorted(profiles, key=lambda p: profile_sort_key(p.get("name", "")))
async def create_profile_standalone(
self,
name: str,
reference_cycle_id: str | None = None,
avg_duration: float | None = None,
) -> None:
"""Create a profile without immediately labeling a cycle.
If reference_cycle_id is provided, use that cycle's characteristics.
If avg_duration is provided (and no reference cycle), use it as baseline."""
if name in self._data.get("profiles", {}):
raise ValueError(f"Profile '{name}' already exists")
profile_data: JSONDict = {}
if reference_cycle_id:
cycle = next(
(c for c in self._data["past_cycles"] if c["id"] == reference_cycle_id),
None,
)
if cycle:
profile_data = {
"avg_duration": cycle["duration"],
"sample_cycle_id": reference_cycle_id,
}
# Label the reference cycle with the new profile so that
# statistics are immediately populated after creation.
if not cycle.get("profile_name"):
cycle["profile_name"] = name
elif avg_duration is not None and avg_duration > 0:
profile_data = {
"avg_duration": float(avg_duration),
}
# Create profile with minimal data (will be updated when cycles are labeled)
profile_data.setdefault("phases", [])
self._data.setdefault("profiles", {})[name] = profile_data
# Build the envelope from any already-labeled cycles (e.g. reference cycle above)
await self.async_rebuild_envelope(name)
await self.async_save()
self._logger.info("Created standalone profile '%s'", name)
async def update_profile(
self, old_name: str, new_name: str, avg_duration: float | None = None
) -> int:
"""Update a profile's name and/or average duration.
Returns number of cycles updated (if renamed)."""
profiles = self._data.get("profiles", {})
if old_name not in profiles:
raise ValueError(f"Profile '{old_name}' not found")
# Handle Rename
renamed = False
if new_name != old_name:
if new_name in profiles:
raise ValueError(f"Profile '{new_name}' already exists")
# Rename in profiles dict
profiles[new_name] = profiles.pop(old_name)
# Rename in envelopes
if "envelopes" in self._data and old_name in self._data["envelopes"]:
self._data["envelopes"][new_name] = self._data["envelopes"].pop(
old_name
)
renamed = True
target_name = new_name if renamed else old_name
# Handle Duration Update
if avg_duration is not None and avg_duration > 0:
profiles[target_name]["avg_duration"] = float(avg_duration)
# If there's an envelope, we ideally update its target_duration too,
# but envelope is usually rebuilt from data.
# However, for manual profiles, envelope might be empty or theoretical.
# Let's log it.
self._logger.info(
"Updated baseline duration for '%s' to %ss",
target_name,
avg_duration,
)
# Update cycles and feedback if renamed
count = 0
if renamed:
# 1. Update past + imported reference cycles (imports carry profile_name
# too; leaving them under the old name orphans them from the matcher).
for cycle in (
list(self._data.get("past_cycles", []))
+ list(self._data.get("reference_cycles", []))
):
if cycle.get("profile_name") == old_name:
cycle["profile_name"] = new_name
count += 1
# 2. Update pending feedback
pending = self.get_pending_feedback()
for req in pending.values():
if req.get("detected_profile") == old_name:
req["detected_profile"] = new_name
# 3. Update feedback history
history = self.get_feedback_history()
for record in history.values():
if record.get("original_detected_profile") == old_name:
record["original_detected_profile"] = new_name
if record.get("corrected_profile") == old_name:
record["corrected_profile"] = new_name
# 4. Update group membership (rename the member in any group)
for g in self._data.get("profile_groups", {}).values():
mems = g.get("members") if isinstance(g, dict) else None
if isinstance(mems, list) and old_name in mems:
g["members"] = [new_name if m == old_name else m for m in mems]
self._logger.info(
"Renamed profile '%s' to '%s', updated %s cycles and associated feedback",
old_name,
new_name,
count,
)
await self.async_save()
return count
async def delete_profile(self, name: str, unlabel_cycles: bool = True) -> int:
"""Delete a profile.
If unlabel_cycles=True, removes profile label from cycles.
If unlabel_cycles=False, cycles keep the label (orphaned).
Returns number of cycles affected."""
if name not in self._data.get("profiles", {}):
raise ValueError(f"Profile '{name}' not found")
# Delete profile
del self._data["profiles"][name]
# Handle cycles (past + imported reference; both carry profile_name, so an
# imported cycle would otherwise keep a dangling label for a deleted profile).
count = 0
for cycle in (
list(self._data.get("past_cycles", []))
+ list(self._data.get("reference_cycles", []))
):
if cycle.get("profile_name") == name:
if unlabel_cycles:
cycle["profile_name"] = None
count += 1
await self.async_save()
action = "unlabeled" if unlabel_cycles else "orphaned"
self._logger.info("Deleted profile '%s', %s %s cycles", name, action, count)
return count
async def clear_all_data(self) -> None:
"""Clear all profiles, cycle data, and derived state."""
self._data["past_cycles"] = []
self._data["reference_cycles"] = []
self._data["profiles"] = {}
self._data["envelopes"] = {}
self._data["suggestions"] = {}
self._data["feedback_history"] = {}
self._data["pending_feedback"] = {}
self._data["auto_adjustments"] = []
self._data["active_cycle"] = None
self._data["last_active_save"] = None
# Newer persisted state must also be wiped, else a "wipe all" leaves trained
# models, groups, matcher tuning, histories, and counters behind.
self._data["custom_phases"] = []
self._data["ml_model_versions"] = {}
self._data["profile_groups"] = {}
self._data["maintenance_log"] = []
self._data["matching_config"] = {}
self._data["match_ranking_history"] = []
self._data["ml_last_training_run"] = None
self._data["ml_training_history"] = {}
self._data["lifetime_energy_wh"] = 0.0
self._data["lifetime_cycle_count"] = 0
self._data["settings_changelog"] = []
self._data["suggestion_apply_cycle_count"] = 0
self._cached_sample_segments = {}
self._cohesion_cache = {}
self._cohesion_cache_generation += 1
await self.async_save()
self._logger.info("Cleared all WashData storage")
async def assign_profile_to_cycle(
self, cycle_id: str, profile_name: str | None
) -> None:
"""Assign an existing profile to a cycle. Rebuilds envelope."""
old_profile = None
cycle = next(
(c for c in self._data["past_cycles"] if c["id"] == cycle_id), None
)
if not cycle:
# Imported reference recording (separate list, never in usage stats):
# relabelling just moves which profile's template it seeds.
ref = next(
(c for c in self.get_reference_cycles() if c.get("id") == cycle_id),
None,
)
if ref is not None:
await self._assign_reference_cycle_profile(ref, profile_name)
return
raise ValueError(f"Cycle {cycle_id} not found")
# Track old profile for envelope rebuild
old_profile = cycle.get("profile_name")
if profile_name and profile_name not in self._data.get("profiles", {}):
raise ValueError(f"Profile '{profile_name}' not found. Create it first.")
# Preserve original auto-assigned label before first manual relabeling
if profile_name and old_profile and not cycle.get("original_auto_label"):
orig_src = cycle.get("label_source", "")
if orig_src in ("auto_match", "auto_label_post", "auto_label_service"):
cycle["original_auto_label"] = old_profile
# Update cycle
cycle["profile_name"] = profile_name if profile_name else None
cycle["label_source"] = "manual" if profile_name else None
# Update profile metadata if this is the first cycle
if profile_name:
profile = self._data["profiles"][profile_name]
if not profile.get("sample_cycle_id"):
profile["sample_cycle_id"] = cycle_id
profile["avg_duration"] = cycle["duration"]
# Rebuild envelopes for affected profiles
if old_profile and old_profile != profile_name:
await self.async_rebuild_envelope(old_profile) # Old profile lost a cycle
if profile_name:
await self.async_rebuild_envelope(profile_name) # New profile gained a cycle
# Apply retention after labeling, in case profile now exceeds cap
await self.async_enforce_retention()
await self.async_save()
self._logger.info("Assigned profile '%s' to cycle %s", profile_name, cycle_id)
# Trigger smart processing to potentially merge now-labeled cycle
await self.async_smart_process_history()
async def _assign_reference_cycle_profile(
self, ref: CycleDict, profile_name: str | None
) -> None:
"""Reassign an imported reference recording to a different profile.
The cycle stays in ``reference_cycles`` (out of usage stats); only which
profile envelope it seeds changes. Rebuilds the old and new envelopes.
``profile_name=None`` clears the label (the recording then seeds nothing).
"""
if profile_name and profile_name not in self._data.get("profiles", {}):
raise ValueError(f"Profile '{profile_name}' not found. Create it first.")
old_profile = ref.get("profile_name")
ref_id = ref.get("id")
ref["profile_name"] = profile_name if profile_name else None
if old_profile and old_profile != profile_name:
# The moved cycle may have been the old profile's sample. Clear that
# stale pointer so the old profile can't resolve the moved trace (now
# another profile's) by id, and drop the old profile if it is now empty
# (mirrors _delete_reference_cycle; prevents repair adopting a real cycle).
op = self._data.get("profiles", {}).get(old_profile)
if op is not None and op.get("sample_cycle_id") == ref_id:
op["sample_cycle_id"] = None
old_has_cycles = any(
c.get("profile_name") == old_profile
for c in list(self._data.get("past_cycles", []))
+ list(self._data.get("reference_cycles", []))
)
if old_has_cycles:
await self.async_rebuild_envelope(old_profile)
else:
self._data.get("profiles", {}).pop(old_profile, None)
self._data.get("envelopes", {}).pop(old_profile, None)
if profile_name:
await self.async_rebuild_envelope(profile_name)
await self.async_save()
self._logger.info(
"Reassigned imported reference cycle %s to profile '%s'",
ref.get("id"), profile_name,
)
async def auto_label_cycles(
self, confidence_threshold: float = 0.75, overwrite: bool = False
) -> dict[str, int]:
"""Auto-label cycles retroactively using profile matching.
Args:
confidence_threshold: Min confidence to apply a label.
overwrite: If True, re-evaluates already labeled cycles.
Returns stats: {labeled: int, relabeled: int, skipped: int, total: int}
"""
stats = {"labeled": 0, "relabeled": 0, "skipped": 0, "total": 0}
cycles = self._data.get("past_cycles", [])
# Filter down if not overwriting
if not overwrite:
target_cycles = [c for c in cycles if not c.get("profile_name")]
else:
target_cycles = cycles
stats["total"] = len(target_cycles)
for cycle in target_cycles:
# Reconstruct power data for matching
power_data = decompress_power_data(cycle)
if not power_data or len(power_data) < 10:
stats["skipped"] += 1
continue
# Try to match
result = await self.async_match_profile(power_data, cycle["duration"])
# Honor the ambiguity safeguard: never auto-label a close/ambiguous match.
if result.best_profile and result.confidence >= confidence_threshold and not result.is_ambiguous:
current_label = cycle.get("profile_name")
# Sanitize: strip heavy current/sample arrays before persisting.
ranking_top5 = [
{
"name": c.get("name"),
"score": round(float(c.get("score", 0.0)), 3),
"profile_duration": c.get("profile_duration"),
}
for c in (getattr(result, "ranking", []) or [])[:5]
]
# If overwriting, check if new match is different and better/valid
if current_label:
if current_label != result.best_profile:
# Preserve original label before first auto-service relabeling
if not cycle.get("original_auto_label"):
orig_src = cycle.get("label_source", "")
if orig_src in ("auto_match", "auto_label_post", "auto_label_service"):
cycle["original_auto_label"] = current_label
cycle["profile_name"] = result.best_profile
cycle["match_confidence"] = float(result.confidence)
cycle["label_source"] = "auto_label_service"
if ranking_top5:
cycle["match_ranking_top5"] = ranking_top5
stats["relabeled"] += 1
self._logger.info(
"Relabeled cycle %s: '%s' -> '%s' (confidence: %.2f)",
cycle["id"],
current_label,
result.best_profile,
result.confidence,
)
else:
cycle["profile_name"] = result.best_profile
cycle["match_confidence"] = float(result.confidence)
cycle["label_source"] = "auto_label_service"
if ranking_top5:
cycle["match_ranking_top5"] = ranking_top5
stats["labeled"] += 1
self._logger.info(
"Auto-labeled cycle %s as '%s' (confidence: %.2f)",
cycle["id"],
result.best_profile,
result.confidence,
)
else:
stats["skipped"] += 1
if stats["labeled"] > 0 or stats["relabeled"] > 0:
await self.async_save()
# Trigger smart processing after bulk labeling
await self.async_smart_process_history()
self._logger.info(
"Auto-labeling complete: %s labeled, %s relabeled, %s skipped",
stats["labeled"],
stats["relabeled"],
stats["skipped"],
)
return stats
async def async_backfill_match_confidence(self) -> int:
"""Populate match_confidence for labeled cycles that predate the field.
Runs the matcher once per cycle with profile_name set but no
match_confidence, and persists the resulting confidence if the same
profile is returned. Returns the number of cycles updated. Safe to
call repeatedly - already-backfilled cycles are skipped.
"""
cycles = self._data.get("past_cycles", []) or []
updated = 0
for cycle in cycles:
if cycle.get("match_confidence") is not None:
continue
profile_name = cycle.get("profile_name")
if not profile_name:
continue
power_data = decompress_power_data(cycle)
if not power_data or len(power_data) < 10:
continue
try:
result = await self.async_match_profile(
power_data, cycle.get("duration", 0)
)
except Exception: # pylint: disable=broad-exception-caught
continue
if result.best_profile == profile_name and result.confidence > 0:
cycle["match_confidence"] = float(result.confidence)
updated += 1
if updated:
await self.async_save()
self._logger.info("Backfilled match_confidence on %d cycles", updated)
return updated
async def async_migrate_cycles_to_compressed(self) -> int:
"""
Migrate all cycles to the compressed format.
Ensures all cycles use [offset_seconds, power] format.
Returns number of cycles migrated.
"""
raw_cycles = self._data.get("past_cycles", [])
cycles: list[CycleDict] = (
cast(list[CycleDict], raw_cycles) if isinstance(raw_cycles, list) else []
)
migrated = 0
for cycle in cycles:
raw_data: list[Any] = cycle.get("power_data", []) or []
if not raw_data:
continue
# Check if already compressed (first element is number or mixed format)
first_elem = raw_data[0][0]
if isinstance(first_elem, (int, float)):
# Already in offset format
continue
# Old format: ISO timestamp strings. Convert to compressed offsets.
try:
compressed = compress_power_data(cycle)
if compressed:
cycle["power_data"] = compressed
migrated += 1
except Exception as e: # pylint: disable=broad-exception-caught
self._logger.warning("Failed to migrate cycle %s: %s", cycle.get("id"), e)
continue
if migrated > 0:
self._logger.info("Migrated %s cycles to compressed format", migrated)
await self.async_save()
return migrated
async def async_smart_process_history(
self
) -> dict[str, int]:
# Orchestrate smart history processing: Cleanup, Retention.
# Split/Merge is now manual via Interactive Editor.
stats = {"cleaned_profiles": 0}
# 1. Cleanup
self._logger.debug("Running maintenance: cleanup_orphaned_profiles")
stats["cleaned_profiles"] = self.cleanup_orphaned_profiles()
# 2. Retention
self._logger.debug("Running maintenance: async_enforce_retention")
await self.async_enforce_retention()
# 3. Save
self._logger.debug("Maintenance complete, saving")
await self.async_save()
return stats
def export_data(
self, entry_data: JSONDict | None = None, entry_options: JSONDict | None = None
) -> JSONDict:
# Return a serializable snapshot of the store for backup/export.
# Includes config entry data/options so users can transfer fine-tuned settings.
opts = entry_options or {}
data = entry_data or {}
_FINGERPRINT_KEYS = (
"off_delay", "min_power", "sampling_interval",
"start_power_threshold", "idle_power_threshold",
"min_cycle_duration", "max_cycle_duration",
"delay_start_detect_enabled",
)
device_fingerprint: JSONDict = {
"device_type": data.get("device_type") or opts.get("device_type", "unknown"),
}
for key in _FINGERPRINT_KEYS:
if key in opts:
device_fingerprint[key] = opts[key]
# Never let a backup/diagnostics export carry the GitHub refresh token: the
# (legacy, now-global) per-device store account may still hold it. Shallow-copy
# so we can drop the credential without mutating live state, and so a caller
# can't alias-mutate self._data through the returned snapshot.
export_store = dict(self._data)
export_store.pop("store_account", None)
return {
"version": STORAGE_VERSION,
"entry_id": self.entry_id,
"exported_at": dt_util.now().isoformat(),
"device_fingerprint": device_fingerprint,
"data": export_store,
"entry_data": data,
"entry_options": opts,
}
async def async_import_data(self, payload: dict[str, Any]) -> dict[str, Any]:
# Import data from JSON payload (migration aware).
# Unwrap HA diagnostics download file (outer HA wrapper: {home_assistant, data, ...})
if "home_assistant" in payload and "data" in payload:
payload = payload["data"]
self._logger.info("Detected HA diagnostics file wrapper, unwrapping 'data'")
# Unwrap our integration's diagnostics format ({entry, manager_state, store_export, ...})
if "store_export" in payload:
payload = payload["store_export"]
self._logger.info("Detected diagnostics store_export format, unwrapping")
version = payload.get("version", 1)
# Handle v1 format (flat structure) - convert to v2
if version == 1 or "data" not in payload:
# V1 format had profiles/past_cycles at top level
data_dict = {
"profiles": payload.get("profiles", {}),
"past_cycles": payload.get("past_cycles", []),
"envelopes": payload.get("envelopes", {}),
}
self._logger.info(
"Importing v1 format: %s cycles", len(data_dict.get("past_cycles", []))
)
else:
# V2 format with nested "data" key
data = payload.get("data")
if not isinstance(data, dict):
raise ValueError(
"Invalid export payload (missing or invalid 'data' key)"
)
data_dict = cast(JSONDict, data)
self._logger.info(
"Importing v2 format: %s cycles", len(data_dict.get("past_cycles", []))
)
# Validate and repair structure
if not isinstance(data_dict.get("profiles"), dict):
data_dict["profiles"] = {}
if not isinstance(data_dict.get("past_cycles"), list):
data_dict["past_cycles"] = []
if not isinstance(data_dict.get("reference_cycles"), list):
data_dict["reference_cycles"] = []
data_dict.setdefault("envelopes", {})
if not data_dict.get("profiles") and not data_dict.get("past_cycles"):
raise ValueError(
"Import payload contains no profiles or cycles — aborting to prevent data loss"
)
self._data = data_dict
self._cached_sample_segments = {}
await self.async_save()
# Strip diagnostic redaction sentinels so they don't overwrite real settings
def _strip_redacted(d: dict) -> dict:
if not isinstance(d, dict):
return {}
return {k: v for k, v in d.items() if v != "**REDACTED**"}
return {
"entry_data": _strip_redacted(payload.get("entry_data", {})),
"entry_options": _strip_redacted(payload.get("entry_options", {})),
}
async def delete_cycle(self, cycle_id: str) -> bool:
"""Delete a cycle by ID."""
cycles = cast(list[CycleDict], self._data.get("past_cycles", []))
initial_len = len(cycles)
cycle_to_delete = next((c for c in cycles if c.get("id") == cycle_id), None)
if not cycle_to_delete:
# Not a real cycle -- it may be an imported store recording, which
# lives in the separate reference_cycles list (never in usage stats).
return await self._delete_reference_cycle(cycle_id)
profile_name = cycle_to_delete.get("profile_name")
self._data["past_cycles"] = [c for c in cycles if c.get("id") != cycle_id]
if len(self._data["past_cycles"]) < initial_len:
# Check profile references
for _p_name, p_data in self.get_profiles().items():
if p_data.get("sample_cycle_id") == cycle_id:
p_data["sample_cycle_id"] = None
# Drop any pending feedback for this cycle so it can't orphan the
# "needs review" count (the count would say 1 while the cycle is gone).
self._data.get("pending_feedback", {}).pop(cycle_id, None)
# Rebuild envelope if cycle belonged to a profile
if profile_name:
await self.async_rebuild_envelope(profile_name)
await self.async_save()
return True
return False
async def _delete_reference_cycle(self, cycle_id: str) -> bool:
"""Delete a single imported store recording from ``reference_cycles``.
Rebuilds the affected profile's envelope so removing a bad import
immediately stops influencing the matcher template. Returns False when
no reference cycle carries that id.
"""
refs = cast(list[CycleDict], self._data.get("reference_cycles", []))
cycle = next((c for c in refs if c.get("id") == cycle_id), None)
if cycle is None:
return False
profile_name = cycle.get("profile_name")
self._data["reference_cycles"] = [c for c in refs if c.get("id") != cycle_id]
# Clear any profile that sampled this now-deleted reference cycle, mirroring
# the real-cycle path in delete_cycle, so no sample id is left dangling.
for _p_name, p_data in self.get_profiles().items():
if p_data.get("sample_cycle_id") == cycle_id:
p_data["sample_cycle_id"] = None
if profile_name:
# If removing this recording leaves the profile with no cycles at all
# (no real, no imported), drop the now-empty profile. Otherwise a
# sampleless import-only profile would be re-populated by
# async_repair_profile_samples stealing an unlabeled real cycle into it.
remaining = any(
c.get("profile_name") == profile_name
for c in list(self._data.get("past_cycles", []))
+ list(self._data.get("reference_cycles", []))
)
if remaining:
await self.async_rebuild_envelope(profile_name)
else:
self._data.get("profiles", {}).pop(profile_name, None)
self._data.get("envelopes", {}).pop(profile_name, None)
await self.async_save()
return True
def get_cycle_power_data(self, cycle_id: str) -> list[tuple[float, float]]:
"""Return decompressed power data for a cycle as [(offset_s, watts), ...].
Returns an empty list if the cycle is not found or has no power data.
"""
cycle = next(
(c for c in self.get_past_cycles() if c.get("id") == cycle_id), None
)
if cycle is None:
# Imported store recordings live in a separate list; the panel opens
# them from the same Cycles table, so look them up here too.
cycle = next(
(c for c in self.get_reference_cycles() if c.get("id") == cycle_id),
None,
)
if cycle is None:
return []
return decompress_power_data(cycle)
async def trim_cycle_power_data(
self,
cycle_id: str,
new_start_s: float,
new_end_s: float,
) -> bool:
"""Trim a cycle's power_data to the window [new_start_s, new_end_s].
Offsets are renormalized so the kept segment starts at 0.0.
The cycle's ``duration``, ``signature``, and ``sampling_interval`` are
recomputed from the trimmed data.
Returns True if successful, False if the cycle was not found or the
resulting data is empty.
"""
cycles = cast(list[CycleDict], self._data.get("past_cycles", []))
cycle = next((c for c in cycles if c.get("id") == cycle_id), None)
if cycle is None:
return False
p_data = decompress_power_data(cycle)
if not p_data:
return False
new_start_s = max(0.0, float(new_start_s))
new_end_s = float(new_end_s)
kept = sorted(
(
(offset, power)
for offset, power in p_data
if new_start_s <= offset <= new_end_s
),
key=lambda x: x[0],
)
if not kept:
return False
# Re-normalize offsets so the trimmed segment starts at 0.0
base = kept[0][0]
renorm: list[list[float]] = [
[round(offset - base, 2), power] for offset, power in kept
]
# Advance start_time when trimming from the front
if base > 0:
start_ts = _value_to_timestamp(cycle.get("start_time"))
if start_ts is not None:
cycle["start_time"] = dt_util.utc_from_timestamp(
start_ts + base
).isoformat()
# Recompute sampling interval
if len(renorm) > 1:
offsets_arr = np.array([r[0] for r in renorm], dtype=float)
intervals = np.diff(offsets_arr)
pos = intervals[intervals > 0]
sampling_interval = float(np.median(pos)) if len(pos) > 0 else 1.0
else:
sampling_interval = 1.0
# Recompute signature
ts_arr = np.array([r[0] for r in renorm], dtype=float)
p_arr = np.array([r[1] for r in renorm], dtype=float)
if len(ts_arr) > 1:
sig = compute_signature(ts_arr, p_arr)
cycle["signature"] = dataclasses.asdict(sig)
else:
cycle["signature"] = None
new_duration = round(renorm[-1][0], 1) if renorm else 0.0
cycle["power_data"] = renorm
cycle["sampling_interval"] = round(sampling_interval, 1)
cycle["duration"] = new_duration
# Keep end_time consistent with the updated start_time and duration
new_start_ts = _value_to_timestamp(cycle.get("start_time"))
if new_start_ts is not None:
cycle["end_time"] = dt_util.utc_from_timestamp(
new_start_ts + new_duration
).isoformat()
# Clear manual_duration override - trimmed duration is now authoritative
cycle.pop("manual_duration", None)
# Mark the cycle as edited so store-upload provenance (qc) can distinguish a
# trimmed cycle from a plain detected one.
meta = cycle.setdefault("meta", {})
if isinstance(meta, dict):
meta["edited"] = True
meta["trim"] = [round(new_start_s, 1), round(new_end_s, 1)]
# Invalidate cached sample segments for this cycle so future lookups
# are recomputed from the trimmed data
stale_keys = [k for k in self._cached_sample_segments if k[0] == cycle_id]
for k in stale_keys:
del self._cached_sample_segments[k]
# Rebuild envelope for the associated profile
profile_name = cycle.get("profile_name")
if profile_name:
await self.async_rebuild_envelope(profile_name)
await self.async_save()
return True
def analyze_split_sync(
self, cycle: CycleDict, min_gap_s: int = 900, idle_power: float = 2.0
) -> list[tuple[float, float]]:
"""Analyze cycle for potential splits (sync for executor)."""
p_data = decompress_power_data(cycle)
if not p_data:
return []
# Parse all points to (rel_t, power)
points: list[tuple[float, float]] = []
for offset_seconds, val in p_data:
points.append((float(offset_seconds), float(val)))
if not points:
return []
valid_segments: list[tuple[float, float]] = []
seg_start = 0.0
for i in range(1, len(points)):
t, _ = points[i]
prev_t, prev_p = points[i-1]
# Detect idle gap
gap = t - prev_t
if prev_p < idle_power and gap > min_gap_s:
# Segment ending at prev_t
if (prev_t - seg_start) > 60:
valid_segments.append((seg_start, prev_t))
seg_start = t
# Last segment
last_t = points[-1][0]
if (last_t - seg_start) > 60:
valid_segments.append((seg_start, last_t))
if len(valid_segments) < 2:
return []
self._logger.debug(
"Analyzed split for %s: found %d segments",
cycle.get("id"),
len(valid_segments)
)
return valid_segments
def build_split_segments_from_offsets(
self,
cycle: CycleDict,
split_offsets_s: list[float],
min_segment_s: float = 60.0,
) -> list[tuple[float, float]]:
"""Build segments for a manual split from explicit offsets (seconds from cycle start).
Returns adjacent [(start, end)] segments covering the cycle window, split at the
given offsets. Offsets are sorted and deduplicated; offsets outside the cycle window
or producing a sub-`min_segment_s` slice are dropped. Returns [] if fewer than two
segments would result.
"""
p_data = decompress_power_data(cycle)
if not p_data:
return []
last_t = float(p_data[-1][0])
if last_t <= 0:
return []
unique_offsets = sorted(
{round(float(o), 3) for o in split_offsets_s if 0.0 < float(o) < last_t}
)
if not unique_offsets:
return []
filtered_offsets: list[float] = []
for offset in unique_offsets:
if offset <= min_segment_s:
continue
if offset >= (last_t - min_segment_s):
continue
if filtered_offsets and (offset - filtered_offsets[-1]) < min_segment_s:
continue
filtered_offsets.append(offset)
if not filtered_offsets:
return []
boundaries = [0.0, *filtered_offsets, last_t]
segments: list[tuple[float, float]] = []
for i in range(len(boundaries) - 1):
seg_start = boundaries[i]
seg_end = boundaries[i + 1]
if (seg_end - seg_start) >= min_segment_s:
segments.append((seg_start, seg_end))
if len(segments) < 2:
return []
self._logger.debug(
"Built manual split for %s: %d segments at offsets %s",
cycle.get("id"),
len(segments),
filtered_offsets,
)
return segments
async def apply_split_interactive(
self, cycle_id: str, segments: list[dict[str, Any]]
) -> list[str]:
"""Apply a manual split config.
segments format: [{"start": float, "end": float, "profile": str|None}]
Returns list of new cycle IDs.
"""
cycles = cast(list[CycleDict], self._data.get("past_cycles", []))
idx = next((i for i, c in enumerate(cycles) if c.get("id") == cycle_id), -1)
if idx == -1:
return []
cycle = cycles[idx]
cycles.pop(idx) # Remove original
new_ids: list[str] = []
original_profile = cycle.get("profile_name")
start_dt_base_parsed = _parse_start_dt(cycle["start_time"])
if not start_dt_base_parsed:
return []
start_ts = start_dt_base_parsed.timestamp()
# Decompress original data
p_data_tuples = decompress_power_data(cycle)
if not p_data_tuples:
return []
# Prepare points (relative seconds)
points: list[tuple[float, float]] = []
for offset_seconds, val in p_data_tuples:
points.append((float(offset_seconds), float(val)))
# Create new cycles
for seg in segments:
if isinstance(seg, (list, tuple)):
seg_tuple = cast(tuple[Any, ...] | list[Any], seg)
seg_start = float(seg_tuple[0])
seg_end = float(seg_tuple[1])
seg_profile = None
else:
seg_start = float(seg["start"])
seg_end = float(seg["end"])
seg_profile = seg.get("profile")
seg_dur = seg_end - seg_start
new_cycle_start = start_dt_base_parsed + timedelta(seconds=seg_start)
new_cycle_start_ts = new_cycle_start.timestamp()
# Extract points for this segment
p_data_abs: list[list[float]] = []
# Find closest state before/at start to ensure continuity?
# Or just take points strictly inside?
# Generally better to capture the state at start 0.
state_val = 0.0
for t, p in points:
if t <= seg_start:
state_val = p
else:
break
# Start point (t=0 relative to new cycle)
p_data_abs.append([round(new_cycle_start_ts, 1), state_val])
for t, p in points:
if seg_start < t <= seg_end:
p_data_abs.append([round(start_ts + t, 1), p])
# Create Cycle Record
new_cycle: dict[str, Any] = {
"start_time": new_cycle_start.isoformat(),
"end_time": (new_cycle_start + timedelta(seconds=seg_dur)).isoformat(),
"duration": round(seg_dur, 1),
"status": "completed",
"power_data": p_data_abs,
"profile_name": seg_profile
}
self.add_cycle(new_cycle)
new_ids.append(new_cycle["id"])
# Fix profile refs (handle original sample cycle logic)
original_sample_id = cycle.get("id")
best_replacement_id = None
longest_dur = 0
new_cycles_objs = [c for c in cycles if c["id"] in new_ids] # 'cycles' is mutated by add_cycle
for c in new_cycles_objs:
d = c.get("duration", 0)
if d > longest_dur:
longest_dur = d
best_replacement_id = c["id"]
if best_replacement_id and original_profile:
p_data = self._data["profiles"].get(original_profile)
if p_data and p_data.get("sample_cycle_id") == original_sample_id:
p_data["sample_cycle_id"] = best_replacement_id
# Rebuild envelopes ONLY for the profiles whose dataset actually changed:
# the original profile (it lost the parent cycle) plus any profile a labeled
# segment was assigned to. This replaces a blanket rebuild-all-envelopes in
# the caller, which re-scanned every profile serially and stalled low-power
# hosts on a split (issue #311 follow-up).
touched: set[str] = set()
if original_profile:
touched.add(original_profile)
for seg in segments:
if isinstance(seg, dict):
seg_prof = seg.get("profile")
if seg_prof:
touched.add(seg_prof)
for name in touched:
await self.async_rebuild_envelope(name)
await self.async_save()
self._logger.info("Interactive Split Applied to %s -> %s", cycle_id, new_ids)
return new_ids
async def apply_merge_interactive(
self, cycle_ids: list[str], target_profile: str | None
) -> str | None:
"""
Merge multiple past cycles into a single cycle record, filling gaps between traces with short zero-power segments.
Parameters:
cycle_ids (list[str]): Unordered set of past-cycle IDs to merge; at least two IDs are required.
The function internally sorts cycles by start_time and mutates the chronologically earliest cycle.
target_profile (str | None): Profile name to assign to the merged cycle, or `None` to leave unlabeled.
Description:
When successful, this sorts the provided cycles by start_time and replaces the earliest cycle with the merged cycle,
removing the other consumed cycles and updating related metadata.
Side effects:
- Updates the store's past_cycles (removes consumed cycles and replaces the first cycle with the merged record).
- Clears any `manual_duration` override on the resulting cycle.
- Updates `sample_cycle_id` references in profiles that pointed to removed cycles.
- Attempts to recompute and store the merged cycle's signature.
- Persists changes to storage and triggers envelope rebuilds for affected profiles.
Returns:
merged_id (str | None): The new merged cycle's ID if the merge was applied, `None` if the merge could not be performed.
"""
if len(cycle_ids) < 2:
return None
cycles = self.get_past_cycles()
target_cycles = [c for c in cycles if c.get("id") in cycle_ids]
if len(target_cycles) != len(cycle_ids):
return None
# Sort by time - use timestamp comparison to handle mixed timezone offsets correctly
def _cycle_start_ts(c: CycleDict) -> float:
ts = _value_to_timestamp(c.get("start_time"))
return ts if ts is not None else float("inf")
target_cycles.sort(key=_cycle_start_ts)
# Collect affected profiles for envelope rebuild
affected_profiles: set[str] = set()
for c in target_cycles:
if c.get("profile_name"):
affected_profiles.add(c["profile_name"])
if target_profile:
affected_profiles.add(target_profile)
# We modify the first cycle (c1) to become the merged one
c1 = target_cycles[0]
ids_to_remove: list[str] = []
# Base setup
c1_start_dt = _parse_start_dt(c1["start_time"])
if not c1_start_dt:
return None
# Helper to get parsed points from a cycle
def get_points(cy: CycleDict) -> list[tuple[float, float, float]]:
# content: [(timestamp, offset, power)]
raw = decompress_power_data(cy)
res: list[tuple[float, float, float]] = []
if not raw:
return []
base_dt = _parse_start_dt(cy["start_time"])
if base_dt is None:
return []
base_t = base_dt.timestamp()
for offset_seconds, val in raw:
t_abs = base_t + float(offset_seconds)
res.append((t_abs, float(offset_seconds), float(val)))
return res
# Start with C1 points
merged_points_abs: list[list[float]] = [] # [timestamp, power]
# Add C1 points
c1_pts = get_points(c1)
for t_abs, _, p in c1_pts:
merged_points_abs.append([t_abs, p])
# Use the maximum t_abs seen so far (guards against out-of-order or corrupted points)
last_t_abs = max((pt[0] for pt in c1_pts), default=c1_start_dt.timestamp())
# Iterate others
max_power = c1.get("max_power", 0)
for next_c in target_cycles[1:]:
c_start_dt = _parse_start_dt(next_c.get("start_time"))
if not c_start_dt:
continue
c_pts = get_points(next_c)
if not c_pts:
continue
current_start_ts = c_pts[0][0]
# --- GAP FILLING ---
gap = current_start_ts - last_t_abs
# If gap > 1s, inject 0W points to ensure graph drops to 0
if gap > 1.0:
merged_points_abs.append([last_t_abs + 0.1, 0.0])
merged_points_abs.append([current_start_ts - 0.1, 0.0])
# Append points; track the running maximum to guard against reversed/corrupt data
for t_abs, _, p in c_pts:
merged_points_abs.append([t_abs, p])
if t_abs > last_t_abs:
last_t_abs = t_abs
max_power = max(max_power, next_c.get("max_power", 0))
ids_to_remove.append(next_c["id"])
# Derive merged end time from power data when available; otherwise fall back to
# the end_time field of the last cycle (handles cycles without recorded power data).
if merged_points_abs:
# Use the maximum absolute timestamp from all collected data points
last_t_abs = max(pt[0] for pt in merged_points_abs)
final_end_dt = dt_util.utc_from_timestamp(last_t_abs)
else:
last_cycle = target_cycles[-1]
fallback_end_dt = _parse_start_dt(last_cycle.get("end_time"))
if fallback_end_dt is not None:
final_end_dt = fallback_end_dt
else:
final_end_dt = c1_start_dt
new_dur = (final_end_dt - c1_start_dt).total_seconds()
c1["end_time"] = final_end_dt.isoformat()
c1["duration"] = round(new_dur, 1)
c1["max_power"] = max_power
c1["profile_name"] = target_profile
# Remove manual_duration override so the freshly computed duration is shown
c1.pop("manual_duration", None)
# Generate new compressed power_data [offset, power]
new_power_data: list[list[float]] = []
c1_start_ts = c1_start_dt.timestamp()
for t_abs, p in merged_points_abs:
offset = round(t_abs - c1_start_ts, 1)
new_power_data.append([offset, float(p)])
c1["power_data"] = new_power_data
# New Hash ID
new_id = hashlib.sha256(f"{c1['start_time']}_{c1['duration']}".encode()).hexdigest()[:12]
old_c1_id = c1["id"]
c1["id"] = new_id
# Update references in profiles
all_removed_ids = ids_to_remove + [old_c1_id]
for p_data in self.get_profiles().values():
if p_data.get("sample_cycle_id") in all_removed_ids:
p_data["sample_cycle_id"] = new_id
# Remove consumed cycles
self._data["past_cycles"] = [
c for c in cycles if c.get("id") not in ids_to_remove
]
# Update signature
try:
ts_arr = np.array([pt[0] for pt in new_power_data], dtype=float)
p_arr = np.array([pt[1] for pt in new_power_data], dtype=float)
if len(ts_arr) > 1:
sig = compute_signature(ts_arr, p_arr)
c1["signature"] = dataclasses.asdict(sig)
except Exception as e: # pylint: disable=broad-exception-caught
self._logger.warning("Failed to update signature for merged cycle %s: %s", new_id, e)
await self.async_save()
self._logger.info("Interactive Merge Applied: %s -> %s", cycle_ids, new_id)
# Rebuild envelopes for all affected profiles
for p_name in affected_profiles:
await self.async_rebuild_envelope(p_name)
return new_id