# 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 . """Signal processing primitives for WashData. Constraint: NumPy only. Constraint: All computations must be dt-aware (robust to irregular cadence). Constraint: Resampling must be segment-based (no interpolation across gaps). """ from __future__ import annotations from dataclasses import dataclass from collections.abc import Sequence from typing import List, Tuple import numpy as np @dataclass class Segment: """A continuous metrics segment suitable for matching. Attributes: timestamps: Uniformly spaced timestamps (seconds) power: Interpolated power values (Watts) mask: Boolean mask (True = valid, False = gap/invalid). In strict segmentation, typically all True, but support mask for partial validity. """ timestamps: np.ndarray power: np.ndarray mask: np.ndarray # Future extensibility: might add other channels here def energy_gap_threshold_s(timestamps: np.ndarray) -> float: """Data-driven gap threshold (seconds) for energy integration. Ten times the median sample interval, clamped to ``[60, 3600]``. Segments longer than this are treated as sensor outages and excluded from the energy sum, without masking valid slow-sampling configurations. Single source for both persistence paths (``manager._on_cycle_end`` / ``ProfileStore.add_cycle``). """ ts = np.asarray(timestamps, dtype=float) if ts.size < 2: return 3600.0 intervals = np.diff(np.sort(ts)) positive = intervals[intervals > 0] median_interval = float(np.median(positive)) if positive.size > 0 else 0.0 return float(np.clip(10.0 * median_interval, 60.0, 3600.0)) def integrate_wh( timestamps: np.ndarray, power: np.ndarray, *, max_gap_s: float | None = None, ) -> float: """Compute energy in Wh using trapezoidal integration. Args: timestamps: Array of timestamps in seconds (must be ascending). power: Array of power values in Watts. max_gap_s: When set, segments whose ``dt`` exceeds this (or is non-positive) are excluded, so sensor-outage gaps don't inflate the total. When ``None`` (default) every segment is integrated - the original behaviour. Returns: Energy in Watt-hours. """ if len(timestamps) < 2: return 0.0 # np.diff(timestamps) is in seconds; divide by 3600 for hours. dt_hours = np.diff(np.asarray(timestamps, dtype=float)) / 3600.0 power = np.asarray(power, dtype=float) # Trapezoidal rule: (p[i] + p[i+1]) / 2 * dt avg_power = (power[:-1] + power[1:]) * 0.5 if max_gap_s is None: return float(np.sum(avg_power * dt_hours)) mask = (dt_hours > 0) & (dt_hours <= float(max_gap_s) / 3600.0) return float(np.sum(avg_power[mask] * dt_hours[mask])) # ─── Time-weighted energy cost (#426) ───────────────────────────────────────── # # A dynamic tariff moves while the appliance runs, so the single price in force # when the cycle ended is not the price the energy was bought at. These helpers # integrate the power trace against a piecewise-constant price timeline instead. # # Pure and hass-free on purpose: the manager (live cycles), the recorder backfill # (historic cycles) and the tests all go through the same math, so a cycle costed # live and the same cycle recosted from recorder history cannot disagree. def compact_price_timeline( points: Sequence[Tuple[float, float]], *, max_points: int = 240, decimals: int = 6, ) -> List[Tuple[float, float]]: """Normalize a ``(offset_s, price)`` timeline for storage. Sorts by offset, rounds prices, drops entries that repeat the previous price (a tariff that did not change costs nothing to record), and coarsens to at most ``max_points`` by keeping the entries that introduce the largest price steps - the points that matter least to the integral go first, so the capped timeline stays close to the uncapped one instead of being truncated at an arbitrary time. """ cleaned: List[Tuple[float, float]] = [] for entry in points or []: try: offset = float(entry[0]) price = round(float(entry[1]), decimals) except (TypeError, ValueError, IndexError, OverflowError): # OverflowError: json keeps an oversized integer literal as an unbounded # int, and float() on one of those raises rather than returning inf. An # imported cycle's price_timeline reaches here unvalidated, and this # function's contract is to drop a malformed entry, not to raise. continue if not np.isfinite(offset) or not np.isfinite(price): # nan AND +-inf: "inf" parses out of a sensor state like any other # float, and an infinite price makes every downstream cost infinite. continue cleaned.append((offset, price)) if not cleaned: return [] cleaned.sort(key=lambda item: item[0]) deduped: List[Tuple[float, float]] = [] for offset, price in cleaned: if deduped and deduped[-1][1] == price: continue deduped.append((offset, price)) if max_points > 0 and len(deduped) > max_points: # Rank the entries by the size of the price step each one introduces and keep # the largest. Index 0 is never a candidate - it anchors the price in force at # cycle start. Ranked in one pass rather than removing the smallest step # repeatedly: this also runs on the event loop (the in-memory bound in # ``manager._append_price_sample``), and the quadratic version is what a # pathologically chatty price entity would pay for. ranked = sorted( range(1, len(deduped)), key=lambda i: abs(deduped[i][1] - deduped[i - 1][1]), reverse=True, ) keep = set(ranked[: max_points - 1]) keep.add(0) deduped = [deduped[i] for i in sorted(keep)] return deduped def integrate_wh_by_price( timestamps: np.ndarray, power: np.ndarray, price_points: Sequence[Tuple[float, float]], *, max_gap_s: float | None = None, ) -> List[Tuple[float, float]]: """Split a trace's energy across a piecewise-constant price timeline. Returns ``[(price, wh), ...]``, one entry per price point, in timeline order. Each trapezoid interval is charged whole to the price in force at its *midpoint* rather than being split at the price boundary. That keeps the sum of the returned Wh **exactly** equal to :func:`integrate_wh` over the same inputs, which is what lets the caller apportion an external meter reading across the segments without the two figures drifting apart. Splitting would also break the ``max_gap_s`` outage mask: a boundary inserted inside a gap would turn one excluded interval into two short included ones. The error is bounded by a single sample interval per price change - seconds against a tariff that steps hourly. """ try: prices = [float(p) for _, p in price_points or []] offsets = np.asarray([float(o) for o, _ in price_points or []], dtype=float) except (TypeError, ValueError, OverflowError): # Same JSON boundary as compact_price_timeline above, which already drops # an entry for exactly these reasons: an imported or hand-edited # price_timeline keeps an oversized integer literal as an unbounded int, # and float() on one raises rather than returning inf. Every in-repo # caller compacts first, so this only binds a direct caller - and for one # of those "unusable timeline" means no timeline, which is the empty # return cycle_cost below already falls back on. Returning a zero Wh per # price instead would claim the timeline was fine and the trace carried no # energy, a different statement that only lands on the same fallback by # accident. return [] if not prices: return [] ts = np.asarray(timestamps, dtype=float) pw = np.asarray(power, dtype=float) if ts.size < 2 or pw.size != ts.size: return [(price, 0.0) for price in prices] dt_hours = np.diff(ts) / 3600.0 avg_power = (pw[:-1] + pw[1:]) * 0.5 energy = avg_power * dt_hours if max_gap_s is not None: mask = (dt_hours > 0) & (dt_hours <= float(max_gap_s) / 3600.0) energy = np.where(mask, energy, 0.0) midpoints = (ts[:-1] + ts[1:]) * 0.5 # side="right" - 1 gives the last price point at or before the midpoint. # Clipped at 0 so a trace that starts before the first price point is charged # at that first price rather than dropped. idx = np.clip(np.searchsorted(offsets, midpoints, side="right") - 1, 0, len(prices) - 1) totals = np.bincount(idx, weights=energy, minlength=len(prices)) return [(prices[i], float(totals[i])) for i in range(len(prices))] def cycle_cost( timestamps: np.ndarray, power: np.ndarray, price_points: Sequence[Tuple[float, float]], *, max_gap_s: float | None = None, report_wh: float | None = None, ) -> Tuple[float, float] | None: """Time-weighted cost of a cycle, and the effective price per kWh it implies. ``report_wh`` is the user-facing energy figure when it differs from the integral - i.e. an external meter's start->end delta (issue #316). The segments are scaled so they sum to it, which apportions the meter's total by the shape of the power trace and guarantees ``cost == report_wh/1000 * effective_price``. Without that, the cost shown next to a kWh figure would be computed from a different amount of energy than the kWh figure itself. Returns ``None`` when the trace carries no energy to charge for, or when ``report_wh`` is given but is not a usable positive figure, so the caller can fall back to the single-price behaviour instead of reporting a false zero or a cost that describes a different amount of energy than the kWh beside it. """ segments = integrate_wh_by_price(timestamps, power, price_points, max_gap_s=max_gap_s) if not segments: return None integrated = sum(wh for _, wh in segments) if integrated <= 0: return None scale = 1.0 if report_wh is not None: try: report = float(report_wh) # OverflowError alongside the rest, as in compact_price_timeline above: an # imported or hand-edited record keeps an oversized integer literal as an # unbounded int, and float() on one raises instead of returning inf. This # function's contract is to hand the decision back, not to raise. except (TypeError, ValueError, OverflowError): return None if not np.isfinite(report) or report <= 0: # The caller asked for the cost of *this* figure. Costing the trace # integral instead would print a price beside a 0 kWh readout, so hand # the decision back rather than answer a question that was not asked. return None scale = report / integrated cost = sum(wh * scale / 1000.0 * price for price, wh in segments) # Finite inputs can still leave the finite range: a price near the float # ceiling against a multi-kWh trace overflows to inf, and a subnormal # `report_wh` (5e-324 passes every check above) makes `scale` underflow, so # the charged energy rounds to exactly 0.0 and the division raises. Both end # in a non-finite or undefined cost, which is what the caller's fallback is # for, so hand the decision back rather than publish one. charged_kwh = integrated * scale / 1000.0 if not np.isfinite(cost) or charged_kwh <= 0.0: return None effective_price = cost / charged_kwh if not np.isfinite(effective_price): return None return cost, effective_price def resample_uniform( timestamps: np.ndarray, power: np.ndarray, dt_s: float = 5.0, gap_s: float = 60.0 ) -> List[Segment]: """Resample irregularly sampled data onto a uniform grid, respecting gaps. Returns a LIST of Segments. Does NOT interpolate across gaps > gap_s. Args: timestamps: Raw timestamps (seconds). power: Raw power values. dt_s: Target uniform step size (seconds). gap_s: Max gap to interpolate across (seconds). Returns: List of Segment objects. """ if len(timestamps) < 2: return [] segments: List[Segment] = [] # Find indices where dt > gap_s diffs = np.diff(timestamps) break_indices = np.where(diffs > gap_s)[0] + 1 # Add start and end indices start_indices = np.concatenate(([0], break_indices)) end_indices = np.concatenate((break_indices, [len(timestamps)])) for start_idx, end_idx in zip(start_indices, end_indices): chunk_ts = timestamps[start_idx:end_idx] chunk_p = power[start_idx:end_idx] if len(chunk_ts) < 2: continue # Define uniform grid for this chunk # Define uniform grid for this chunk (start at first timestamp) # Simple approach: start at t[0], go to t[-1] stepping by dt_s grid_start = chunk_ts[0] grid_end = chunk_ts[-1] # Ensure at least two points if grid_end - grid_start < dt_s: continue # arange(start, end + epsilon, dt) target_ts = np.arange(grid_start, grid_end + 0.001, dt_s) # Use numpy interp (linear interpolation) # It's safe here because we know max gap < gap_s within this chunk interpolated_p = np.interp(target_ts, chunk_ts, chunk_p) segments.append( Segment( timestamps=target_ts, power=interpolated_p, mask=np.ones_like(target_ts, dtype=bool), ) ) return segments def resample_to_n(power: list[float], n: int) -> list[float]: """Resample a power trace to exactly *n* evenly-spaced points via linear interpolation. Works on raw power-value lists (no timestamp required — assumes uniform original spacing). Returns a plain Python list so callers can convert to NumPy as needed. Args: power: Input power values. 2+ points are interpolated; fewer are handled explicitly (see Returns). n: Desired number of output points. Returns: List of *n* float values, except: - returns the input unchanged when it already has exactly *n* points; - returns ``[]`` for non-positive *n* or an empty input (no data to resample — a "missing" marker, not fabricated zeros); - returns *n* copies of the sole value for a single-sample input. """ if len(power) == n: return list(power) if n < 1: return [] src = np.asarray(power, dtype=float) # An empty trace has no data to resample: return empty (a "missing" marker) # rather than fabricating n zeros that read as real zero-power samples. # Callers already guard empty/short input before calling. if src.size == 0: return [] # A single sample can only be replicated: return n copies of that value. if src.size == 1: return [float(src[0])] * n src_x = np.linspace(0.0, 1.0, src.size) dst_x = np.linspace(0.0, 1.0, n) # Return native Python floats (not np.float64) so callers/JSON get plain floats. return [float(v) for v in np.interp(dst_x, src_x, src)] def resample_adaptive( timestamps: np.ndarray, power: np.ndarray, min_dt: float = 5.0, gap_s: float = 300.0, ) -> Tuple[List[Segment], float]: """Resample data using an adaptive time step based on input cadence. Target dt is based on observed cadence with a lower bound: ``target_dt = max(min_dt, median_interval)``. - If data is dense (for example 1s), it is downsampled to ``min_dt``. - If data is sparse (for example 30s), cadence is preserved. Args: timestamps: Raw timestamps (seconds). power: Raw power values. min_dt: Minimum allowed dt (seconds). gap_s: Max gap to interpolate across. Returns: Tuple of ``(segments, used_dt_s)`` where ``segments`` are gap-aware, uniformly sampled chunks and ``used_dt_s`` is the chosen target step. """ if len(timestamps) < 2: return [], min_dt # Determine cadence diffs = np.diff(timestamps) # Filter strictly zero diffs (duplicates) valid_diffs = diffs[diffs > 0.001] if len(valid_diffs) == 0: median_dt = min_dt else: median_dt = float(np.median(valid_diffs)) # Logic: Never resample finer than sensor (median_dt). # Also enforce min_dt (don't go finer than 5s). # We ignore max_dt for clamping down, to respect "never finer" rule. min_dt = max(min_dt, 1e-3) # Guard against non-positive step target_dt = max(min_dt, median_dt) gap_s = max(gap_s, target_dt * 1.5, 1e-3) # Guard against non-positive gap # Delegate to uniform resampler with chosen dt segments = resample_uniform(timestamps, power, dt_s=target_dt, gap_s=gap_s) return segments, target_dt