ashvale-station/ashvale/storage.py

# Copyright 2026 Kemal Yaylali
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Durable storage: SQLite in WAL mode with tiered downsampling.

An SD card is a consumable. The write pattern here is deliberately
gentle: one row every `persist_period_s`, WAL journalling, a compaction
pass that folds week-old raw rows into 5-minute means and quarter-old
5-minute rows into hourly means. A year of station history lands around
30 MB, which the Pi will not notice.
"""

from __future__ import annotations

import sqlite3
import threading
import time
from typing import Any, Dict, Iterable, List, Optional, Tuple

import numpy as np

TIER_RAW = 0
TIER_5MIN = 1
TIER_HOUR = 2

COLUMNS = [
    "ts", "temp_raw", "temp_h", "temp_p", "temp_c", "temp_smooth", "temp_rate",
    "hum", "hum_smooth",
    "press", "press_slp", "press_smooth", "press_rate", "cpu_temp", "dew_c",
    "lux", "r", "g", "b", "pitch", "roll", "yaw", "compass",
    "ax", "ay", "az", "gx", "gy", "gz",
]

SCHEMA = f"""
PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;
PRAGMA temp_store=MEMORY;

CREATE TABLE IF NOT EXISTS telemetry (
    ts REAL PRIMARY KEY,
    {", ".join(f"{c} REAL" for c in COLUMNS if c != "ts")},
    tier INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_telemetry_tier_ts ON telemetry(tier, ts);

CREATE TABLE IF NOT EXISTS forecasts (
    issued_ts REAL NOT NULL,
    valid_ts REAL NOT NULL,
    horizon_s INTEGER NOT NULL,
    target TEXT NOT NULL,
    mu REAL, lo REAL, hi REAL,
    model TEXT,
    -- What each ensemble member said at issue time. The blend weights are
    -- learned from how these actually turned out, and the learned member is
    -- not reconstructable after the fact: the RLS has moved on. If they are
    -- not written down here the Hedge update has nothing honest to eat.
    m_persistence REAL, m_climatology REAL, m_learned REAL, m_setpoint REAL,
    -- A matured forecast stays readable for an hour so the scorecard can
    -- aggregate a rolling window, but it must teach the learner exactly once.
    scored INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (issued_ts, horizon_s, target)
);
CREATE INDEX IF NOT EXISTS idx_forecast_valid ON forecasts(valid_ts);

CREATE TABLE IF NOT EXISTS labels (
    ts REAL NOT NULL,
    kind TEXT NOT NULL,
    value REAL NOT NULL,
    note TEXT,
    PRIMARY KEY (ts, kind)
);

CREATE TABLE IF NOT EXISTS scores (
    ts REAL NOT NULL,
    target TEXT NOT NULL,
    horizon_s INTEGER NOT NULL,
    mae REAL, rmse REAL, bias REAL,
    mae_persistence REAL, skill REAL, coverage REAL, n INTEGER,
    PRIMARY KEY (ts, target, horizon_s)
);

CREATE TABLE IF NOT EXISTS events (
    ts REAL NOT NULL,
    kind TEXT NOT NULL,
    severity TEXT,
    detail TEXT
);
CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts);
"""


def _f(v: Any) -> Optional[float]:
    return None if v is None else float(v)


# Columns the forecasts table has gained since the first schema, with their
# declarations. Same hazard as COLUMNS above: a live station's table already
# exists, so CREATE TABLE IF NOT EXISTS will not add them.
FORECAST_COLUMNS = [
    ("m_persistence", "REAL"),
    ("m_climatology", "REAL"),
    ("m_learned", "REAL"),
    ("m_setpoint", "REAL"),
    ("scored", "INTEGER NOT NULL DEFAULT 0"),
]


class Store:
    def __init__(self, path: str):
        self.path = path
        self._local = threading.local()
        with self._conn() as conn:
            conn.executescript(SCHEMA)
            self._migrate(conn)

    @staticmethod
    def _migrate(conn: sqlite3.Connection) -> None:
        """Add any column COLUMNS has gained since this database was created.

        CREATE TABLE IF NOT EXISTS is a no-op against a table that already
        exists, so a new entry in COLUMNS reaches a fresh install and silently
        misses every station that has been running. The failure then surfaces
        as an OperationalError inside insert_telemetry, which is on the sample
        loop, so a column addition would take a live station down rather than
        merely leaving a gap in its record.
        """
        have = {row[1] for row in conn.execute("PRAGMA table_info(telemetry)")}
        if not have:
            return
        for col in COLUMNS:
            if col in have:
                continue
            if not col.isidentifier():
                raise ValueError(f"refusing to splice a non-identifier column: {col!r}")
            conn.execute(f"ALTER TABLE telemetry ADD COLUMN {col} REAL")

        have = {row[1] for row in conn.execute("PRAGMA table_info(forecasts)")}
        if not have:
            return
        for col, decl in FORECAST_COLUMNS:
            if col in have:
                continue
            if not col.isidentifier():
                raise ValueError(f"refusing to splice a non-identifier column: {col!r}")
            conn.execute(f"ALTER TABLE forecasts ADD COLUMN {col} {decl}")

    def _conn(self) -> sqlite3.Connection:
        conn = getattr(self._local, "conn", None)
        if conn is None:
            conn = sqlite3.connect(self.path, timeout=20.0, check_same_thread=False)
            conn.row_factory = sqlite3.Row
            self._local.conn = conn
        return conn

    # ------------------------------------------------------------ writes

    def insert_telemetry(self, row: Dict[str, Any], tier: int = TIER_RAW) -> None:
        payload = {c: float(row.get(c)) if row.get(c) is not None else None for c in COLUMNS}
        payload["tier"] = tier
        cols = ", ".join(payload.keys())
        marks = ", ".join("?" for _ in payload)
        with self._conn() as conn:
            conn.execute(
                f"INSERT OR REPLACE INTO telemetry ({cols}) VALUES ({marks})",
                list(payload.values()),
            )

    def insert_forecast(self, issued_ts: float, horizon_s: int, target: str,
                        mu: float, lo: float, hi: float, model: str,
                        members: Optional[Dict[str, float]] = None) -> None:
        m = members or {}
        with self._conn() as conn:
            conn.execute(
                "INSERT OR REPLACE INTO forecasts "
                "(issued_ts, valid_ts, horizon_s, target, mu, lo, hi, model, "
                " m_persistence, m_climatology, m_learned, m_setpoint, scored) "
                "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,0)",
                (issued_ts, issued_ts + horizon_s, horizon_s, target,
                 float(mu), float(lo), float(hi), model,
                 _f(m.get("persistence")), _f(m.get("climatology")),
                 _f(m.get("learned")), _f(m.get("setpoint"))),
            )

    def mark_forecasts_scored(self, keys: Iterable[Tuple[float, int, str]]) -> int:
        """Flag forecasts as already fed to the learner."""
        rows = [(float(i), int(h), str(t)) for i, h, t in keys]
        if not rows:
            return 0
        with self._conn() as conn:
            cur = conn.executemany(
                "UPDATE forecasts SET scored = 1 "
                "WHERE issued_ts = ? AND horizon_s = ? AND target = ?", rows)
            return int(cur.rowcount)

    def insert_label(self, ts: float, kind: str, value: float, note: str = "") -> None:
        with self._conn() as conn:
            conn.execute(
                "INSERT OR REPLACE INTO labels (ts, kind, value, note) VALUES (?,?,?,?)",
                (ts, kind, float(value), note),
            )

    def insert_score(self, ts: float, target: str, horizon_s: int, **kw) -> None:
        with self._conn() as conn:
            conn.execute(
                "INSERT OR REPLACE INTO scores "
                "(ts, target, horizon_s, mae, rmse, bias, mae_persistence, skill, coverage, n) "
                "VALUES (?,?,?,?,?,?,?,?,?,?)",
                (ts, target, horizon_s, kw.get("mae"), kw.get("rmse"), kw.get("bias"),
                 kw.get("mae_persistence"), kw.get("skill"), kw.get("coverage"), kw.get("n")),
            )

    def log_event(self, kind: str, severity: str, detail: str, ts: Optional[float] = None) -> None:
        with self._conn() as conn:
            conn.execute("INSERT INTO events (ts, kind, severity, detail) VALUES (?,?,?,?)",
                         (ts or time.time(), kind, severity, detail))

    # ------------------------------------------------------------- reads

    def all_for_recompute(self) -> Dict[str, np.ndarray]:
        """Every stored row's *raw* inputs, oldest first.

        Only the columns a re-derivation actually needs. The raw sensor values
        are never overwritten, which is precisely what makes recomputation
        possible after a calibration changes k or the humidity offset.
        """
        cols = ["ts", "temp_raw", "cpu_temp", "hum", "press"]
        with self._conn() as conn:
            rows = conn.execute(
                f"SELECT {', '.join(cols)} FROM telemetry ORDER BY ts ASC").fetchall()
        if not rows:
            return {c: np.empty(0) for c in cols}
        arr = np.array(rows, dtype=object)
        out = {}
        for i, c in enumerate(cols):
            out[c] = np.array([np.nan if v is None else float(v) for v in arr[:, i]],
                              dtype=float)
        return out

    def apply_recompute(self, ts: np.ndarray, updates: Dict[str, np.ndarray],
                        chunk: int = 2000) -> int:
        """Write recomputed derived columns back, in chunks.

        Chunked because a year of tiered history is a six-figure row count and a
        single statement would hold the whole parameter list in memory on a
        512 MB board.
        """
        names = list(updates.keys())
        sql = (f"UPDATE telemetry SET {', '.join(n + ' = ?' for n in names)} "
               f"WHERE ts = ?")
        n_written = 0
        with self._conn() as conn:
            for start in range(0, ts.size, chunk):
                stop = min(start + chunk, ts.size)
                batch = [
                    tuple(
                        [None if not np.isfinite(updates[n][i]) else float(updates[n][i])
                         for n in names] + [float(ts[i])]
                    )
                    for i in range(start, stop)
                ]
                conn.executemany(sql, batch)
                n_written += len(batch)
        return n_written

    def window(self, hours: float, columns: Optional[Iterable[str]] = None) -> Dict[str, np.ndarray]:
        """Return the last `hours` of telemetry as column arrays, oldest first."""
        cols = list(columns) if columns else COLUMNS
        since = time.time() - hours * 3600.0
        with self._conn() as conn:
            cur = conn.execute(
                f"SELECT {', '.join(cols)} FROM telemetry WHERE ts >= ? ORDER BY ts ASC",
                (since,),
            )
            rows = cur.fetchall()
        if not rows:
            return {c: np.empty(0, dtype=float) for c in cols}
        arr = np.array([[r[c] if r[c] is not None else np.nan for c in cols] for r in rows],
                       dtype=float)
        return {c: arr[:, i] for i, c in enumerate(cols)}

    # ------------------------------------------------- historical access

    @staticmethod
    def auto_bucket(start: float, end: float, target_points: int = 700) -> int:
        """Pick a sensible aggregation bucket for a requested span.

        The browser cannot draw more than about a thousand points usefully
        and the Pi should not serialise more than it must, so the bucket
        grows with the span. Snapped to familiar durations so the x-axis
        reads in round numbers rather than 437-second increments.
        """
        span = max(float(end) - float(start), 1.0)
        raw = span / max(int(target_points), 1)
        ladder = [30, 60, 120, 300, 600, 900, 1800, 3600, 7200,
                  10800, 21600, 43200, 86400, 604800]
        for step in ladder:
            if raw <= step:
                return step
        return ladder[-1]

    def range_series(self, start: float, end: float,
                     bucket_s: Optional[int] = None) -> Dict[str, Any]:
        """Bucket-aggregated telemetry between two epoch timestamps.

        Aggregation happens in SQLite rather than numpy: pulling 90 days of
        rows into Python to average them would cost more memory than the
        Zero 2 W has to spare. Min and max travel alongside the mean so the
        UI can shade a true range band instead of implying the mean was the
        whole story.
        """
        start, end = float(start), float(end)
        if end <= start:
            return {"n": 0, "bucket_s": 0, "series": {}}
        bucket = int(bucket_s or self.auto_bucket(start, end))

        # The alias must not be a bare single letter: the telemetry table has
        # r, g and b colour columns, and SQLite resolves an unqualified name in
        # GROUP BY to a real column before a result alias. `GROUP BY b` silently
        # grouped by the blue channel and returned one row per sample while
        # cheerfully reporting the requested bucket size.
        sql = f"""
            SELECT CAST(ts / {bucket} AS INTEGER) * {bucket} AS bucket_ts,
                   AVG(temp_smooth)  AS temp,  MIN(temp_smooth) AS temp_lo,
                   MAX(temp_smooth)  AS temp_hi,
                   AVG(hum_smooth)   AS hum,   MIN(hum_smooth)  AS hum_lo,
                   MAX(hum_smooth)   AS hum_hi,
                   AVG(press_slp)    AS press, MIN(press_slp)   AS press_lo,
                   MAX(press_slp)    AS press_hi,
                   AVG(dew_c)        AS dew,   AVG(lux)         AS lux,
                   AVG(temp_rate)    AS temp_rate,
                   AVG(press_rate)   AS press_rate,
                   AVG(cpu_temp)     AS cpu,   COUNT(*)         AS n
            FROM telemetry
            WHERE ts >= ? AND ts <= ?
            GROUP BY bucket_ts ORDER BY bucket_ts ASC
        """
        with self._conn() as conn:
            rows = conn.execute(sql, (start, end)).fetchall()
        if not rows:
            return {"n": 0, "bucket_s": bucket, "series": {}}

        keys = ["temp", "temp_lo", "temp_hi", "hum", "hum_lo", "hum_hi",
                "press", "press_lo", "press_hi", "dew", "lux",
                "temp_rate", "press_rate", "cpu", "n"]
        out: Dict[str, list] = {"ts": [float(r["bucket_ts"]) for r in rows]}
        for k in keys:
            out[k] = [r[k] for r in rows]
        return {"n": len(rows), "bucket_s": bucket,
                "start": start, "end": end, "series": out}

    def daily_summary(self, start: float, end: float) -> List[Dict[str, Any]]:
        """Per-calendar-day extremes and means, in the station's local time.

        Local time, not UTC: a `daily minimum` that straddles midnight in
        the wrong timezone is the kind of quiet wrongness nobody notices
        until they compare against the Met Office and lose an afternoon.
        """
        sql = """
            SELECT date(ts, 'unixepoch', 'localtime') AS day,
                   MIN(ts) AS first_ts, MAX(ts) AS last_ts, COUNT(*) AS n,
                   MIN(temp_smooth) AS temp_min, MAX(temp_smooth) AS temp_max,
                   AVG(temp_smooth) AS temp_mean,
                   MIN(hum_smooth)  AS hum_min,  MAX(hum_smooth)  AS hum_max,
                   AVG(hum_smooth)  AS hum_mean,
                   MIN(press_slp)   AS press_min, MAX(press_slp)  AS press_max,
                   AVG(press_slp)   AS press_mean,
                   AVG(dew_c) AS dew_mean, MAX(lux) AS lux_max
            FROM telemetry
            WHERE ts >= ? AND ts <= ?
            GROUP BY day ORDER BY day DESC
        """
        with self._conn() as conn:
            return [dict(r) for r in conn.execute(sql, (float(start), float(end))).fetchall()]

    def extremes(self) -> Dict[str, Any]:
        """All-time records held by the station, each with when it happened."""
        pairs = [
            ("temp_max", "temp_smooth", "DESC"), ("temp_min", "temp_smooth", "ASC"),
            ("hum_max", "hum_smooth", "DESC"), ("hum_min", "hum_smooth", "ASC"),
            ("press_max", "press_slp", "DESC"), ("press_min", "press_slp", "ASC"),
            ("dew_max", "dew_c", "DESC"), ("dew_min", "dew_c", "ASC"),
            ("rate_rise", "press_rate", "DESC"), ("rate_fall", "press_rate", "ASC"),
        ]
        # Physical sanity bounds. A Kalman filter's rate estimate is garbage
        # for the first few samples after it initialises, which happens on
        # every restart, and an unfiltered MAX() will faithfully enshrine that
        # transient as an all-time record of -37 hPa/h forever. The most
        # extreme real sea-level pressure changes on Earth are around
        # 10 hPa/h in an explosively deepening cyclone.
        bounds = {"press_rate": 10.0, "temp_rate": 25.0}

        out: Dict[str, Any] = {}
        with self._conn() as conn:
            for name, col, order in pairs:
                guard = ""
                if col in bounds:
                    guard = f" AND ABS({col}) <= {bounds[col]}"
                row = conn.execute(
                    f"SELECT ts, {col} AS v FROM telemetry "
                    f"WHERE {col} IS NOT NULL{guard} ORDER BY {col} {order} LIMIT 1"
                ).fetchone()
                out[name] = {"ts": row["ts"], "value": row["v"]} if row else None
            span = conn.execute("SELECT MIN(ts) AS a, MAX(ts) AS b, COUNT(*) AS n "
                                "FROM telemetry").fetchone()
        out["coverage"] = {"first_ts": span["a"], "last_ts": span["b"],
                           "rows": span["n"]}
        return out

    def iter_csv(self, start: float, end: float):
        """Yield CSV lines for export. Generator, so a year of history does
        not have to exist in memory at once on a 512 MB board."""
        cols = ["ts", "temp_smooth", "hum_smooth", "press_slp", "dew_c",
                "temp_rate", "press_rate", "cpu_temp", "lux", "tier"]
        yield "iso_time," + ",".join(cols) + "\n"
        with self._conn() as conn:
            cur = conn.execute(
                f"SELECT {', '.join(cols)} FROM telemetry "
                f"WHERE ts >= ? AND ts <= ? ORDER BY ts ASC",
                (float(start), float(end)),
            )
            while True:
                chunk = cur.fetchmany(500)
                if not chunk:
                    break
                for r in chunk:
                    iso = time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(r["ts"]))
                    vals = ["" if r[c] is None else
                            (f"{r[c]:.4f}" if isinstance(r[c], float) else str(r[c]))
                            for c in cols]
                    yield iso + "," + ",".join(vals) + "\n"

    def storage_stats(self) -> Dict[str, Any]:
        """Rows per resolution tier, so the retention policy is visible."""
        with self._conn() as conn:
            rows = conn.execute(
                "SELECT tier, COUNT(*) AS n, MIN(ts) AS a, MAX(ts) AS b "
                "FROM telemetry GROUP BY tier ORDER BY tier"
            ).fetchall()
            page = conn.execute("PRAGMA page_count").fetchone()[0]
            size = conn.execute("PRAGMA page_size").fetchone()[0]
        names = {TIER_RAW: "raw", TIER_5MIN: "5 minute", TIER_HOUR: "hourly"}
        return {
            "tiers": [{"tier": r["tier"], "label": names.get(r["tier"], "?"),
                       "rows": r["n"], "first_ts": r["a"], "last_ts": r["b"]}
                      for r in rows],
            "bytes": int(page) * int(size),
        }

    def latest(self) -> Optional[Dict[str, Any]]:
        with self._conn() as conn:
            cur = conn.execute("SELECT * FROM telemetry ORDER BY ts DESC LIMIT 1")
            row = cur.fetchone()
        return dict(row) if row else None

    def row_count(self) -> int:
        with self._conn() as conn:
            return int(conn.execute("SELECT COUNT(*) FROM telemetry").fetchone()[0])

    def newest_ts(self) -> Optional[float]:
        with self._conn() as conn:
            row = conn.execute("SELECT MAX(ts) FROM telemetry").fetchone()
        return float(row[0]) if row and row[0] is not None else None

    def span_days(self) -> float:
        with self._conn() as conn:
            row = conn.execute("SELECT MIN(ts), MAX(ts) FROM telemetry").fetchone()
        if not row or row[0] is None:
            return 0.0
        return (row[1] - row[0]) / 86400.0

    def due_forecasts(self, now: Optional[float] = None) -> List[sqlite3.Row]:
        """Forecasts whose validity time has passed and can now be scored."""
        now = now or time.time()
        with self._conn() as conn:
            return conn.execute(
                "SELECT * FROM forecasts WHERE valid_ts <= ? AND valid_ts >= ? ORDER BY valid_ts",
                (now, now - 7 * 86400),
            ).fetchall()

    def scorecard(self) -> List[Dict[str, Any]]:
        with self._conn() as conn:
            rows = conn.execute(
                "SELECT s.* FROM scores s JOIN ("
                "  SELECT target, horizon_s, MAX(ts) AS mts FROM scores GROUP BY target, horizon_s"
                ") m ON s.target = m.target AND s.horizon_s = m.horizon_s AND s.ts = m.mts "
                "ORDER BY s.target, s.horizon_s"
            ).fetchall()
        return [dict(r) for r in rows]

    def recent_events(self, limit: int = 25) -> List[Dict[str, Any]]:
        with self._conn() as conn:
            rows = conn.execute(
                "SELECT * FROM events ORDER BY ts DESC LIMIT ?", (limit,)
            ).fetchall()
        return [dict(r) for r in rows]

    def labels(self, kind: str, hours: float = 24 * 30) -> Dict[str, np.ndarray]:
        since = time.time() - hours * 3600.0
        with self._conn() as conn:
            rows = conn.execute(
                "SELECT ts, value FROM labels WHERE kind = ? AND ts >= ? ORDER BY ts",
                (kind, since),
            ).fetchall()
        if not rows:
            return {"ts": np.empty(0), "value": np.empty(0)}
        return {
            "ts": np.array([r["ts"] for r in rows], dtype=float),
            "value": np.array([r["value"] for r in rows], dtype=float),
        }

    # -------------------------------------------------------- compaction

    def compact(self, raw_retention_days: float, five_min_retention_days: float) -> Dict[str, int]:
        """Fold old high-resolution rows into means. Returns rows removed per tier."""
        now = time.time()
        removed = {"raw": 0, "5min": 0}
        removed["raw"] = self._fold(TIER_RAW, TIER_5MIN, 300,
                                    now - raw_retention_days * 86400)
        removed["5min"] = self._fold(TIER_5MIN, TIER_HOUR, 3600,
                                     now - five_min_retention_days * 86400)
        with self._conn() as conn:
            conn.execute("PRAGMA incremental_vacuum")
        return removed

    def _fold(self, from_tier: int, to_tier: int, bucket_s: int, older_than: float) -> int:
        agg_cols = [c for c in COLUMNS if c != "ts"]
        select = ", ".join(f"AVG({c}) AS {c}" for c in agg_cols)
        with self._conn() as conn:
            rows = conn.execute(
                f"SELECT CAST(ts / {bucket_s} AS INTEGER) * {bucket_s} AS bucket, {select} "
                f"FROM telemetry WHERE tier = ? AND ts < ? GROUP BY bucket",
                (from_tier, older_than),
            ).fetchall()
            if not rows:
                return 0
            cur = conn.execute("SELECT COUNT(*) FROM telemetry WHERE tier = ? AND ts < ?",
                               (from_tier, older_than))
            n_before = int(cur.fetchone()[0])
            conn.execute("DELETE FROM telemetry WHERE tier = ? AND ts < ?",
                         (from_tier, older_than))
            payload = [
                tuple([float(r["bucket"])] + [r[c] for c in agg_cols] + [to_tier])
                for r in rows
            ]
            marks = ", ".join("?" for _ in range(len(agg_cols) + 2))
            conn.executemany(
                f"INSERT OR REPLACE INTO telemetry (ts, {', '.join(agg_cols)}, tier) "
                f"VALUES ({marks})",
                payload,
            )
        return n_before - len(rows)


def resample(ts: np.ndarray, values: Dict[str, np.ndarray], grid_s: int,
             max_gap_grid: int = 3):
    """Bin irregular samples onto a regular grid, mean-aggregating each bin.

    Returns (grid_ts, {name: array}) with NaN in bins that had no data and
    linear interpolation across gaps no longer than `max_gap_grid` bins.
    Anything longer stays NaN so the learner never trains on invention.
    """
    if ts.size == 0:
        return np.empty(0), {k: np.empty(0) for k in values}

    start = np.floor(ts[0] / grid_s) * grid_s
    stop = np.floor(ts[-1] / grid_s) * grid_s
    grid = np.arange(start, stop + grid_s, grid_s, dtype=float)
    if grid.size == 0:
        return np.empty(0), {k: np.empty(0) for k in values}

    idx = np.clip(((ts - start) / grid_s).astype(int), 0, grid.size - 1)
    out = {}
    counts = np.bincount(idx, minlength=grid.size).astype(float)
    for name, arr in values.items():
        clean = np.nan_to_num(arr, nan=0.0)
        mask = (~np.isnan(arr)).astype(float)
        total = np.bincount(idx, weights=clean, minlength=grid.size)
        n = np.bincount(idx, weights=mask, minlength=grid.size)
        with np.errstate(invalid="ignore", divide="ignore"):
            binned = np.where(n > 0, total / np.maximum(n, 1e-9), np.nan)
        out[name] = _interp_short_gaps(binned, max_gap_grid)
    out["_count"] = counts
    return grid, out


def _interp_short_gaps(arr: np.ndarray, max_gap: int) -> np.ndarray:
    """Linear fill for runs of NaN up to `max_gap` long; leave longer runs alone."""
    a = arr.copy()
    isnan = np.isnan(a)
    if not isnan.any() or isnan.all():
        return a
    valid = np.flatnonzero(~isnan)
    filled = np.interp(np.arange(a.size), valid, a[valid])

    # find NaN runs and only accept the short ones
    edges = np.flatnonzero(np.diff(np.concatenate(([0], isnan.view(np.int8), [0]))))
    for start, stop in zip(edges[::2], edges[1::2]):
        if (stop - start) <= max_gap and start > 0 and stop < a.size:
            a[start:stop] = filled[start:stop]
    return a