feat: server-side time-series downsampling + dashboard/API cleanup

README.md

@@ -154,6 +154,88 @@ See [examples/datatool_dashboard.py](examples/datatool_dashboard.py) for a full
 
 To regenerate the screenshots after UI changes, see [docs/generate_screenshots.py](docs/generate_screenshots.py).
 
+## Server-side downsampling
+
+Large time series are downsampled on the server so the plot only transports the
+points the current zoom level can show. This is driven by `PlotConfig.downsample`:
+
+```python
+from opensemantic.base.view._config import DashboardConfig, PlotConfig, DownsampleConfig
+
+config = DashboardConfig(
+    plot=PlotConfig(
+        downsample=DownsampleConfig(
+            enabled=True,      # downsample when the backend supports it
+            max_points=2000,   # target points per channel
+            method="auto",     # auto | sample | average | minmax
+            edge_anchors=True, # keep the window's first/last real datapoints
+        )
+    )
+)
+```
+
+It only engages on a PostgREST/TimescaleDB backend (the `downsample_tool_channel`
+RPC). On a SQLite/local backend, the RPC being absent, or any error, the read
+silently falls back to the full-resolution path - downsampling never breaks a
+read. The `DataToolView` plot also reloads at a finer resolution when you zoom in.
+
+Strategies (N = number of buckets):
+
+- `sample` (default): one real datapoint nearest each bucket center. Schema
+  agnostic; works for scalar and composite channels. N rows.
+- `average`: structure-preserving deep average per bucket (every numeric leaf
+  averaged, non-numeric keys carried), bucket-center timestamp. N rows.
+- `minmax`: the real min and max datapoint of every numeric sub-characteristic
+  per bucket. Scalar: 2N real rows; composite: the real per-leaf extremes. Best
+  at preserving spikes.
+- `auto`: `minmax` for numeric channels, `sample` for text channels.
+
+`average`/`minmax` skip non-numeric leaves (comments, labels) and fall back to
+`sample` when a channel has no numeric leaf at all.
+
+**Unit normalization caveat:** `average` and `minmax` compare and combine the
+bare stored numbers per leaf, so they are only correct when all stored values of
+a leaf share the same unit. The archive stores base-unit-normalized values, but
+data ingested without normalization (mixed units in one channel) will produce
+wrong `average`/`minmax` results. `sample` returns whole real rows and is
+unaffected.
+
+The RPC lives in pgstack's `postgres/config/optional/100_init_tsdb_schema.sql`
+(uses only core, Apache-2 TimescaleDB; no toolkit dependency). It is created at
+database init; apply it manually (`psql -f` / pgAdmin) on an already-running
+cluster. To measure the speedup, see
+[benchmarks/bench_downsample.py](benchmarks/bench_downsample.py).
+
+[examples/downsample_demo.py](examples/downsample_demo.py) is an interactive
+demo: four channels carry the same 100k-point signal (with narrow spikes), one
+per strategy - the channels are named `raw`, `sample`, `average`, `minmax`.
+Selecting `raw` loads slowly with full detail; `sample`/`average` load fast but
+drop the spikes; `minmax` keeps them. Box-zoom into a flat stretch and click
+"Load current range" to re-fetch that window at finer resolution - the hidden
+spikes reappear on `sample`/`average`. Needs a running pgstack with the RPC
+applied; seed once with `python examples/downsample_demo.py`, then
+`panel serve examples/downsample_demo.py`.
+
+![Downsampling demo](docs/downsample_demo.gif)
+
+![Strategy comparison](docs/screenshot_downsample_raw.png)
+
+*Full window, all four channels: `raw` and `minmax` keep the spikes; `sample`
+and `average` smooth them away at the coarse full-window resolution.*
+
+![Zoomed, before reload](docs/screenshot_downsample_zoom_select.png)
+
+*Box-zoomed around a spike on the `sample` channel: only the coarse full-window
+points are shown, so the spike is still hidden.*
+
+![Zoom reveals the peak](docs/screenshot_downsample_zoom.png)
+
+*After "Load current range": the window is re-fetched at finer buckets and the
+hidden spike reappears. The toolbar reset returns to the full window.*
+
+To regenerate these, see
+[docs/generate_downsample_screenshots.py](docs/generate_downsample_screenshots.py).
+
 ## ProcessObjectView (Process/Object Dashboard UI)
 
 Where `DataToolView` is centered on data tools, `ProcessObjectView` is centered

benchmarks/bench_downsample.py

@@ -0,0 +1,219 @@
+"""Benchmark the server-side downsampling RPC against a live pgstack.
+
+Seeds escalating series sizes for a scalar and a composite channel, then
+times a full-resolution read against each downsampling strategy and reports
+wall time, rows returned and approximate payload size, so the speedup vs
+full-resolution and the relative cost of the deep aggregates are visible.
+
+Gated on the same env vars as the integration tests; prints a skip notice
+otherwise:
+
+    TEST_PGRST_URL, TEST_PGRST_JWT_SECRET  (required)
+    TEST_PGRST_JWT_ROLE, TEST_PGRST_SCHEMA (optional)
+    BENCH_SIZES   comma-separated point counts (default "10000,100000")
+    BENCH_MAXPTS  target points per downsampled read (default 2000)
+    BENCH_REPEAT  timed repeats per case (default 3)
+    TEST_PG_DSN   optional libpq DSN; if psycopg is installed an
+                  EXPLAIN ANALYZE of one RPC call is printed.
+
+Run:  python benchmarks/bench_downsample.py
+"""
+
+import asyncio
+import datetime as dt
+import json
+import os
+import statistics
+import sys
+import time
+from pathlib import Path
+
+# Load tests/.env if present, mirroring tests/conftest.py, so the benchmark
+# can be run with the same configuration as the integration tests.
+_env_path = Path(__file__).resolve().parent.parent / "tests" / ".env"
+if _env_path.exists():
+    try:
+        from dotenv import load_dotenv
+
+        load_dotenv(_env_path)
+    except ImportError:
+        pass
+
+_URL = os.environ.get("TEST_PGRST_URL")
+_SECRET = os.environ.get("TEST_PGRST_JWT_SECRET")
+_ROLE = os.environ.get("TEST_PGRST_JWT_ROLE", "api_user")
+_SCHEMA = os.environ.get("TEST_PGRST_SCHEMA", "api")
+_SIZES = [int(s) for s in os.environ.get("BENCH_SIZES", "10000,100000").split(",")]
+_MAXPTS = int(os.environ.get("BENCH_MAXPTS", "2000"))
+_REPEAT = int(os.environ.get("BENCH_REPEAT", "3"))
+_PG_DSN = os.environ.get("TEST_PG_DSN")
+
+WRITE_CHUNK = 5000
+
+
+def _make_db():
+    import jwt
+    from postgrest import AsyncPostgrestClient
+
+    from opensemantic.base import PostgrestTimeSeriesDatabaseController
+    from opensemantic.core import Label
+
+    token = jwt.encode({"role": _ROLE}, _SECRET, algorithm="HS256")
+    client = AsyncPostgrestClient(
+        base_url=_URL,
+        schema=_SCHEMA,
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    db = PostgrestTimeSeriesDatabaseController(
+        name="bench", label=[Label(text="Bench")], buffered=False
+    )
+    db.set_client(client)
+    return db
+
+
+def _make_controller(db):
+    """A DataToolController with a scalar and a composite channel, bound to db."""
+    from uuid import uuid4
+
+    from opensemantic import compute_scoped_uuid
+    from opensemantic.base.v1 import DataChannel, DataToolController
+    from opensemantic.core.v1 import Label as LabelV1
+
+    parent = uuid4()
+    ctrl = DataToolController(
+        uuid=str(parent),
+        name="Bench",
+        label=[LabelV1(text="Bench")],
+        data_channels=[
+            DataChannel(
+                uuid=str(compute_scoped_uuid(parent, "scalar")),
+                osw_id="placeholder",
+                name="scalar",
+                label=[LabelV1(text="scalar")],
+            ),
+            DataChannel(
+                uuid=str(compute_scoped_uuid(parent, "composite")),
+                osw_id="placeholder",
+                name="composite",
+                label=[LabelV1(text="composite")],
+            ),
+        ],
+    )
+    ctrl.archive_database = db
+    return ctrl
+
+
+async def _timed_read(db, tool_id, ch, start, end, method):
+    from opensemantic.base import DownsampleParams, ReadToolChannelRawParams
+
+    ds = None
+    if method != "raw":
+        ds = DownsampleParams(max_points=_MAXPTS, method=method)
+    times = []
+    rows = []
+    for _ in range(_REPEAT):
+        t0 = time.perf_counter()
+        rows = await db.read_tool_channel_raw(
+            ReadToolChannelRawParams(
+                tool_osw_id=tool_id,
+                channel_osw_id=ch,
+                start=start,
+                end=end,
+                downsample=ds,
+            )
+        )
+        times.append((time.perf_counter() - t0) * 1000.0)
+    payload_kb = len(json.dumps(rows)) / 1024.0
+    return statistics.median(times), len(rows), payload_kb
+
+
+def _explain(tool_id, ch, start, end):
+    if not _PG_DSN:
+        return
+    try:
+        import psycopg
+    except Exception:
+        print("\n(psycopg not installed; skipping EXPLAIN ANALYZE)")
+        return
+    sql = (
+        "EXPLAIN (ANALYZE, BUFFERS) "
+        "SELECT * FROM api.downsample_tool_channel(%s, %s, %s, %s, %s, NULL, 'minmax')"
+    )
+    print("\nEXPLAIN ANALYZE (minmax):")
+    try:
+        with psycopg.connect(_PG_DSN) as conn, conn.cursor() as cur:
+            cur.execute(sql, (tool_id, ch, start, end, _MAXPTS))
+            for (line,) in cur.fetchall():
+                print("  " + line)
+    except Exception as e:
+        print(f"  EXPLAIN failed: {e}")
+
+
+async def _run():
+    from opensemantic.base import DeleteToolParams
+    from opensemantic.base._demo_data import seed_channel_series
+
+    db = _make_db()
+    header = (
+        f"{'size':>9} {'channel':>9} {'method':>8} {'rows':>8} {'ms':>9} {'KB':>10}"
+    )
+    print(header)
+    print("-" * len(header))
+    base = dt.datetime(2023, 1, 1, tzinfo=dt.timezone.utc)
+    for idx, n in enumerate(_SIZES):
+        is_last = idx == len(_SIZES) - 1
+        ctrl = _make_controller(db)
+        tool_id = ctrl.get_osw_id()
+        ch_scalar = ctrl.get_channel_by_name("scalar").get_osw_id().split("#")[-1]
+        ch_comp = ctrl.get_channel_by_name("composite").get_osw_id().split("#")[-1]
+
+        def _value(channel, i, _n=n):
+            if channel.name == "composite":
+                return {
+                    "temperature": {"value": float(i)},
+                    "humidity": {"value": float(_n - i)},
+                }
+            return {"value": float(i)}
+
+        try:
+            await seed_channel_series(
+                ctrl, n_points=n, base_ts=base, value_fn=_value, chunk_size=WRITE_CHUNK
+            )
+            await asyncio.sleep(1.0)  # let PostgREST settle after the writes
+            start, end = base, base + dt.timedelta(seconds=n - 1)
+            for label, ch in (("scalar", ch_scalar), ("composite", ch_comp)):
+                for method in ("raw", "sample", "average", "minmax"):
+                    ms, rows, kb = await _timed_read(
+                        db, tool_id, ch, start, end, method
+                    )
+                    print(
+                        f"{n:>9} {label:>9} {method:>8} {rows:>8} "
+                        f"{ms:>9.1f} {kb:>10.1f}"
+                    )
+            if is_last:
+                # EXPLAIN while the tool still exists (largest size).
+                _explain(tool_id, ch_scalar, start, end)
+        finally:
+            try:
+                await db.delete_tool(DeleteToolParams(tool_osw_id=tool_id))
+            except Exception:
+                pass
+
+
+def main():
+    if not (_URL and _SECRET):
+        print(
+            "Skipping benchmark: set TEST_PGRST_URL and TEST_PGRST_JWT_SECRET "
+            "to run against a live pgstack."
+        )
+        return 0
+    try:
+        asyncio.run(_run())
+    except Exception as e:
+        print(f"Benchmark failed (is pgstack up and the RPC applied?): {e}")
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())