112 files

custom_components/ha_washdata/signal_processing.py

@@ -21,7 +21,10 @@ 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
@@ -97,6 +100,195 @@ def integrate_wh(
 
 
 
+# ─── 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]: