# 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 dataclasses import dataclass 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])) 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