From f3349958650ec1537a82de956e38e87037b0267d Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Wed, 2 Sep 2026 01:18:46 +0200 Subject: [PATCH] feat(webui): sporadic-trigger capture, CSV export and UI rework Brings the Go hub and web SPA work developed on feature/udpscope onto main, without the udpscope client itself. The trigger engine could not capture a sporadic event: it armed on the live tail only, so a burst shorter than one push window was already past by the time the FSM looked for it. It now searches the ring history for the crossing, which also makes a capture reproducible from the same data rather than dependent on push timing (wshub/trigger.go, ringbuf.go, history.go). Adds CSV/JSON export of the visible window (wshub/export.go) and reworks the SPA: per-signal axis controls, a readable trigger panel, and a fix for the flicker caused by repainting on every push instead of on a frame tick (static/app.js, index.html, style.css). BUFFER_AND_TRIGGER.md documents the ring/decimation/trigger interaction, which is otherwise only inferable from the three files that implement it. Co-Authored-By: Claude Opus 4.6 --- Client/udpstreamer/BUFFER_AND_TRIGGER.md | 425 +++++++++++ Client/udpstreamer/go.mod | 12 +- Client/udpstreamer/go.sum | 34 + Client/udpstreamer/main.go | 1 + Client/udpstreamer/static/app.js | 710 +++++++++++++++--- Client/udpstreamer/static/index.html | 6 +- Client/udpstreamer/static/style.css | 8 + Common/Client/go/go.mod | 16 +- Common/Client/go/go.sum | 21 + .../go/wshub/capture_shortwindow_test.go | 75 ++ Common/Client/go/wshub/export.go | 119 +++ Common/Client/go/wshub/export_test.go | 87 +++ Common/Client/go/wshub/history.go | 43 +- Common/Client/go/wshub/ringbuf.go | 15 +- Common/Client/go/wshub/ringbuf_test.go | 9 +- Common/Client/go/wshub/trigger.go | 132 +++- .../Client/go/wshub/trigger_sporadic_test.go | 367 +++++++++ Common/Client/go/wshub/trigger_test.go | 64 +- 18 files changed, 1980 insertions(+), 164 deletions(-) create mode 100644 Client/udpstreamer/BUFFER_AND_TRIGGER.md create mode 100644 Common/Client/go/wshub/capture_shortwindow_test.go create mode 100644 Common/Client/go/wshub/export.go create mode 100644 Common/Client/go/wshub/export_test.go create mode 100644 Common/Client/go/wshub/trigger_sporadic_test.go diff --git a/Client/udpstreamer/BUFFER_AND_TRIGGER.md b/Client/udpstreamer/BUFFER_AND_TRIGGER.md new file mode 100644 index 0000000..772570c --- /dev/null +++ b/Client/udpstreamer/BUFFER_AND_TRIGGER.md @@ -0,0 +1,425 @@ +# Buffer time-window & trigger logic in `Client/udpstreamer` + +How the UDP Scope client acquires, buffers, times and triggers waveforms. +The pipeline has two halves that must be read together: + +- the **Go hub** (`Common/Client/go/wshub/`) — owns the UDP sockets, the + full-resolution sample storage, the disk history, and the trigger FSM; +- the **browser SPA** (`static/app.js`) — owns the display buffers, the rolling + window, and the trigger capture rendering. + +The same SPA is also served by `Client/webui` and talks to the C++ StreamHub, +which mirrors the Go hub's behaviour (same trigger FSM states, same binary +frames). Everything below describes the Go-hub path; the wire contracts are +identical on both. + +--- + +## 1. End-to-end data flow + +``` +MARTe2 RT app ──UDP/UDPS──▶ sources.go: runSession() + │ CONFIG + DATA packets, 17-byte header, HRT timestamp per frame + ▼ +udpsprotocol.ParseData() → []DataSample{HRTTimestamp, WallTime, Values} + ▼ +Hub.Run() dataCh → pending[sourceID] (drained every 30 Hz tick) + ▼ +buildBinaryDataMessageForSource() + ├─ rebuild per-sample timestamps from TimeMode / calibration / monotonic snap + ├─ h.ingest(key, n, t, v) ← FULL rate: ring.write + hist.write + trigger.feed + └─ minMaxDecimate(…, maxPushPoints=50) → WS binary v1 frame to clients + ▼ +browser: onBinaryData() → pushBuffer() into per-signal circular buffers + ▼ +renderDirtyPlots() (rAF loop) → buildUPlotData() → uPlot +``` + +The 30 Hz push is the **only** live path to the browser and it is decimated to +≤50 points/signal/tick. Everything that needs full resolution — zoom, trigger +captures, disk history — is fed independently through `ingest()` and never goes +over the wire until asked. + +--- + +## 2. Hub-side buffers: `sigRing` (`wshub/ringbuf.go`) + +One ring per `"sourceId:signalName"` key. A fixed-capacity circular buffer of +Float64 `(t, v)` pairs with a `sync.RWMutex` (writes from `Hub.Run()`, reads +from HTTP/WS handler goroutines). + +### 2.1 Min/max bucketing + +`bucket` is how many source samples collapse into **one min/max pair** on the +way in: + +- `bucket == 1` — the stream is stored verbatim; +- `bucket > 1` — each group contributes its minimum and its maximum, emitted in + time order (`flushBucketLocked`), so the stored timestamps stay + non-decreasing (reads binary-search `rb.t`). + +Bucketing is what lets an arbitrarily long window fit a fixed per-signal memory +budget at a megasample rate. Samples already stored keep the resolution they +were written at; the ring converges on a new bucket as it rolls +(`setBucket`). + +### 2.2 Source-rate measurement + +`sigRing` keeps its own source-sample accounting (`srcCount`, `srcT0`, `srcT1`, +reset every `srcRateWindowSec = 10 s`), because once `bucket > 1` neither `size` +nor the stored timespan measures the real incoming rate. `sourceRate()` is used +by the tuning sweep and by the history writer. + +### 2.3 Ring tuning (`retuneRings`, every 1 s) + +`activeWindowSec()` decides how far back the rings must reach: + +1. an **armed trigger** owns the window: `cfg.windowSec + captureLagSec` + (`captureLagSec = captureMarginSec + 1/30 ≈ 0.183 s` — the capture is read + out a post-window + margin + one push tick after the trigger, so the rings + must hold that much extra or the front of the capture has already rolled); +2. otherwise the **widest window any connected client is displaying** + (`wsClient.displayWindowSec`, set by the SPA's `setWindow` command), with a + `defaultLiveWindowSec = 10 s` fallback while nobody has said; + +Then per ring, with `budget = ringBudget()` (default `defaultRingPts = 10 M`, +floor `ringCapInitial = 250 k`): + +- grow to the budget first (`grow()` preserves all samples, never shrinks); +- compute the needed bucket with `ringBucketFor(rate, window, capacity)` + (`ceil(2·rate·window·ringHeadroom / capacity)`, `ringHeadroom = 1.25`); +- apply it with **hysteresis**: keep the current bucket while its coverage is + between `need` and `2·need`, so a rate jittering across the boundary does not + flip the resolution every second. + +The history archive is re-sized from the same window (`hist.setWindow`) so a +zoom or capture that outlives the rings can fall back to it. + +### 2.4 Reading: `slice(t0, t1)` + +Binary search for `t0` then `t1` over the circular layout, returning copies of +the pairs in `[t0, t1]`. Safe to use without holding the lock. + +--- + +## 3. Disk history (`wshub/history.go`) + +Optional (`EnableHistory`, `CloseHistory`), enabled by the hub configuration. +Every sample goes to disk through `ingest → hist.write` at full rate, in files +sized for the *current* window (not a retention period). It exists to back +three things the rings cannot: + +- **zoom past the window**: `readRange(key, t0, t1, maxOut)`; +- **captures the rings have rolled past**: `captureRange(trigTime−pre, trigTime+post)` + lifts each capture into a file of its own so nothing overwrites it before the + next trigger; +- **short captures**: `backfillCaptureHead` prepends the front of the window the + ring no longer holds (the ring only *becomes* as long as the window after a + re-tune; the archive was written straight through). + +--- + +## 4. Live push to the browser + +`Hub.Run()` drains `pending[sourceID]` on a 30 Hz ticker. Even with no client +connected the frame is built: that is what keeps feeding rings, history and the +trigger, and keeps push cursors advancing so a late client does not get a +backlog burst. + +`buildBinaryDataMessageForSource` reconstructs per-sample timestamps per signal +`TimeMode`: + +| Mode | Timestamp reconstruction | +|---|---| +| `FirstSample` / `LastSample` | scalar TimeSignal value × `timerToSec` (µs→s or ns→s for u64), calibrated once against `WallTime`; samples spaced by `1/SamplingRate` | +| `FullArray` | per-element TimeSignal array, calibrated once against `WallTime` | +| scalar (`n == 1`) | `WallTime` of the UDP arrival | +| `PacketTime` (default, n>1) | inter-packet wall-clock gaps divided by n (single-packet ticks use the gap from the previous tick) | + +**Monotonic snapping** (optional, `setMonotonic` command / "Sync TS" checkbox): +when enabled, the inter-frame anchor gap is smoothed with an EMA +(`monotonicEMAAlpha = 0.01`, initialised from the nominal `n·dt`) and small +deviations (< `monotonicTolerance = 5 ms`) are snapped to the smoothed gap, +removing the software-dispatch jitter overlaps/gaps described in the StreamHub +docs while tracking the true hardware rate (no accumulated drift). + +The live frame is a **binary v1** WS message: + +``` +[u8 1][u8 srcIdLen][srcId][u32 nSigs] +{[u16 keyLen][key][u32 N][f64 t×N][f64 v×N]} +``` + +with each signal min/max-decimated to `maxPushPoints = 50` (`minMaxDecimate`: +the range is split into `threshold/2` buckets, each contributing its min and max +in time order — a scope-style envelope that keeps glitches on screen). + +--- + +## 5. Browser-side buffers (`static/app.js`) + +### 5.1 Capacity & growth + +- `MAX_CAP = 2 000 000` — hard ceiling per buffer (~32 MB/signal at Float64 t+v); +- `DEFAULT_CAP = 100 000` — starting size for scalars; +- `TEMPORAL_CAP = 500 000` — starting size for array signals (the hub pushes + ≤50 pts/signal/tick, so this already covers ~5 min); +- `growBufferForWindow(buf, windowSec)` — **sizes from the buffer's own span**, + not the signal's sampling rate: the incoming rate here is the hub-decimated + ~1.5 kpts/s regardless of the source rate, so rate-based sizing overshot by + three orders of magnitude. Grows only when the buffer is full, to + `windowSec × 1.5` headroom, capped at `MAX_CAP`. +- `growBuffer` copies all existing samples into a larger array (preserving + circular order). + +### 5.2 The window + +`windowSec` (default 5 s, options 1 s … 10 min) is the rolling viewport. +Changing it: + +1. updates `windowSec`; +2. `sendWindow()` → WS `setWindow` → hub `displayWindowSec` → ring re-tune; +3. grows every local buffer via `growBufferForWindow`; +4. evicts the decimation cache (a different window invalidates all cached + renderings). + +The rolling "now" anchor is **data-driven, not wall-clock**: +`computePlotNow(p)` takes the newest timestamp of each contributing source and +uses the min-of-max over sources that are still active (a source lagging the +fastest by more than `windowSec` is treated as stale and excluded). This keeps +the window tracking real data regardless of clock skew between hub and browser. + +### 5.3 Slicing & rendering + +- `getBufferSliceRange(buf, t0, t1)` — binary search on the circular layout, + O(log n + window size); +- `getBufferSliceRangeWithBrackets` — same plus one point on each side so lines + still cross a nearly-empty zoom window; +- `supplementWithBrackets` — same bracketing for sparse server-fetched zoom data. + +`buildLiveData(p)`: + +1. slices every trace in `[t0, t1]`; +2. picks the **master** signal: highest `SamplingRate`, then most points; +3. decimates the master to ~2× plot width (`DECIM_MIN = 200` floor) via a + background worker (`decimateAsync`, stale-while-revalidate cache keyed per + plot/range/data-generation); +4. resamples every other trace onto the master grid with `resampleLinear`; +5. normalises Y (`applyVScaleNorm`: calibration `v·scale+offset`, then + `(y − offset)/div`). + +### 5.4 Zoom + +A zoom pins `p.xRange` and asks the hub for hi-res data over the exact range +(WS `zoom` request or HTTP `/api/zoom`). The hub answers from the full-res +rings — or from the **held copy of the last trigger capture** (`captureHold` +double buffer) while that window is still relevant — decimated to the requested +point budget. The browser prefers the fetched data when it exists, falls back +to its own circular buffers otherwise, and always brackets with local points. + +--- + +## 6. Hub-side trigger FSM (`wshub/trigger.go`) + +### 6.1 States and configuration + +``` +idle ──arm──▶ armed ──edge──▶ collecting ──window elapsed──▶ triggered + ▲ ▲ (pre/post latched) │ + │ └───────────── rearm (normal mode, after holdoff) ◀─────────┘ + └────────────── disarm / single mode stays triggered +``` + +Configuration (`trigConfig`, client-settable via WS `setTrigger`): + +| field | meaning | clamp | +|---|---|---| +| `signalKey` | `"src:sig"` or `"src:sig[i]"` | — | +| `edge` | `rising` / `falling` / `both` | — | +| `threshold` | **raw** units (SPA converts calibrated → raw) | — | +| `windowSec` | capture window | `[1e-4, 600]` | +| `prePercent` | pre-trigger share | `[0, 100]` | +| `mode` | `normal` (auto-rearm) / `single` | — | +| `holdoffSec` | re-arm delay after a capture, double-trigger guard | `[0, 60]` | + +### 6.2 Edge detection (`feed`) + +Called from `ingest` with every full-resolution batch for the trigger signal. +Level tracking (`prevValue`/`prevValid`) compares consecutive samples against +the threshold; `[i]`-suffixed keys stride the flattened batch by `nElem` to +watch one column. On a qualifying edge in `armed` state: `latchWindowLocked` +freezes `trigTime` and the pre/post split (so later config edits cannot move a +capture's axis). + +### 6.3 Buffer-fill gate + +Before accepting an edge, the FSM checks that the trigger signal's ring reaches +back far enough that the capture will come back whole (`fillLocked`): + +``` +need = windowSec − growth × postSec, floored at the pre-window +``` + +`growth` is the measured span-growth rate of the ring (`setBuffered`, refreshed +by `refreshTriggerFill` from the tick and from trigger commands). A still-filling +ring grows 1 s of span per second, so the gate reduces to the pre-window; a +full ring at a long window needs the whole window. While holding off, the level +is still tracked so the first edge after the gate opens is measured against the +right predecessor. The SPA shows the hold-off as an armed trigger with a +`bufferFill %` badge. + +### 6.4 Window timing and the pending edge + +`dueCapture` waits for the window on the **sample clock**, not the wall clock: +`lastT ≥ trigTime + post + captureMarginSec(0.15)`. This avoids cutting a +capture short when the stream's timestamps lag real time. Three ways it fires: + +1. the samples themselves covered the window; +2. wall-clock fallback when no sample was ever seen (Force from idle); +3. `captureStallSec = 2 s` of stream silence — deliver what was collected + rather than leaving the client stuck in "collecting". + +While a capture is in flight the comparator keeps running. The **first** +qualifying edge at/after `notBefore = trigTime + max(post, holdoffSec)` is +remembered (`pendingT`/`pendingValid`) and fired immediately on the automatic +`rearm()`. Without this the trigger was deaf through the whole post-window + +holdoff, which rounded sparse pulse trains up to whole periods (a 1 Hz train at +a 1 s window was caught at 0.5 Hz). + +### 6.5 Holdoff and rearm + +`markTriggered` moves `collecting → triggered` and, in `normal` mode (not +stopped), schedules `rearmAt = now + cfg.holdoffSec`. `dueRearm` consumes it; +`rearm()` re-arms immediately on a pending edge or returns to `armed`. The +holdoff is measured from the trigger point, overlapping the post-window rather +than adding to it. + +`Force()` fires immediately at the most recent sample time (wall clock if no +sample yet) — the "Force" button. + +### 6.6 Capture assembly (`buildTriggerCapture`) + +On a due capture the hub builds the **binary v2** frame: + +``` +[u8 2][f64 trigTime][f64 preSec][f64 postSec][u32 nSig] +{[u16 keyLen][fullKey][u32 N][f64 t×N][f64 v×N]} +``` + +For every ring: + +1. `slice(trigTime−pre, trigTime+post)`; +2. `backfillCaptureHead` from disk history for the front the ring lost; +3. if it is still short by more than `shortCaptureTol = 1 %` of the window, + log it explicitly (nothing can recover data the ring never held); +4. keep the **full-resolution** slice in the `captureHold` double buffer + (so a zoom into the capture can be answered after the rings roll past); +5. min/max-decimate to `trigCapturePts = 20 000` per signal for the wire — + a 60 s window at 1 MSps is ~960 MB raw per signal and would be dropped by + the send path anyway. + +The double buffer is published (`capture.publish`) only once the frame is known +good, so a shot that yielded nothing leaves the previous capture on screen. +Dropped frames (client send-queue full) are logged. + +`triggerTick` (every push tick) drives the whole FSM: re-tune rings → open +pending history files → refresh the fill measurement → due capture (send + mark +triggered + `hist.captureRange`) or due rearm → broadcast state only when it +changed (`stateUnsent`). + +### 6.7 WS commands + +| message | effect | +|---|---| +| `setTrigger {signal, edge, threshold, windowSec, prePercent, mode, holdoffSec}` | replace config | +| `arm` / `rearm` | explicit arm (discards pending edge) | +| `disarm` | → idle | +| `trigStop {stopped}` | pause/resume auto-rearm | +| `forceTrigger` | fire now | + +Every command also refreshes the buffer-fill measurement synchronously — at +1 MSps the ring crosses the fill threshold many times inside one 33 ms tick, so +waiting for the next tick would fire on a stale measurement. + +--- + +## 7. Browser-side trigger (`static/app.js`) + +### 7.1 State handling + +`onTriggerState(msg)` tracks the FSM broadcast: + +- **armed** — shows `bufferFill %` while the hub is holding off on the fill + gate, so a trigger that is not yet fireable does not look broken; +- **collecting** — clears the previous snapshot, latches `trigTime` (and + `preSec`/`postSec` if the hub sent them), and lets live data sweep into the + trigger axis (see 7.3); +- **triggered / idle** — bookkeeping for the Rearm/Stop buttons. + +### 7.2 Capture handling + +`onTriggerCapture` parses the v2 frame into `trig.snapshot[key] = {t, v}` plus +the latched `_preS`/`_postS`. It is **ignored when the client did not enable +the trigger** (`trig.enabled`), because the hub keeps an armed trigger across +client sessions and applying a foreign capture would clobber this client's zoom +and scales. On receipt: the horizontal zoom is dropped so the whole capture is +visible, but **vertical scales (V/div, offset) persist** — they are user +settings and must survive from shot to shot. + +### 7.3 Rendering modes (`buildUPlotData`) + +| state | renderer | source | +|---|---|---| +| collecting, no snapshot yet | `buildTrigFillData` | live buffers, drawn on the *final* trigger axis (relative seconds, `[-pre, +post]`) so the trace sweeps in from the left | +| armed, not fired | freeze last frame | — | +| snapshot present | `buildTrigData` | the capture (or a hi-res zoom reply that covers ≥98 % of the view, else the snapshot) | +| otherwise | `buildLiveData` | rolling window | + +`buildTrigData` converts to trigger-relative time (`t − trigT`), picks the +master by rate/count, decimates (cached per range+source-tag), resamples the +other traces, and normalises Y. + +### 7.4 Threshold in calibrated units + +The trigger threshold is held in **calibrated units** (what the user sees on +the Y axis). `sendTrigConfig()` inverts it through the signal's calibration +before sending: `raw = (calibrated − offset)/scale`, so the hub's raw +comparator fires exactly when `GAIN·signal + OFFSET` crosses the threshold. +The threshold line (`drawTriggerMarker`) maps the same calibrated threshold +through the signal's vscale: `y_norm = (threshold − offset)/div`. + +--- + +## 8. Key constants + +| constant | value | file | +|---|---|---| +| push rate | 30 Hz | `hub.go` | +| `maxPushPoints` (live) | 50 pts/signal/tick | `hub.go` | +| `trigCapturePts` (capture) | 20 000 pts/signal | `trigger.go` | +| `captureMarginSec` | 0.15 s | `trigger.go` | +| `captureStallSec` | 2.0 s | `trigger.go` | +| `autoRearmDelaySec` (default holdoff) | 0.2 s | `trigger.go` | +| `maxTriggerWindowSec` | 600 s | `trigger.go` | +| `ringBudget` default | 10 000 000 pts/signal | `hub.go` | +| `ringCapInitial` | 250 000 | `hub.go` | +| `ringCapScalar` | 100 000 | `hub.go` | +| `ringHeadroom` | 1.25 | `ringbuf.go` | +| `defaultLiveWindowSec` | 10 s | `ringbuf.go` | +| `captureLagSec` | 0.15 + 1/30 ≈ 0.183 s | `ringbuf.go` | +| `monotonicTolerance` | 5 ms | `hub.go` | +| `monotonicEMAAlpha` | 0.01 | `hub.go` | +| `MAX_CAP` (browser) | 2 000 000 pts/signal | `app.js` | +| `DEFAULT_CAP` / `TEMPORAL_CAP` | 100 000 / 500 000 | `app.js` | +| `DECIM_MIN` | 200 | `app.js` | + +--- + +## 9. The C++ StreamHub mirror + +The Go hub and the C++ StreamHub implement the same WS contracts and must stay +in sync (`AGENTS.md`): same `triggerState` FSM strings, same v2 capture frame, +same command set (`setTrigger` including `holdoffSec`, `arm`, `disarm`, +`trigStop`, `forceTrigger`), same `trigCapturePts`/`kTrigCapturePts` cap, and +the same ring/history windowing intent (`Source/Applications/StreamHub/`). A +protocol change on one side must be mirrored on the other. diff --git a/Client/udpstreamer/go.mod b/Client/udpstreamer/go.mod index 75041b4..399fb51 100644 --- a/Client/udpstreamer/go.mod +++ b/Client/udpstreamer/go.mod @@ -1,12 +1,22 @@ module udpstreamer-webui -go 1.21 +go 1.24.9 require marte2/common v0.0.0 require ( + github.com/andybalholm/brotli v1.1.1 // indirect + github.com/google/uuid v1.6.0 // indirect github.com/gorilla/websocket v1.5.1 // indirect + github.com/klauspost/compress v1.17.9 // indirect + github.com/parquet-go/bitpack v1.0.0 // indirect + github.com/parquet-go/jsonlite v1.0.0 // indirect + github.com/parquet-go/parquet-go v0.32.0 // indirect + github.com/pierrec/lz4/v4 v4.1.21 // indirect + github.com/twpayne/go-geom v1.6.1 // indirect golang.org/x/net v0.17.0 // indirect + golang.org/x/sys v0.38.0 // indirect + google.golang.org/protobuf v1.34.2 // indirect ) replace marte2/common => ../../Common/Client/go diff --git a/Client/udpstreamer/go.sum b/Client/udpstreamer/go.sum index 272772f..64e19c7 100644 --- a/Client/udpstreamer/go.sum +++ b/Client/udpstreamer/go.sum @@ -1,4 +1,38 @@ +github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU= +github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU= +github.com/alecthomas/assert/v2 v2.10.0 h1:jjRCHsj6hBJhkmhznrCzoNpbA3zqy0fYiUcYZP/GkPY= +github.com/alecthomas/assert/v2 v2.10.0/go.mod h1:Bze95FyfUr7x34QZrjL+XP+0qgp/zg8yS+TtBj1WA3k= +github.com/alecthomas/repr v0.4.0 h1:GhI2A8MACjfegCPVq9f1FLvIBS+DrQ2KQBFZP1iFzXc= +github.com/alecthomas/repr v0.4.0/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4= +github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA= +github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA= +github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY= github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY= +github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM= +github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg= +github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= +github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= +github.com/parquet-go/bitpack v1.0.0 h1:AUqzlKzPPXf2bCdjfj4sTeacrUwsT7NlcYDMUQxPcQA= +github.com/parquet-go/bitpack v1.0.0/go.mod h1:XnVk9TH+O40eOOmvpAVZ7K2ocQFrQwysLMnc6M/8lgs= +github.com/parquet-go/jsonlite v1.0.0 h1:87QNdi56wOfsE5bdgas0vRzHPxfJgzrXGml1zZdd7VU= +github.com/parquet-go/jsonlite v1.0.0/go.mod h1:nDjpkpL4EOtqs6NQugUsi0Rleq9sW/OtC1NnZEnxzF0= +github.com/parquet-go/parquet-go v0.32.0 h1:NWDqTUHfrCS4cJP/Fj2HlxvqsrVedWG3sayMkf+znzM= +github.com/parquet-go/parquet-go v0.32.0/go.mod h1:navtkAYr2LGoJVp141oXPlO/sxLvaOe3la2JEoD8+rg= +github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ= +github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= +github.com/twpayne/go-geom v1.6.1 h1:iLE+Opv0Ihm/ABIcvQFGIiFBXd76oBIar9drAwHFhR4= +github.com/twpayne/go-geom v1.6.1/go.mod h1:Kr+Nly6BswFsKM5sd31YaoWS5PeDDH2NftJTK7Gd028= +github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= +github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM= golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE= +golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= +golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= +google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= diff --git a/Client/udpstreamer/main.go b/Client/udpstreamer/main.go index 4eb9fc2..936b6cc 100644 --- a/Client/udpstreamer/main.go +++ b/Client/udpstreamer/main.go @@ -94,6 +94,7 @@ func main() { http.Handle("/", http.FileServer(http.FS(sub))) http.HandleFunc("/ws", hub.HandleWebSocket) http.HandleFunc("/api/zoom", hub.HandleZoom) + http.HandleFunc("/api/export", hub.HandleExport) http.HandleFunc("/version", func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, buildVersion) }) diff --git a/Client/udpstreamer/static/app.js b/Client/udpstreamer/static/app.js index 778cd68..6cc3028 100644 --- a/Client/udpstreamer/static/app.js +++ b/Client/udpstreamer/static/app.js @@ -476,14 +476,24 @@ let cursorsDirty = false; // if true, redraw all plots to update cursor lines // Rolling-window anchor used to keep cursors visually fixed while live data scrolls. let _cursorAnchorNow = null; -// Horizontal value rulers — stored in normalized division units (the shared -// y scale, -4.5…4.5) so one pair applies to every plot regardless of V/div. -const rulers = { mode: 'off', yA: null, yB: null }; +// Horizontal value rulers. The on/off toggle is global, but each plot keeps its +// own pair of normalized-division positions (rulerState), so dragging Y1 in +// one plot does not move it in the others. +const rulers = { mode: 'off', plotId: null }; +const rulerState = {}; // plotId → { yA, yB } -// Layout — [label, cssClass, cols, rows] +function getRulerState(plotId) { + if (!rulerState[plotId]) rulerState[plotId] = { yA: null, yB: null }; + return rulerState[plotId]; +} + +// Layout — [label, cssClass, cols, rows, (optional) plotCount]. +// Custom (non-uniform) layouts carry an explicit plotCount; the grid template +// and the spanning cells are defined in style.css under #plot-grid.. const LAYOUTS = [ ['1×1', 'l1x1', 1, 1], ['1×2', 'l1x2', 1, 2], ['2×1', 'l2x1', 2, 1], ['1×3', 'l1x3', 1, 3], ['3×1', 'l3x1', 3, 1], ['2×2', 'l2x2', 2, 2], ['1×4', 'l1x4', 1, 4], ['4×1', 'l4x1', 4, 1], + ['1+2', 'l1p2', 2, 2, 3], // one plot spanning the top row, two below ]; let currentLayout = 'l1x1'; let colFrs = [1]; // fractional column sizes (sum = cols) @@ -816,6 +826,7 @@ function onConfig(msg) { } buildSidebar(); buildTrigSignalSelect(); + maybeRestoreViewLate(); } /* ════════════════════════════════════════════════════════════════ @@ -925,12 +936,21 @@ function wsSend(obj) { function sendWindow() { wsSend({ type: 'setWindow', seconds: windowSec }); } -// trig.threshold is held in calibrated units. The hub's comparator runs on raw +// trig.threshold is held in calibrated units. The hub.s comparator runs on raw // samples, so invert on the way out: raw = (calibrated - offset) / scale. function sendTrigConfig() { const cal = trig.signal ? calForKey(trig.signal) : Calib.IDENTITY; + // A negative calibration gain flips the signal on screen (v_cal = v_raw·scale + // + offset with scale < 0), so a calibrated rising edge is a raw FALLING + // edge. The hub compares raw samples, so send the raw direction that matches + // the edge the user picked on the calibrated trace. + let edge = trig.edge; + if (cal.scale < 0) { + if (edge === 'rising') edge = 'falling'; + else if (edge === 'falling') edge = 'rising'; + } wsSend({ - type: 'setTrigger', signal: trig.signal, edge: trig.edge, + type: 'setTrigger', signal: trig.signal, edge: edge, threshold: Calib.invertCal(trig.threshold, cal), windowSec: trig.windowSec, prePercent: trig.prePercent, mode: trig.mode, holdoffSec: trig.holdoffSec, }); @@ -1332,7 +1352,11 @@ function decimateAsync(cacheKey, t, v, threshold, gen) { return result; } } - return cached || null; // stale entry, or nothing to draw yet + // Never hand out a stale decimation: drawing it (at its old timestamps) and + // then the fresh one a frame later is what makes the trace jump/shimmer on + // every push. Return null instead — the caller holds the previous render + // until the worker's fresh result lands (it flags the plot for redraw). + return null; } // Evict stale decimation cache entries for a plot (call when zoom range changes). @@ -1838,15 +1862,9 @@ function drawCursorLines(u, p) { if (vNorm === null) return; const cy = u.valToPos(vNorm, 'y', true); if (cy < bbox.top || cy > bbox.top + bbox.height) return; - // Un-transform normalized value back to real units for display - // y_norm = (y_raw - offset) / divValue → y_raw = y_norm * divValue + offset - const vs = sigVScale[vsKeyFor(p.id, key)]; - let vReal = vNorm; - if (vs) { - const dv = vs._resolvedDiv || vs.divValue || 1; - const ofs = vs._resolvedOffset != null ? vs._resolvedOffset : (vs.offset || 0); - vReal = vNorm * dv + ofs; - } + // Calibrated value at the cursor time, from the raw source (matches + // the hover and the cursor readouts in every display mode). + const vReal = calibratedValueAt(key, val); const tc = getSigStyle(key).color; // Diamond marker at intersection ctx.fillStyle = tc; @@ -1860,7 +1878,7 @@ function drawCursorLines(u, p) { ctx.closePath(); ctx.fill(); // Value text next to diamond (real units) - const str = Math.abs(vReal) >= 10000 ? vReal.toExponential(2) : parseFloat(vReal.toPrecision(4)).toString(); + const str = vReal === null ? '—' : (Math.abs(vReal) >= 10000 ? vReal.toExponential(2) : parseFloat(vReal.toPrecision(4)).toString()); ctx.fillStyle = tc; ctx.font = '11px monospace'; const currentAlign = ctx.textAlign; @@ -1898,6 +1916,8 @@ function rulerRawValue(p, yNorm) { // Draw the horizontal value rulers (called from the draw hook). function drawRulerLines(u, p) { if (rulers.mode !== 'on') return; + const rs = rulerState[p.id]; + if (!rs) return; const { ctx, bbox } = u; if (!bbox) return; @@ -1926,8 +1946,8 @@ function drawRulerLines(u, p) { ctx.restore(); }; - drawLine(rulers.yA, 'rgba(166,227,161,0.85)', 'Y1'); - drawLine(rulers.yB, 'rgba(243,139,168,0.85)', 'Y2'); + drawLine(rs.yA, 'rgba(166,227,161,0.85)', 'Y1'); + drawLine(rs.yB, 'rgba(243,139,168,0.85)', 'Y2'); } // Compute the rolling-window anchor ("newest common timestamp") for a plot. @@ -2133,8 +2153,9 @@ function createUPlot(p) { const rect = p.uplot.over.getBoundingClientRect(); const { min, max } = p.uplot.scales.y; const toY = val => rect.top + (1 - (val - min) / (max - min)) * rect.height; - if (rulers.yA !== null && Math.abs(clientY - toY(rulers.yA)) <= CURSOR_SNAP_PX) return 'A'; - if (rulers.yB !== null && Math.abs(clientY - toY(rulers.yB)) <= CURSOR_SNAP_PX) return 'B'; + const rs = rulerState[p.id]; + if (rs && rs.yA !== null && Math.abs(clientY - toY(rs.yA)) <= CURSOR_SNAP_PX) return 'A'; + if (rs && rs.yB !== null && Math.abs(clientY - toY(rs.yB)) <= CURSOR_SNAP_PX) return 'B'; return null; } @@ -2171,8 +2192,10 @@ function createUPlot(p) { // Set cursor position immediately on mousedown if (yTarget) { - if (yTarget === 'A') rulers.yA = _rulerValFromEvent(e); - else rulers.yB = _rulerValFromEvent(e); + rulers.plotId = p.id; // the readout follows the plot whose rulers moved + const rs = getRulerState(p.id); + if (yTarget === 'A') rs.yA = _rulerValFromEvent(e); + else rs.yB = _rulerValFromEvent(e); } else if (target === 'A') cursors.tA = _cursorValFromEvent(e); else cursors.tB = _cursorValFromEvent(e); updateCursorReadout(); @@ -2180,8 +2203,9 @@ function createUPlot(p) { const onMove = ev => { if (yTarget) { - if (yTarget === 'A') rulers.yA = _rulerValFromEvent(ev); - else rulers.yB = _rulerValFromEvent(ev); + const rs = getRulerState(p.id); + if (yTarget === 'A') rs.yA = _rulerValFromEvent(ev); + else rs.yB = _rulerValFromEvent(ev); } else if (target === 'A') cursors.tA = _cursorValFromEvent(ev); else cursors.tB = _cursorValFromEvent(ev); updateCursorReadout(); @@ -2447,8 +2471,12 @@ function buildLiveData(p) { let dec; if (cached) { dec = cached; + } else if (p.uplot && p.uplot.data && p.uplot.data[0] && p.uplot.data[0].length) { + // Fresh decimation not ready yet — hold the previous render so the trace + // does not flicker between a stale decimation and the fresh one. + return p.uplot.data; } else { - // Worker job submitted — sync fallback this frame so the plot isn't blank. + // First render: worker job submitted, nothing on screen yet — sync. dec = decimate(masterRaw.t, masterRaw.v, targetPts); } sharedT = dec.t; @@ -2521,7 +2549,14 @@ function buildTrigData(p) { // same-length snapshot slice for the same range, so it is tagged separately. const cacheKey = `${p.id}:${masterKey}:${t0.toFixed(6)}:${t1.toFixed(6)}:${masterRaw.t.length}:${usedFetched ? 'hi' : 'snap'}`; const cachedDec = decimateAsync(cacheKey, masterRaw.t, masterRaw.v, targetPts); - const dec = cachedDec || decimate(masterRaw.t, masterRaw.v, targetPts); + let dec; + if (cachedDec) { + dec = cachedDec; + } else if (p.uplot && p.uplot.data && p.uplot.data[0] && p.uplot.data[0].length) { + return p.uplot.data; // hold the previous render until the fresh decimation lands + } else { + dec = decimate(masterRaw.t, masterRaw.v, targetPts); + } // Convert absolute → relative seconds const sharedT = new Float64Array(dec.t.length); for (let i = 0; i < dec.t.length; i++) sharedT[i] = dec.t[i] - trigT; @@ -2578,8 +2613,15 @@ function buildTrigFillData(p) { masterV = masterRaw.v; } else { const cacheKey = `${p.id}:${masterKey}:trigfill`; - const dec = decimateAsync(cacheKey, masterRaw.t, masterRaw.v, targetPts, _dataGen) || - decimate(masterRaw.t, masterRaw.v, targetPts); + const decd = decimateAsync(cacheKey, masterRaw.t, masterRaw.v, targetPts, _dataGen); + let dec; + if (decd) { + dec = decd; + } else if (p.uplot && p.uplot.data && p.uplot.data[0] && p.uplot.data[0].length) { + return p.uplot.data; // hold until the fresh decimation is ready + } else { + dec = decimate(masterRaw.t, masterRaw.v, targetPts); + } sharedAbsT = dec.t; masterV = dec.v; } @@ -2753,7 +2795,19 @@ function updateCursorBtnVisibility() { under one — so a zoom, a pan or a new capture can leave them outside the viewport entirely, with no way to get them back: they are dragged by grabbing their line, and an off-screen line cannot be grabbed. */ +function resetRulers() { + // Re-place every plot's rulers at the default ±2 divisions, like + // resetCursors re-places the vertical cursors. + plots.forEach(p => { + const rs = getRulerState(p.id); + rs.yA = -2; rs.yB = 2; + }); + updateCursorReadout(); + cursorsDirty = true; +} + function resetCursors() { + resetRulers(); const refPlot = plots.find(p => p.uplot); if (!refPlot) return; const { min, max } = refPlot.uplot.scales.x; @@ -2790,9 +2844,13 @@ document.getElementById('btn-ruler').addEventListener('click', () => { rulers.mode = rulers.mode === 'off' ? 'on' : 'off'; const btn = document.getElementById('btn-ruler'); btn.classList.toggle('active', rulers.mode === 'on'); - if (rulers.mode === 'on' && rulers.yA === null && rulers.yB === null) { - // Auto-place at ±2 divisions from the centre on first use. - rulers.yA = -2; rulers.yB = 2; + if (rulers.mode === 'on') { + // Auto-place every plot at ±2 divisions from the centre on first use; + // afterwards each plot keeps its own positions. + plots.forEach(pl => { + const rs = getRulerState(pl.id); + if (rs.yA === null && rs.yB === null) { rs.yA = -2; rs.yB = 2; } + }); } updateCursorReadout(); cursorsDirty = true; @@ -2809,16 +2867,9 @@ function getValueAtCursor(p, t) { if (!p.uplot || t === null) return null; const key = plotActiveSignal[p.id] || (p.traces.length === 1 ? p.traces[0] : null); if (!key) return null; - const idx = p.traces.indexOf(key); - if (idx < 0) return null; - const vNorm = interpAtTime(p.uplot, idx + 1, t); - if (vNorm === null) return null; - // Un-normalize: y_norm = (y_raw - offset) / divValue - const vs = sigVScale[p.id + ':' + key]; - if (!vs) return vNorm; - const dv = vs._resolvedDiv != null ? vs._resolvedDiv : (vs.divValue || 1); - const ofs = vs._resolvedOffset != null ? vs._resolvedOffset : (vs.offset || 0); - return vNorm * dv + ofs; + // Interpolate the raw wire value and apply the calibration explicitly, so + // cursor readouts match the hover in every display mode. + return calibratedValueAt(key, t); } // Update per-plot cursor value readouts (A, B, ΔV) for all plots. @@ -2851,6 +2902,64 @@ function rawFromNorm(p, key, vNorm) { const ofs = vs._resolvedOffset != null ? vs._resolvedOffset : (vs.offset || 0); return vNorm * dv + ofs; } +// Linear interpolation of a sorted (t, v) pair at absolute time tAbs. Returns +// null outside the data's range — never fabricated, so an export or readout +// cannot invent samples the signal never had. +function interpSortedRaw(t, v, tAbs) { + if (!t || t.length === 0) return null; + if (tAbs < t[0] || tAbs > t[t.length - 1]) return null; + let lo = 0, hi = t.length - 1; + while (lo < hi) { const m = (lo + hi) >> 1; if (t[m] < tAbs) lo = m + 1; else hi = m; } + if (lo === 0) return v[0] ?? null; + const t0 = t[lo - 1], t1 = t[lo]; + const v0 = v[lo - 1], v1 = v[lo]; + if (v0 == null || v1 == null) return v0 ?? v1 ?? null; + return v0 + (tAbs - t0) / (t1 - t0) * (v1 - v0); +} + +// Binary-search linear interpolation of a circular buffer at time t. +function interpCircular(buf, t) { + if (!buf || buf.size === 0) return null; + const { cap, size, head } = buf; + const start = (size === cap) ? head : 0; + const physAt = k => (start + k) % cap; + let lo = 0, hi = size; + while (lo < hi) { const m = (lo + hi) >> 1; if (buf.t[physAt(m)] < t) lo = m + 1; else hi = m; } + if (lo === 0) return buf.v[physAt(0)] ?? null; + if (lo >= size) return buf.v[physAt(size - 1)] ?? null; + const t0 = buf.t[physAt(lo - 1)], t1 = buf.t[physAt(lo)]; + const v0 = buf.v[physAt(lo - 1)], v1 = buf.v[physAt(lo)]; + if (v0 == null || v1 == null) return v0 ?? v1 ?? null; + return v0 + (t - t0) / (t1 - t0) * (v1 - v0); +} + +// Raw (uncalibrated) value of `key` at absolute time tAbs, from the best +// available raw source: trigger snapshot → fetched zoom data → live push +// buffer. All three store wire values, so calibration is applied here, at the +// point of display, exactly once. +function rawAtAbsTime(key, tAbs) { + if (trig.snapshot) { + const s = trig.snapshot[key]; + if (s && s.t.length) { const v = interpSortedRaw(s.t, s.v, tAbs); if (v != null) return v; } + } + for (const p of plots) { + const zd = zoomData[p.id]; + if (!zd) continue; + const s = zd.signals[key]; + if (s && s.t.length) { const v = interpSortedRaw(s.t, s.v, tAbs); if (v != null) return v; } + } + const buf = buffers[key]; + if (buf && buf.size) { const v = interpCircular(buf, tAbs); if (v != null) return v; } + return null; +} + +// Calibrated value of `key` at axis time t. Under a trigger the axis is +// relative to the trigger instant, so convert to absolute first. +function calibratedValueAt(key, t) { + const tAbs = (inTrigWindow() && trig.trigTime != null) ? trig.trigTime + t : t; + const raw = rawAtAbsTime(key, tAbs); + return raw === null ? null : Calib.applyCal(raw, calForKey(key)); +} function hideHoverReadout() { document.getElementById('hover-readout').style.display = 'none'; @@ -2870,13 +2979,14 @@ function showHoverReadout(p, e) { const tStr = inTrigWindow() ? fmtDuration(t, span, true) : fmtLiveTime(t, span); let html = '
' + escHtml(tStr) + '
'; p.traces.forEach((key, idx) => { - const vNorm = interpAtTime(p.uplot, idx + 1, t); const name = key.includes(':') ? key.slice(key.indexOf(':') + 1) : key; - // rawFromNorm inverts the vscale transform, which Task 7 made operate on - // calibrated values — so this is already in calibrated units. const unit = unitForKey(key); - const val = vNorm === null ? '—' - : (_fmtVal(rawFromNorm(p, key, vNorm)) + (unit ? ' ' + unit : '')); + // Interpolate the raw wire value and apply the calibration explicitly, + // so the hover is correct in every display mode (analog, digital, + // mixed) and independent of the vscale state. + const vCal = calibratedValueAt(key, t); + const val = vCal === null ? '—' + : (_fmtVal(vCal) + (unit ? ' ' + unit : '')); html += '
' + '' + escHtml(name) + '' + @@ -2894,17 +3004,25 @@ function showHoverReadout(p, e) { el.style.top = Math.max(4, y) + 'px'; } -// Update the Y1/Y2/ΔY ruler readout, expressed in the raw units of the first -// plot that has an active (or sole) signal. +// Update the Y1/Y2/ΔY ruler readout, expressed in the raw units of the plot +// whose rulers were last moved, falling back to the first plot with a signal. function updateRulerReadout() { const box = document.getElementById('ruler-readout'); const on = rulers.mode === 'on'; box.style.display = on ? '' : 'none'; if (!on) return; - const ref = plots.find(p => p.uplot && p.traces.length > 0 && - rulerRawValue(p, 0) !== null); - const conv = y => (y === null || !ref) ? null : rulerRawValue(ref, y); - const vA = conv(rulers.yA), vB = conv(rulers.yB); + let ref = null; + if (rulers.plotId !== null) { + const pl = plots.find(p => p.id === rulers.plotId); + if (pl && pl.uplot && pl.traces.length > 0) ref = pl; + } + if (!ref) { + ref = plots.find(p => p.uplot && p.traces.length > 0 && + rulerRawValue(p, 0) !== null) || null; + } + const rs = ref ? rulerState[ref.id] : null; + const conv = y => (y === null || !ref || !rs) ? null : rulerRawValue(ref, y); + const vA = conv(rs ? rs.yA : null), vB = conv(rs ? rs.yB : null); document.getElementById('cur-y1').textContent = 'Y1: ' + fmtVal(vA); document.getElementById('cur-y2').textContent = 'Y2: ' + fmtVal(vB); document.getElementById('cur-dy').textContent = @@ -3355,23 +3473,42 @@ function initPlotCfgBar(plotId, p) { /* ════════════════════════════════════════════════════════════════ Layout management ════════════════════════════════════════════════════════════════ */ -// Returns the number of plot cells in a layout (cols × rows). +// Returns the number of plot cells in a layout. Custom layouts carry an +// explicit plotCount; uniform ones are cols × rows. function layoutPlotCount(cls) { + const entry = LAYOUTS.find(l => l[1] === cls); + if (entry) { + if (entry.length >= 5) return entry[4]; + return entry[2] * entry[3]; + } const m = cls.match(/^l(\d+)x(\d+)$/); return m ? parseInt(m[1]) * parseInt(m[2]) : 1; } -// Build a small SVG grid thumbnail for a given cols×rows layout. -function layoutSVG(cols, rows) { +// Build a small SVG grid thumbnail for a layout entry. Custom (non-uniform) +// layouts draw their own cell arrangement. +function layoutSVG(entry) { const W = 28, H = 20, GAP = 1.5, PAD = 1.5; - const cw = (W - PAD * 2 - GAP * (cols - 1)) / cols; - const ch = (H - PAD * 2 - GAP * (rows - 1)) / rows; let rects = ''; - for (let r = 0; r < rows; r++) { - for (let c = 0; c < cols; c++) { - const x = (PAD + c * (cw + GAP)).toFixed(1); - const y = (PAD + r * (ch + GAP)).toFixed(1); - rects += ``; + if (entry[1] === 'l1p2') { + // 1+2: one full-width cell on top, two side by side below. + const cw = (W - PAD * 2 - GAP) / 2; + const ch = (H - PAD * 2 - GAP) / 2; + const y2 = (PAD + ch + GAP).toFixed(1); + const x2 = (PAD + cw + GAP).toFixed(1); + rects += ``; + rects += ``; + rects += ``; + } else { + const [, , cols, rows] = entry; + const cw = (W - PAD * 2 - GAP * (cols - 1)) / cols; + const ch = (H - PAD * 2 - GAP * (rows - 1)) / rows; + for (let r = 0; r < rows; r++) { + for (let c = 0; c < cols; c++) { + const x = (PAD + c * (cw + GAP)).toFixed(1); + const y = (PAD + r * (ch + GAP)).toFixed(1); + rects += ``; + } } } return `` @@ -3401,7 +3538,7 @@ function applyLayout(cls) { // Update button label const btn = document.getElementById('btn-layout'); - if (btn) btn.innerHTML = layoutSVG(cols, rows) + ' ' + label + ' ▾'; + if (btn) btn.innerHTML = layoutSVG(entry) + ' ' + label + ' ▾'; // Update active state in menu document.querySelectorAll('.layout-menu-item') @@ -3436,11 +3573,12 @@ function applyLayout(cls) { function buildLayoutMenu() { const menu = document.getElementById('layout-menu'); - LAYOUTS.forEach(([label, cls, cols, rows]) => { + LAYOUTS.forEach(entry => { + const [label, cls] = entry; const item = document.createElement('button'); item.className = 'layout-menu-item' + (cls === currentLayout ? ' active' : ''); item.dataset.layout = cls; - item.innerHTML = layoutSVG(cols, rows) + '' + label + ''; + item.innerHTML = layoutSVG(entry) + '' + label + ''; item.addEventListener('click', () => { applyLayout(cls); menu.classList.remove('open'); @@ -3467,9 +3605,20 @@ function buildLayoutMenu() { /* ════════════════════════════════════════════════════════════════ Export CSV (all plots) — fetches full-resolution data from ring ════════════════════════════════════════════════════════════════ */ +// Shared busy state for the export dropdown: prevents re-entry and shows +// progress on the selector while a (possibly large) export runs. +let exportBusy = false; +function setExportBusy(busy) { + exportBusy = busy; + const sel = document.getElementById('export-select'); + if (!sel) return; + sel.disabled = busy; + const ph = sel.querySelector('option[value=""]'); + if (ph) ph.textContent = busy ? '\u23f3 Exporting\u2026' : '\u23ea Export'; +} + async function exportAllCSV() { - const btn = document.getElementById('btn-csv-all'); - if (btn.disabled) return; + if (exportBusy) return; const inTrigMode = trig.enabled && trig.snapshot !== null; @@ -3482,8 +3631,8 @@ async function exportAllCSV() { let t0, t1, relOffset = 0; if (inTrigMode) { // Export the full trigger window around the trigger event. - t0 = trig.trigTime - trigPreSec(); - t1 = trig.trigTime + trigPostSec(); + t0 = trig.trigTime - activePreSec(); + t1 = trig.trigTime + activePostSec(); relOffset = trig.trigTime; } else { // Use the current zoom range if active, else the rolling window. @@ -3504,60 +3653,52 @@ async function exportAllCSV() { t1 = plotNow; } } + if (!(t1 > t0)) return; - // Show loading state. - const origLabel = btn.textContent; - btn.textContent = '⏳ Downloading…'; - btn.disabled = true; + exportBusy = true; + // Cap the export. A full window at a megasample rate is hundreds of MB raw + // (the old exact-timestamp merge exploded into millions of rows and crashed + // the tab); ask the hub for a min/max-decimated envelope — the same scope + // style reduction the live view uses — and cap the number of rows. + const BUDGET = 100000; // max rows per signal + setExportBusy(true); - // Fetch full-resolution ring data (n=0 → no decimation). let ringSignals = null; - try { - ringSignals = await wsZoomRequest(t0, t1, 0, keys); - } catch (e) { - console.warn('CSV export: ring fetch failed, falling back to push buffer', e); - } finally { - btn.textContent = origLabel; - btn.disabled = false; + if (!inTrigMode) { + try { + ringSignals = await wsZoomRequest(t0, t1, BUDGET, keys); + } catch (e) { + console.warn('CSV export: ring fetch failed, falling back to local data', e); + } } + setExportBusy(false); - // Build per-signal time/value arrays. - // Priority: ring buffer (full res) → trigger snapshot → push buffer. + // Per-signal raw source: hub ring (whole window, decimated) → trigger + // snapshot (already \u226420k pts) → local push buffer. const slices = keys.map(key => { - const rd = ringSignals && ringSignals[key]; - if (rd && rd.t && rd.t.length > 0) { - const t = rd.t, v = rd.v; - if (inTrigMode) { - return { t: Array.from(t).map(ts => ts - relOffset), v: Array.from(v) }; - } - return { t: Array.from(t), v: Array.from(v) }; + if (!inTrigMode) { + const rd = ringSignals && ringSignals[key]; + if (rd && rd.t && rd.t.length > 0) return { key, t: rd.t, v: rd.v }; } - // Fallback: push buffer or trigger snapshot. if (inTrigMode) { const raw = trig.snapshot[key] || { t: new Float64Array(0), v: new Float64Array(0) }; - return { t: Array.from(raw.t).map(ts => ts - relOffset), v: Array.from(raw.v) }; + return { key, t: raw.t, v: raw.v }; } - const buf = buffers[key]; if (!buf) return { t: [], v: [] }; - const sl = getBufferSliceRange(buf, t0, t1); - return { t: Array.from(sl.t), v: Array.from(sl.v) }; + const buf = buffers[key]; + const sl = buf ? getBufferSliceRange(buf, t0, t1) : { t: new Float64Array(0), v: new Float64Array(0) }; + return { key, t: sl.t, v: sl.v }; }); + const present = slices.filter(s => s.t.length > 0); + if (!present.length) return; - // Merge all timestamps and build aligned rows. - const allT = new Set(); - slices.forEach(s => s.t.forEach(t => allT.add(t))); - const sortedT = Array.from(allT).sort((a, b) => a - b); - if (!sortedT.length) return; + // Master time grid = the signal with the most samples; every other signal is + // resampled onto it (linear, no extrapolation). Cells outside a signal's own + // span stay empty rather than being fabricated, so continuous signals export + // without holes and no value is invented. + let master = present[0]; + present.forEach(s => { if (s.t.length > master.t.length) master = s; }); - const lookups = slices.map(s => { - const m = new Map(); - s.t.forEach((t, i) => m.set(t, s.v[i])); - return m; - }); - - // Strip "sourceId:" prefix from column headers for readability, and append - // the effective unit. These values come straight from the ring/history/ - // snapshot and never pass through applyVScaleNorm, so calibrate them here. - const cals = keys.map(k => calForKey(k)); + const cals = new Map(keys.map(k => [k, calForKey(k)])); const displayKeys = keys.map(k => { const name = k.includes(':') ? k.split(':').slice(1).join(':') : k; const u = unitForKey(k); @@ -3566,10 +3707,21 @@ async function exportAllCSV() { }); const timeCol = '"' + (inTrigMode ? 'time_rel_s' : 'time_s') + '"'; const hdr = [timeCol, ...displayKeys].join(','); - const rows = sortedT.map(t => - [t.toFixed(9), ...lookups.map((lk, i) => - lk.has(t) ? Calib.applyCal(lk.get(t), cals[i]) : '')].join(',') - ); + + const rows = new Array(master.t.length); + for (let i = 0; i < master.t.length; i++) { + const tAbs = master.t[i]; + const cells = present.map(s => { + if (s === master) { + return Calib.applyCal(master.v[i], cals.get(s.key)); + } + const v = interpSortedRaw(s.t, s.v, tAbs); + return v === null ? '' : Calib.applyCal(v, cals.get(s.key)); + }); + const tt = inTrigMode ? tAbs - relOffset : tAbs; + rows[i] = [tt.toFixed(9), ...cells].join(','); + } + const blob = new Blob([hdr + '\n' + rows.join('\n')], { type: 'text/csv' }); const a = document.createElement('a'); a.href = URL.createObjectURL(blob); @@ -3754,6 +3906,10 @@ function deletePlot(plotId) { let _dbgTick = 0; let _dataGen = 0; // incremented each time new data arrives function renderDirtyPlots() { + // Schedule the next frame FIRST: an exception below must never kill the + // animation loop, or every plot would freeze until a page refresh. + requestAnimationFrame(renderDirtyPlots); + try { // Compute global "now" once — shared by all rolling-window plots this frame. const globalPlotNow = getGlobalNow(); @@ -3823,7 +3979,7 @@ function renderDirtyPlots() { plots.forEach(p => { if (!p.needsRedraw || !p.uplot || p.traces.length === 0) return; - + try { const inTrigModeNow = inTrigWindow(); // The x tick formatter and the cursor-sync group are baked into the uPlot // options at construction. A plot built in live mode therefore keeps @@ -3838,7 +3994,10 @@ function renderDirtyPlots() { if (isRolling && _dataGen === p.lastDataGen && p.uplot.data && p.uplot.data[0] && p.uplot.data[0].length > 0) { p.needsRedraw = false; zoomGuard = true; - p.uplot.setScale('x', { min: globalPlotNow - windowSec, max: globalPlotNow }); + // Use the same per-plot anchor as the rebuild path, so the rolling window + // does not jump when the frame switches between the two. + const plotNow = computePlotNow(p); + p.uplot.setScale('x', { min: plotNow - windowSec, max: plotNow }); zoomGuard = false; return; } @@ -3875,12 +4034,25 @@ function renderDirtyPlots() { p.uplot.setScale('x', { min: plotNow - windowSec, max: plotNow }); } zoomGuard = false; + p._errCount = 0; + } catch (e) { + // One bad plot must not kill the whole render loop. Track consecutive + // failures and self-heal by rebuilding the uPlot instance. + p._errCount = (p._errCount || 0) + 1; + console.error(`[render] plot ${p.id}:`, e); + p.needsRedraw = true; // retry next frame + if (p._errCount >= 30) { + p._errCount = 0; + try { createUPlot(p); } catch (e2) { console.error(`[render] rebuild plot ${p.id}:`, e2); } + } + } }); // Keep per-plot cursor value readouts in sync with live data. if (cursors.mode === 'on') updatePlotCursorReadouts(); - - requestAnimationFrame(renderDirtyPlots); + } catch (e) { + console.error('[render]', e); + } } @@ -3958,6 +4130,7 @@ function onSources(msg) { }); buildSidebar(); if (statsOpen) _refreshStatsSelector(); + maybeRestoreViewLate(); } function addSourceWS(label, addr, multicastGroup, dataPort) { @@ -4582,7 +4755,303 @@ initSignalMenu(); const cb = document.getElementById('cb-monotonic'); if (cb) cb.checked = localStorage.getItem('udpscope.monotonic') === '1'; } -document.getElementById('btn-csv-all').addEventListener('click', exportAllCSV); +// Export every stored sample of the plotted signals as a Parquet file, served +// by the Go hub's /api/export. Full resolution (no decimation) and hole-free +// (each signal keeps its own timestamps — long format). The file can be huge +// (hundreds of MB at high rates), so stream it to disk when the File System +// Access API is available instead of holding it in a Blob. +async function exportParquet() { + if (exportBusy) return; + + const inTrigMode = trig.enabled && trig.snapshot !== null; + const keys = []; + plots.forEach(p => p.traces.forEach(k => { if (!keys.includes(k)) keys.push(k); })); + if (!keys.length) return; + + // Same range resolution as the CSV export. + let t0, t1; + if (inTrigMode) { + t0 = trig.trigTime - activePreSec(); + t1 = trig.trigTime + activePostSec(); + } else { + const refPlot = plots.find(p => p.xRange); + if (refPlot) { + [t0, t1] = refPlot.xRange; + } else { + let plotNow = -Infinity; + keys.forEach(k => { + const buf = buffers[k]; + if (buf && buf.size > 0) { + const t = buf.t[(buf.head - 1 + buf.cap) % buf.cap]; + if (t > plotNow) plotNow = t; + } + }); + if (!isFinite(plotNow)) plotNow = Date.now() / 1000; + t0 = plotNow - windowSec; + t1 = plotNow; + } + } + if (!(t1 > t0)) return; + + exportBusy = true; + setExportBusy(true); + try { + const url = '/api/export?t0=' + t0.toFixed(9) + '&t1=' + t1.toFixed(9) + + '&signals=' + encodeURIComponent(keys.join(',')); + const resp = await fetch(url); + if (!resp.ok) { + alert('Parquet export failed (HTTP ' + resp.status + ').\n\n' + + 'The /api/export endpoint is provided by the Go hub; the C++ ' + + 'StreamHub does not serve it.'); + return; + } + const filename = 'signals_' + Date.now() + '.parquet'; + if (window.showSaveFilePicker && resp.body) { + try { + const handle = await window.showSaveFilePicker({ + suggestedName: filename, + types: [{ description: 'Parquet', accept: { 'application/vnd.apache.parquet': ['.parquet'] } }], + }); + const writable = await handle.createWritable(); + await resp.body.pipeTo(writable); + return; + } catch (e) { + if (e && e.name === 'AbortError') return; // user cancelled the picker + console.warn('parquet export: file picker failed, falling back to Blob', e); + } + } + const blob = await resp.blob(); + const a = document.createElement('a'); + a.href = URL.createObjectURL(blob); + a.download = filename; + a.click(); + URL.revokeObjectURL(a.href); + } catch (e) { + console.warn('parquet export failed', e); + alert('Parquet export failed: ' + e.message); + } finally { + setExportBusy(false); + } +} + +// Export dropdown: dispatch on selection, then reset to the placeholder so the +// same format can be chosen again. +document.getElementById('export-select').addEventListener('change', () => { + const sel = document.getElementById('export-select'); + const fmt = sel.value; + sel.value = ''; + if (fmt === 'csv') exportAllCSV(); + else if (fmt === 'parquet') exportParquet(); +}); + +/* ════════════════════════════════════════════════════════════════ + View-state persistence (cookie) + ════════════════════════════════════════════════════════════════ */ +// The whole client view — layout, plots (traces/titles/modes), window, trigger +// configuration, rulers, sources — is serialised into one cookie so a reload +// restores the previous view. Cookies are size-limited, so the state degrades +// gracefully (rulers → trigger → sources → traces) when it would not fit. +const VIEW_COOKIE = 'udpscope.view'; +const VIEW_COOKIE_MAX = 3500; // encoded chars; browsers cap cookies at ~4 KiB + +function packViewState() { + const state = { + v: 1, + windowSec: windowSec, + layout: currentLayout, + plots: plots.map(p => ({ + title: p.title, + mode: p.mode, + traces: p.traces.map(k => { + const colon = k.indexOf(':'); + const name = colon >= 0 ? k.slice(colon + 1) : k; + return { key: k, label: srcLabelForKey(k), name }; + }), + })), + trig: { + enabled: trig.enabled, signal: trig.signal, edge: trig.edge, + threshold: trig.threshold, windowSec: trig.windowSec, + prePercent: trig.prePercent, mode: trig.mode, holdoffSec: trig.holdoffSec, + }, + rulers: { + mode: rulers.mode, + plotId: plots.findIndex(p => p.id === rulers.plotId), + states: plots.map(p => { + const rs = rulerState[p.id]; + return rs ? { yA: rs.yA, yB: rs.yB } : { yA: null, yB: null }; + }), + }, + sources: Object.values(sourcesMap).map(s => ({ + label: s.label || s.addr || s.id, addr: s.addr, + })), + }; + let s = JSON.stringify(state); + const tooBig = () => encodeURIComponent(s).length > VIEW_COOKIE_MAX; + if (tooBig()) { delete state.rulers; s = JSON.stringify(state); } + if (tooBig()) { delete state.trig; s = JSON.stringify(state); } + if (tooBig()) { delete state.sources; s = JSON.stringify(state); } + if (tooBig()) { + state.plots = state.plots.map(p => ({ title: p.title, mode: p.mode })); + s = JSON.stringify(state); + } + if (tooBig()) { state.plots = []; s = JSON.stringify(state); } + return s; +} + +// Saves are gated until the saved view has been re-applied (phase 2) or the +// grace timeout fires: otherwise the very first periodic save would overwrite +// the cookie with the not-yet-restored (empty) state and destroy it. +let _viewSaveReady = false; +function saveViewState() { + if (!_viewSaveReady) return; + try { + const s = packViewState(); + document.cookie = VIEW_COOKIE + '=' + encodeURIComponent(s) + + '; path=/; max-age=31536000; SameSite=Lax'; + } catch (e) { + console.warn('view cookie save failed', e); + } +} + +function readViewState() { + try { + const prefix = VIEW_COOKIE + '='; + const m = document.cookie.split('; ').find(c => c.startsWith(prefix)); + if (!m) return null; + const st = JSON.parse(decodeURIComponent(m.slice(prefix.length))); + return (st && st.v === 1) ? st : null; + } catch (e) { + return null; + } +} + +// Phase 1 (init): layout, plot cards (titles/modes), window, rulers. Traces and +// the trigger need sources/signals loaded, so they are applied in phase 2. +function restoreViewState() { + const st = readViewState(); + if (!st) return; + + if (st.layout && LAYOUTS.some(l => l[1] === st.layout)) applyLayout(st.layout); + + const plotState = st.plots || []; + plotState.forEach((ps, i) => { + const p = plots[i]; + if (!p) return; + if (ps.title && ps.title !== 'Plot ' + p.id) { + p.title = ps.title; + const tEl = document.getElementById('ptitle-' + p.id); + if (tEl) tEl.textContent = ps.title; + const inp = document.querySelector('#pcfg-' + p.id + ' .pcfg-title-input'); + if (inp) inp.value = ps.title; + } + if (ps.mode && ps.mode !== p.mode) { + p.mode = ps.mode; + document.querySelectorAll('#pcfg-' + p.id + ' .pcfg-mode-btn') + .forEach(b => b.classList.toggle('active', b.dataset.mode === ps.mode)); + } + }); + + if (st.windowSec != null) { + windowSec = st.windowSec; + const sel = document.getElementById('window-select'); + if (sel && [...sel.options].some(o => o.value === String(st.windowSec))) { + sel.value = String(st.windowSec); + } + } + + if (st.rulers) { + rulers.mode = st.rulers.mode === 'on' ? 'on' : 'off'; + const btn = document.getElementById('btn-ruler'); + if (btn) btn.classList.toggle('active', rulers.mode === 'on'); + (st.rulers.states || []).forEach((rs, i) => { + const p = plots[i]; + if (!p || !rs) return; + const cur = getRulerState(p.id); + cur.yA = rs.yA; cur.yB = rs.yB; + }); + if (st.rulers.plotId != null && plots[st.rulers.plotId]) { + rulers.plotId = plots[st.rulers.plotId].id; + } + } +} + +// Rebuild a saved trace key against the current sources: source ids change +// across restarts, so match by label and fall back to the saved id-key. +function restoreTraceKey(entry) { + const src = Object.values(sourcesMap).find(s => (s.label || s.id) === entry.label); + return src ? (src.id + ':' + entry.name) : entry.key; +} + +// Phase 2 (first sources + signals): reconcile sources, re-apply traces, then +// the trigger configuration and the window to the hub. +let _viewLateRestored = false; +function maybeRestoreViewLate() { + if (_viewLateRestored) return; + const st = readViewState(); + if (!st) { _viewLateRestored = true; return; } + + // Add any saved sources the hub does not already have (it persists its own). + const known = Object.values(sourcesMap).map(s => (s.label || s.addr || s.id) + '\u0000' + s.addr); + let added = false; + (st.sources || []).forEach(sv => { + const k = (sv.label || sv.addr) + '\u0000' + (sv.addr || ''); + if (!known.includes(k)) { addSourceWS(sv.label, sv.addr, sv.multicastGroup, sv.dataPort); added = true; } + }); + if (added) return; // re-enter when the new sources appear + + // Traces and the trigger selector need at least one source with signals. + if (!Object.values(sourcesMap).some(s => (s.signals || []).length > 0)) return; + _viewLateRestored = true; + _viewSaveReady = true; + + (st.plots || []).forEach((ps, i) => { + const p = plots[i]; + if (!p) return; + (ps.traces || []).forEach(t => addTraceTo(p.id, restoreTraceKey(t))); + }); + + if (st.windowSec != null) { + windowSec = st.windowSec; + sendWindow(); + } + + const t = st.trig; + if (t) { + trig.edge = t.edge || trig.edge; + if (t.threshold != null) trig.threshold = t.threshold; + if (t.windowSec != null) trig.windowSec = t.windowSec; + if (t.prePercent != null) trig.prePercent = t.prePercent; + trig.mode = t.mode || trig.mode; + if (t.holdoffSec != null) trig.holdoffSec = t.holdoffSec; + trig.signal = t.signal || ''; + const el = id => document.getElementById(id); + if (el('trig-edge')) el('trig-edge').value = trig.edge; + if (el('trig-window')) el('trig-window').value = String(trig.windowSec); + if (el('trig-mode')) el('trig-mode').value = trig.mode; + if (el('trig-holdoff')) el('trig-holdoff').value = trig.holdoffSec; + if (el('trig-pre')) el('trig-pre').value = String(trig.prePercent); + if (el('trig-pre-val')) el('trig-pre-val').textContent = trig.prePercent + '%'; + refreshTrigThresholdField(); + const selSig = document.getElementById('trig-signal'); + if (selSig && trig.signal) { + const base = trig.signal.replace(/\[\d+\]$/, ''); + if ([...selSig.options].some(o => o.value === base)) selSig.value = base; + } + if (t.enabled) openTrigBar(true); + else updateTrigStatusBadge('idle'); + } +} + +// Periodic save keeps the cookie current without wiring every control; the +// pagehide save captures the final state on close/reload. +setInterval(saveViewState, 3000); +window.addEventListener('pagehide', saveViewState); +// Hub down / never connected: stop gating after 10 s so layout, window and +// rulers still persist even though traces and the trigger could not be +// restored. +setTimeout(() => { _viewSaveReady = true; }, 10000); + + document.getElementById('history-badge').addEventListener('click', toggleHistoryPanel); document.getElementById('btn-hist-cancel').addEventListener('click', toggleHistoryPanel); document.getElementById('btn-hist-apply').addEventListener('click', applyHistoryBudget); @@ -4592,6 +5061,7 @@ document.getElementById('stats-source-sel').addEventListener('change', e => { statsSelectedSrc = e.target.value || null; renderStats(); }); +restoreViewState(); resolveHub().then(connectWS); requestAnimationFrame(renderDirtyPlots); fetch('/version').then(r => r.text()).then(v => { diff --git a/Client/udpstreamer/static/index.html b/Client/udpstreamer/static/index.html index bdce18e..6d0c300 100644 --- a/Client/udpstreamer/static/index.html +++ b/Client/udpstreamer/static/index.html @@ -41,7 +41,11 @@ - + diff --git a/Client/udpstreamer/static/style.css b/Client/udpstreamer/static/style.css index b84a045..427d558 100644 --- a/Client/udpstreamer/static/style.css +++ b/Client/udpstreamer/static/style.css @@ -14,6 +14,11 @@ *, *::before, *::after { box-sizing: border-box; margin: 0; padding: 0; } html, body { height:100%; background:var(--bg); color:var(--text); font-family:'Segoe UI',system-ui,sans-serif; font-size:14px; overflow:hidden; } +/* Uniform 0.9x compaction — scales every element (fonts, bars, plots, + spacing) while reflowing layout. `zoom` (Chrome/Edge/Safari, Firefox 126+) + is preferred over `transform: scale` because it reflows, so fixed-position + bars and JS-computed offsets stay aligned. */ +html { zoom: 0.9; } ::-webkit-scrollbar { width:6px; } ::-webkit-scrollbar-track { background:var(--mantle); } ::-webkit-scrollbar-thumb { background:var(--surface1); border-radius:3px; } @@ -256,6 +261,9 @@ input[type=range].trig-range::-webkit-slider-thumb { #plot-grid.l2x3 { grid-template-columns:1fr 1fr; grid-template-rows:1fr 1fr 1fr; } #plot-grid.l1x4 { grid-template-columns:1fr; grid-template-rows:1fr 1fr 1fr 1fr; } #plot-grid.l4x1 { grid-template-columns:1fr 1fr 1fr 1fr; grid-template-rows:1fr; } +/* 1+2 layout: one plot spanning the top row, two side by side below. */ +#plot-grid.l1p2 { grid-template-columns:1fr 1fr; grid-template-rows:1fr 1fr; } +#plot-grid.l1p2 .plot-card:first-child { grid-column: 1 / -1; } /* ── Plot card ────────────────────────────────────────────────── */ .plot-card { diff --git a/Common/Client/go/go.mod b/Common/Client/go/go.mod index 25ddedf..7484841 100644 --- a/Common/Client/go/go.mod +++ b/Common/Client/go/go.mod @@ -1,7 +1,19 @@ module marte2/common -go 1.21 +go 1.24.9 require github.com/gorilla/websocket v1.5.1 -require golang.org/x/net v0.17.0 // indirect +require ( + github.com/andybalholm/brotli v1.1.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/klauspost/compress v1.17.9 // indirect + github.com/parquet-go/bitpack v1.0.0 // indirect + github.com/parquet-go/jsonlite v1.0.0 // indirect + github.com/parquet-go/parquet-go v0.32.0 // indirect + github.com/pierrec/lz4/v4 v4.1.21 // indirect + github.com/twpayne/go-geom v1.6.1 // indirect + golang.org/x/net v0.17.0 // indirect + golang.org/x/sys v0.38.0 // indirect + google.golang.org/protobuf v1.34.2 // indirect +) diff --git a/Common/Client/go/go.sum b/Common/Client/go/go.sum index 272772f..8c30f05 100644 --- a/Common/Client/go/go.sum +++ b/Common/Client/go/go.sum @@ -1,4 +1,25 @@ +github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA= +github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY= github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY= +github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= +github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= +github.com/parquet-go/bitpack v1.0.0 h1:AUqzlKzPPXf2bCdjfj4sTeacrUwsT7NlcYDMUQxPcQA= +github.com/parquet-go/bitpack v1.0.0/go.mod h1:XnVk9TH+O40eOOmvpAVZ7K2ocQFrQwysLMnc6M/8lgs= +github.com/parquet-go/jsonlite v1.0.0 h1:87QNdi56wOfsE5bdgas0vRzHPxfJgzrXGml1zZdd7VU= +github.com/parquet-go/jsonlite v1.0.0/go.mod h1:nDjpkpL4EOtqs6NQugUsi0Rleq9sW/OtC1NnZEnxzF0= +github.com/parquet-go/parquet-go v0.32.0 h1:NWDqTUHfrCS4cJP/Fj2HlxvqsrVedWG3sayMkf+znzM= +github.com/parquet-go/parquet-go v0.32.0/go.mod h1:navtkAYr2LGoJVp141oXPlO/sxLvaOe3la2JEoD8+rg= +github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ= +github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= +github.com/twpayne/go-geom v1.6.1 h1:iLE+Opv0Ihm/ABIcvQFGIiFBXd76oBIar9drAwHFhR4= +github.com/twpayne/go-geom v1.6.1/go.mod h1:Kr+Nly6BswFsKM5sd31YaoWS5PeDDH2NftJTK7Gd028= +github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM= golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE= +golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= +golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= +google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= diff --git a/Common/Client/go/wshub/capture_shortwindow_test.go b/Common/Client/go/wshub/capture_shortwindow_test.go new file mode 100644 index 0000000..0ab359a --- /dev/null +++ b/Common/Client/go/wshub/capture_shortwindow_test.go @@ -0,0 +1,75 @@ +package wshub + +import ( + "math" + "testing" +) + +// A short window at a high sample rate fits in a ring's initial capacity, so the +// retune sweep used to leave it there — and a ring holding exactly the window has +// already rolled past the front of a capture by the time that capture is read, +// which happens a post-window plus captureMarginSec after the trigger fires. +// +// 1 MSps over a 200 ms window: 200 k points fit in the 250 k initial ring, and +// every shot came back missing its first 123 ms. +func TestCaptureWholeAtHighRateShortWindow(t *testing.T) { + const ( + key = "s1:Ch1" + rate = 1e6 + window = 0.2 + prePct = 20.0 + batchSec = 1.0 / 30.0 + simSec = 6.0 + ) + + h := NewHub() + h.SetRingBudget(defaultRingPts) + h.rings[key] = newSigRing(ringCapInitial) + h.trigger.SetConfig(trigConfig{signalKey: key, edge: "rising", threshold: 0, + windowSec: window, prePercent: prePct, mode: "normal", holdoffSec: 0.2}) + + rateHz := float64(rate) + nBatch := int(rateHz * batchSec) + ts := make([]float64, nBatch) + vs := make([]float64, nBatch) + + armed, shots := false, 0 + for now := 0.0; now < simSec; now += batchSec { + for i := range ts { + ts[i] = now + float64(i)/rateHz + vs[i] = math.Sin(2 * math.Pi * 5 * ts[i]) // a rising crossing every 200 ms + } + h.ingest(key, 1, ts, vs) + h.retuneRings(now) + h.refreshTriggerFill() + + if !armed && now > 2 { + h.trigger.Arm() + armed = true + } + trigTime, pre, post, ok := h.trigger.dueCapture(now + batchSec) + if !ok { + if h.trigger.dueRearm(now + batchSec) { + h.trigger.Arm() + } + continue + } + t0 := trigTime - pre + buf := h.buildTriggerCapture(trigTime, pre, post) + if buf == nil { + t.Fatalf("shot at t=%.4f produced no frame at all", trigTime) + } + first, last, n := decodeCaptureSpan(t, buf, key) + shots++ + if lost := first - t0; lost > shortCaptureTol*window { + _, span := h.rings[key].stats() + t.Errorf("shot at t=%.4f is missing %.0f ms at the front of its %.0f ms window "+ + "(got [%.4f,%.4f], %d pts; ring holds %.4f s in %d points)", + trigTime, 1e3*lost, 1e3*window, first, last, n, span, h.rings[key].capacity()) + } + h.trigger.markTriggered(now + batchSec) + } + if shots < 3 { + t.Fatalf("only %d shots in %.0f s", shots, simSec) + } +} diff --git a/Common/Client/go/wshub/export.go b/Common/Client/go/wshub/export.go new file mode 100644 index 0000000..77ee9bb --- /dev/null +++ b/Common/Client/go/wshub/export.go @@ -0,0 +1,119 @@ +package wshub + +import ( + "fmt" + "net/http" + "sort" + "strconv" + "strings" + "time" + + "github.com/parquet-go/parquet-go" +) + +// ExportSample is one row of the binary export: a single stored sample, in +// long ("tidy") form, keyed by source and signal with its own timestamp. +// +// Keeping each signal's samples as its own rows — rather than resampling onto a +// shared time grid — is what makes the export hole-free: per-signal streams of +// different lengths export exactly as stored, nothing is fabricated, and +// nothing is dropped. +type ExportSample struct { + Source string `parquet:"source"` + Signal string `parquet:"signal"` + Time float64 `parquet:"time"` + Value float64 `parquet:"value"` +} + +// exportChunkRows bounds each batched write and, via MaxRowsPerRowGroup, the +// size of each parquet row group: memory stays bounded however large the +// export is, because a finished row group is flushed to the HTTP stream. +const exportChunkRows = 65536 + +// exportWriteBuffer is the parquet writer's output buffer: larger than the +// 32KiB default means fewer writes on the HTTP stream for a multi-GB export. +const exportWriteBuffer = 1 << 20 + +// HandleExport serves GET /api/export?t0=..&t1=..[&signals=a,b] as a Parquet +// file containing every stored sample of the named signals in [t0, t1]. +// +// Unlike /api/zoom there is no decimation: the file holds the full contents of +// the rings. At rates above the ring budget those contents are min/max buckets +// (the finest resolution the hub retains); at lower rates they are verbatim. +func (h *Hub) HandleExport(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + t0, err0 := strconv.ParseFloat(q.Get("t0"), 64) + t1, err1 := strconv.ParseFloat(q.Get("t1"), 64) + if err0 != nil || err1 != nil || t1 <= t0 { + http.Error(w, "invalid t0/t1", http.StatusBadRequest) + return + } + var keys []string + if s := strings.TrimSpace(q.Get("signals")); s != "" { + keys = strings.Split(s, ",") + for i := range keys { + keys[i] = strings.TrimSpace(keys[i]) + } + } + + // Snapshot the rings we will read. A signal removed mid-export must not + // silently drop rows from the file. + h.ringsMu.RLock() + refs := make(map[string]*sigRing) + if keys == nil { + for k, rb := range h.rings { + refs[k] = rb + } + } else { + for _, k := range keys { + if rb, ok := h.rings[k]; ok { + refs[k] = rb + } + } + } + h.ringsMu.RUnlock() + if len(refs) == 0 { + http.Error(w, "no signals", http.StatusNotFound) + return + } + + // Deterministic column order. + names := make([]string, 0, len(refs)) + for k := range refs { + names = append(names, k) + } + sort.Strings(names) + + w.Header().Set("Content-Type", "application/vnd.apache.parquet") + w.Header().Set("Content-Disposition", + fmt.Sprintf("attachment; filename=\"signals_%d.parquet\"", time.Now().Unix())) + + writer := parquet.NewGenericWriter[ExportSample](w, + parquet.MaxRowsPerRowGroup(exportChunkRows), + parquet.WriteBufferSize(exportWriteBuffer), + ) + batch := make([]ExportSample, 0, exportChunkRows) + for _, key := range names { + st, sv := refs[key].slice(t0, t1) + colon := strings.IndexByte(key, ':') + source, signal := key, key + if colon >= 0 { + source = key[:colon] + signal = key[colon+1:] + } + for i := range st { + batch = append(batch, ExportSample{Source: source, Signal: signal, Time: st[i], Value: sv[i]}) + if len(batch) >= exportChunkRows { + if _, err := writer.Write(batch); err != nil { + // Client went away or the stream broke; stop writing. + return + } + batch = batch[:0] + } + } + } + if len(batch) > 0 { + _, _ = writer.Write(batch) + } + _ = writer.Close() +} diff --git a/Common/Client/go/wshub/export_test.go b/Common/Client/go/wshub/export_test.go new file mode 100644 index 0000000..f32745d --- /dev/null +++ b/Common/Client/go/wshub/export_test.go @@ -0,0 +1,87 @@ +package wshub + +import ( + "bytes" + "net/http/httptest" + "testing" + + "github.com/parquet-go/parquet-go" +) + +func TestHandleExportParquetFullResolution(t *testing.T) { + h := NewHub() + // Two signals with different lengths and offset time bases: the export must + // keep every sample of each, on its own timestamps (no holes, no + // resampling, no decimation). + sig1 := newSigRing(10000) + sig2 := newSigRing(10000) + t1, v1 := make([]float64, 1000), make([]float64, 1000) + for i := range t1 { + t1[i] = float64(i) * 0.001 + v1[i] = float64(i) * 2 + } + sig1.write(t1, v1) + t2, v2 := make([]float64, 500), make([]float64, 500) + for i := range t2 { + t2[i] = 0.1 + float64(i)*0.002 + v2[i] = -float64(i) + } + sig2.write(t2, v2) + h.rings["s1:Ch1"] = sig1 + h.rings["s1:Ch2"] = sig2 + + req := httptest.NewRequest("GET", "/api/export?t0=0&t1=2&signals=s1:Ch1,s1:Ch2", nil) + rec := httptest.NewRecorder() + h.HandleExport(rec, req) + if rec.Code != 200 { + t.Fatalf("status = %d, want 200 (body: %s)", rec.Code, rec.Body.String()) + } + + reader := parquet.NewGenericReader[ExportSample](bytes.NewReader(rec.Body.Bytes())) + defer reader.Close() + + var got []ExportSample + buf := make([]ExportSample, 1000) + for { + n, err := reader.Read(buf) + got = append(got, buf[:n]...) + if err != nil { + break + } + } + if len(got) != 1500 { + t.Fatalf("rows = %d, want 1500 (every sample of both signals)", len(got)) + } + ch1 := filterExportSamples(got, "s1", "Ch1") + ch2 := filterExportSamples(got, "s1", "Ch2") + if len(ch1) != 1000 || len(ch2) != 500 { + t.Fatalf("ch1=%d ch2=%d rows, want 1000/500 (no holes, no resampling)", len(ch1), len(ch2)) + } + if ch1[0].Time != 0 || ch1[0].Value != 0 || ch1[999].Time != 0.999 || ch1[999].Value != 1998 { + t.Fatalf("ch1 endpoints wrong: first=%+v last=%+v", ch1[0], ch1[999]) + } + if ch2[0].Time != 0.1 || ch2[499].Time != 0.1+499*0.002 || ch2[499].Value != -499 { + t.Fatalf("ch2 endpoints wrong: first=%+v last=%+v", ch2[0], ch2[499]) + } +} + +func filterExportSamples(rows []ExportSample, source, signal string) []ExportSample { + out := make([]ExportSample, 0, len(rows)) + for _, r := range rows { + if r.Source == source && r.Signal == signal { + out = append(out, r) + } + } + return out +} + +func TestHandleExportParquetBadRange(t *testing.T) { + h := NewHub() + h.rings["s1:Ch1"] = newSigRing(10) + req := httptest.NewRequest("GET", "/api/export?t0=2&t1=1", nil) + rec := httptest.NewRecorder() + h.HandleExport(rec, req) + if rec.Code != 400 { + t.Fatalf("status = %d, want 400 for inverted range", rec.Code) + } +} diff --git a/Common/Client/go/wshub/history.go b/Common/Client/go/wshub/history.go index c6d5756..1f03b3c 100644 --- a/Common/Client/go/wshub/history.go +++ b/Common/Client/go/wshub/history.go @@ -422,6 +422,26 @@ func (hw *historyWriter) window() float64 { return hw.windowSec } +// coversWindow reports whether the archive file for key currently spans at +// least sec seconds. When true, backfillCaptureHead can reconstruct a capture's +// front out of the archive, so the trigger need not wait for the ring to cover +// the whole window on its own. +func (hw *historyWriter) coversWindow(key string, sec float64) bool { + if !hw.enabled() || !(sec > 0) { + return false + } + hw.mu.RLock() + hf, ok := hw.files[key] + hw.mu.RUnlock() + if !ok { + return false + } + hf.mu.RLock() + span := hf.tNewest - hf.tOldest + hf.mu.RUnlock() + return span >= sec +} + // setWindow points the archive at the timespan the clients are looking at, and // re-sizes the files that no longer match it. It reports whether any file's // geometry changed, which invalidates what clients know about the archive. @@ -842,10 +862,7 @@ func (hf *histFile) readAfter(after, t0, t1 float64, max int) ([]byte, float64, // The run wraps at most once, so it costs at most two reads. buf := make([]byte, n*histPairSize) start := (oldest + lo) % capacity - head := int(capacity-start) * histPairSize - if head > len(buf) { - head = len(buf) - } + head := min(int(capacity-start)*histPairSize, len(buf)) if _, err := hf.f.ReadAt(buf[:head], int64(histHeaderSize)+int64(start)*histPairSize); err != nil { return nil, 0, err } @@ -898,10 +915,8 @@ func (hf *histFile) writePairs(t, v []float64) error { binary.LittleEndian.PutUint64(buf[i*histPairSize+8:], math.Float64bits(v[i])) } - first := int(hf.capacity - hf.head) - if first > n { - first = n - } + first := min(int(hf.capacity-hf.head), n) + off := int64(histHeaderSize) + int64(hf.head)*histPairSize if _, err := hf.f.WriteAt(buf[:first*histPairSize], off); err != nil { return err @@ -1089,10 +1104,7 @@ func (hw *historyWriter) readRange(key string, t0, t1 float64, maxOut int) ([]fl // Read in contiguous runs: the range wraps at most once. buf := make([]byte, n*histPairSize) start := (oldest + lo) % capacity - first := int(capacity - start) - if first > n { - first = n - } + first := min(int(capacity-start), n) if _, err := hf.f.ReadAt(buf[:first*histPairSize], int64(histHeaderSize)+int64(start)*histPairSize); err != nil { return nil, nil @@ -1265,7 +1277,7 @@ func (h *Hub) handleSetHistoryBudget(env map[string]interface{}) { // handleHistoryZoom answers a historyZoom request from disk. Same request and // reply shape as "zoom", so clients can fall back to it transparently when a // window reaches further back than the in-memory rings hold. -func (h *Hub) handleHistoryZoom(c *wsClient, env map[string]interface{}) { +func (h *Hub) handleHistoryZoom(c *wsClient, env map[string]any) { if !h.hist.enabled() { msg, _ := json.Marshal(map[string]any{ "type": "historyZoom", "reqId": env["reqId"], @@ -1287,10 +1299,7 @@ func (h *Hub) handleHistoryZoom(c *wsClient, env map[string]interface{}) { // oversampled relative to the plot's point budget and thinned afterwards. // The cap keeps a request for "no decimation" over a multi-hour window from // pulling the whole file into memory. - readCap := n * histReadOversample - if readCap > histMaxReadPoints { - readCap = histMaxReadPoints - } + readCap := min(n*histReadOversample, histDefaultMaxPoints) signals := make(map[string]sigData) for _, k := range strings.Split(sigCSV, ",") { diff --git a/Common/Client/go/wshub/ringbuf.go b/Common/Client/go/wshub/ringbuf.go index cce162d..3c0a1bf 100644 --- a/Common/Client/go/wshub/ringbuf.go +++ b/Common/Client/go/wshub/ringbuf.go @@ -221,6 +221,19 @@ func ringCoverage(bucket, capacity int) int { return capacity / 2 * bucket } +// captureLagSec is how much further back than the window itself a ring has to +// reach to deliver a capture of it. +// +// A capture is not read out when its last sample arrives but captureMarginSec +// later, and then only on the next push tick — so by the time the window is +// extracted, its oldest sample is that much deeper in the ring. A ring holding +// exactly the window has already overwritten the front of its own capture, which +// is what made every shot at a short window come back missing its head. The +// pre/post split does not enter into it: the harvest is a post-window after the +// trigger and the read reaches a pre-window before it, so the two sum to the +// window whatever the split. +const captureLagSec = captureMarginSec + 1.0/30.0 + // activeWindowSec is the timespan the buffers must cover. An armed trigger owns // it: its pre-window has to already be in the ring when the trigger fires or // there is nothing to back-fill the capture from. Otherwise it is the widest @@ -228,7 +241,7 @@ func ringCoverage(bucket, capacity int) int { func (h *Hub) activeWindowSec() float64 { if h.trigger != nil && h.trigger.Active() { if cfg := h.trigger.Config(); cfg.windowSec > 0 { - return cfg.windowSec + return cfg.windowSec + captureLagSec } } widest := 0.0 diff --git a/Common/Client/go/wshub/ringbuf_test.go b/Common/Client/go/wshub/ringbuf_test.go index 84d1442..32f81a1 100644 --- a/Common/Client/go/wshub/ringbuf_test.go +++ b/Common/Client/go/wshub/ringbuf_test.go @@ -120,7 +120,9 @@ func TestActiveWindowSecTakesTheWidestClientWindow(t *testing.T) { } // An armed trigger owns the window: its pre-window has to be in the buffer -// before the trigger fires or the capture has nothing to back-fill from. +// before the trigger fires or the capture has nothing to back-fill from. The +// buffers must reach back past the window itself, because the capture is read +// out a margin and a tick after its last sample lands. func TestActiveWindowSecPrefersTheArmedTrigger(t *testing.T) { h := NewHub() c := &wsClient{} @@ -128,8 +130,9 @@ func TestActiveWindowSecPrefersTheArmedTrigger(t *testing.T) { h.clients[c] = true h.trigger.SetConfig(trigConfig{signalKey: "s1:sig", windowSec: 45, mode: "normal"}) - if got := h.activeWindowSec(); got != 45 { - t.Fatalf("activeWindowSec = %v, want the trigger's 45", got) + if got := h.activeWindowSec(); got != 45+captureLagSec { + t.Fatalf("activeWindowSec = %v, want the trigger's 45 plus the %v harvest lag", + got, captureLagSec) } } diff --git a/Common/Client/go/wshub/trigger.go b/Common/Client/go/wshub/trigger.go index 64d4aa9..6ff6d8a 100644 --- a/Common/Client/go/wshub/trigger.go +++ b/Common/Client/go/wshub/trigger.go @@ -92,6 +92,14 @@ type triggerEngine struct { bufGrowth float64 bufKnown bool bufRateOK bool + // bufCoverage is the maximum span (seconds) the ring can reach at its + // current bucket and capacity — the gate must never demand more than this, + // or a ring whose coverage is below the window can never satisfy it. 0 = + // unknown (no measurable rate). + bufCoverage float64 + // bufArchived is true when the disk history already spans the trigger + // window, so a short capture's front can be back-filled from it. + bufArchived bool // Reference point the growth is measured against. bufRefSpan, bufRefWall float64 @@ -108,6 +116,22 @@ type triggerEngine struct { firedPost float64 firedValid bool + // The edge to fire on as soon as the FSM rearms, in sample time. Recorded + // while a capture is still being collected or handed out, for edges late + // enough that a capture of them would not overlap the one in flight. + // + // Without this the trigger is deaf from its own trigger point until the + // capture has been harvested — a post-window plus captureMarginSec — and + // then for the holdoff on top of that, and afterwards waits for a FRESH + // edge. On a sparse pulse train that rounds the capture spacing up to a + // whole pulse period: at the default 1 s window the blind stretch comes to + // 1.15 s, so a 1 Hz train was caught at 0.5 Hz and a wider window lost whole + // multiples. Remembering the edge instead makes the blind stretch exactly + // the post-window it has to be, since the capture is built from the edge's + // own timestamp and the ring still holds everything around it. + pendingT float64 + pendingValid bool + rearmAt float64 // wall-clock seconds; 0 when no rearm is pending } @@ -163,10 +187,14 @@ func (te *triggerEngine) SetConfig(cfg trigConfig) { if base != te.baseKey { // The buffer measurement belongs to the old signal's ring. te.bufKnown, te.bufRateOK = false, false + te.bufCoverage, te.bufArchived = 0, false } te.baseKey, te.elemIdx = base, idx te.prevValid = false te.prevValue = 0 + // An edge held over from the old configuration would be latched against the + // new window, whose fill the gate has not vouched for. + te.pendingValid = false } func (te *triggerEngine) Config() trigConfig { @@ -175,15 +203,37 @@ func (te *triggerEngine) Config() trigConfig { return te.cfg } +// Arm starts a fresh acquisition. It is the user's own arm, so it discards any +// edge remembered during the previous capture: the user asked for the next +// event, not for one that has already been and gone. func (te *triggerEngine) Arm() { te.mu.Lock() te.state = trigArmed te.prevValid = false te.prevValue = 0 + te.pendingValid = false te.rearmAt = 0 te.mu.Unlock() } +// rearm is the automatic arm at the end of a capture. Unlike Arm it honours an +// edge that arrived while the capture was being collected, firing on it at once +// rather than waiting for the next one — see pendingT. It also keeps the level +// tracked through the dead time, so the first sample after rearming is compared +// against its real predecessor instead of being spent seeding one. +func (te *triggerEngine) rearm() { + te.mu.Lock() + te.rearmAt = 0 + if te.pendingValid { + t := te.pendingT + te.pendingValid = false + te.latchWindowLocked(t) + } else { + te.state = trigArmed + } + te.mu.Unlock() +} + func (te *triggerEngine) Disarm() { te.mu.Lock() te.state = trigIdle @@ -191,6 +241,7 @@ func (te *triggerEngine) Disarm() { te.prevValid = false te.prevValue = 0 te.firedValid = false + te.pendingValid = false te.rearmAt = 0 te.mu.Unlock() } @@ -243,13 +294,16 @@ const bufGrowthIntervalSec = 0.5 const bufGrowthSmooth = 0.5 // setBuffered records how far back the trigger signal's ring reaches, at wall -// clock now, and derives how fast that is growing. Pass known=false when there -// is no such ring. -func (te *triggerEngine) setBuffered(span float64, known bool, now float64) { +// clock now, and derives how fast that is growing. coverage is the maximum +// span (seconds) the ring can reach at its current bucket/capacity; archived +// says the disk history already spans the trigger window. Pass known=false when +// there is no ring to measure. +func (te *triggerEngine) setBuffered(span, coverage float64, archived, known bool, now float64) { te.mu.Lock() defer te.mu.Unlock() if !known { te.bufKnown, te.bufRateOK = false, false + te.bufCoverage, te.bufArchived = 0, false return } if !te.bufKnown { @@ -257,6 +311,8 @@ func (te *triggerEngine) setBuffered(span float64, known bool, now float64) { te.bufRefSpan, te.bufRefWall = span, now } te.bufSpan = span + te.bufCoverage = coverage + te.bufArchived = archived dt := now - te.bufRefWall if dt < bufGrowthIntervalSec { return @@ -294,9 +350,17 @@ func (te *triggerEngine) setBuffered(span float64, known bool, now float64) { // anyway. A full one grows only as fast as its incoming samples free space — // re-bucketing to a longer window replaces dense old samples with sparse new // ones — and it is that case, growth well below 1, where firing on the -// pre-window alone delivers a capture whose front has been overwritten by the -// time it is read. In the steady state growth is 0 and need is the whole -// window, which a ring tuned for that window already exceeds, so nothing waits. +// +// Two escapes keep an armed trigger from staying deaf forever: +// +// - archived — the disk history already spans the window, so the front of a +// capture can be back-filled from it; the ring only needs to +// hold the pre-window worth of recent data. +// - coverage — never demand more than the ring can physically reach. If its +// coverage saturates below the window (a measured source rate +// that over-estimates the true one), the gate opens once the +// ring is full anyway and a short capture is delivered instead +// of deafness. func (te *triggerEngine) fillNeedLocked() float64 { pre := te.cfg.windowSec * te.cfg.prePercent / 100 growth := 0.0 // until measured, assume the buffer will not fill on its own @@ -307,6 +371,14 @@ func (te *triggerEngine) fillNeedLocked() float64 { if need < pre { need = pre } + if te.bufArchived { + // The archive back-fills the front; the ring holds the post-trigger + // window live, so the pre-window is all it needs to have reached. + return pre + } + if te.bufCoverage > 0 && need > te.bufCoverage { + need = te.bufCoverage + } return need } @@ -366,7 +438,11 @@ func (te *triggerEngine) feed(key string, nElem int, t, v []float64) { te.lastT = t[len(t)-1] te.lastTOK = true te.lastFeedWall = float64(time.Now().UnixNano()) / 1e9 - if te.state != trigArmed { + + // A capture in flight does not stop the comparator; it only changes what an + // edge does. See pendingT. + inFlight := te.state == trigCollecting || te.state == trigTriggered + if te.state != trigArmed && !inFlight { return } step, start := 1, 0 @@ -381,12 +457,20 @@ func (te *triggerEngine) feed(key string, nElem int, t, v []float64) { // which is what made the first shot after a window change come back short. // Track the level meanwhile, so the first edge once the buffer is deep // enough is still measured against the right previous sample. - if te.fillLocked() < 1 { + if !inFlight && te.fillLocked() < 1 { for i := start; i < len(v); i += step { te.prevValue, te.prevValid = v[i], true } return } + // The earliest trigger point a new capture may take. The one in flight owns + // everything up to the end of its own post-window, and the holdoff — a guard + // against re-triggering on the ringing of the SAME event — is measured from + // its trigger point too, so the two overlap rather than add. + notBefore := math.Inf(1) + if inFlight && te.firedValid { + notBefore = te.trigTime + math.Max(te.firedPost, te.cfg.holdoffSec) + } thr := te.cfg.threshold for i := start; i < len(t); i += step { if !te.prevValid { @@ -406,10 +490,19 @@ func (te *triggerEngine) feed(key string, nElem int, t, v []float64) { default: fired = up } - if fired { + if !fired { + continue + } + if !inFlight { te.latchWindowLocked(t[i]) return } + // Keep the FIRST qualifying edge and go on tracking the level: a later + // one would be no more use, and stopping here would leave prevValue + // stale by the time the FSM rearms. + if !te.pendingValid && t[i] >= notBefore { + te.pendingT, te.pendingValid = t[i], true + } } } @@ -592,19 +685,32 @@ func (h *Hub) refreshTriggerFill() { return } now := float64(time.Now().UnixNano()) / 1e9 + key := h.trigger.baseSignalKey() var rb *sigRing - if key := h.trigger.baseSignalKey(); key != "" { + if key != "" { rb = h.getRing(key) } if rb == nil { // Nothing to measure. Do not gate on a signal the hub does not carry: // that would leave the trigger armed forever, which is worse than a // short capture. - h.trigger.setBuffered(0, false, now) + h.trigger.setBuffered(0, 0, false, false, now) return } _, span := rb.stats() - h.trigger.setBuffered(span, true, now) + // Maximum span the ring can ever reach at its current bucket/capacity, in + // seconds. The gate must never demand more than this, or a ring whose + // coverage is below the window (a measured source rate that over-estimates + // the true one) can never satisfy it. + coverage := 0.0 + if rate := rb.sourceRate(); rate > 0 { + coverage = float64(ringCoverage(rb.bucketSize(), rb.capacity())) / rate + } + // If the disk archive already spans the trigger window, the front of a + // short capture can be back-filled from it, so the ring need not cover the + // whole window on its own. + archived := h.hist.coversWindow(key, h.trigger.Config().windowSec) + h.trigger.setBuffered(span, coverage, archived, true, now) } // triggerTick services the trigger FSM; called from Hub.Run() on every push tick. @@ -640,7 +746,7 @@ func (h *Hub) triggerTick() { // file of its own, where nothing overwrites it until the next trigger. h.hist.captureRange(trigTime-pre, trigTime+post) } else if h.trigger.dueRearm(nowSec) { - h.trigger.Arm() + h.trigger.rearm() } if h.trigger.stateUnsent() { diff --git a/Common/Client/go/wshub/trigger_sporadic_test.go b/Common/Client/go/wshub/trigger_sporadic_test.go new file mode 100644 index 0000000..a900758 --- /dev/null +++ b/Common/Client/go/wshub/trigger_sporadic_test.go @@ -0,0 +1,367 @@ +package wshub + +import ( + "encoding/binary" + "math" + "testing" +) + +/* +Sporadic-signal trigger coverage. + +Every other trigger test in this package feeds a periodic waveform, or a +hand-built two-sample batch. Neither can show a trigger that is blind most of +the time: a sine crosses the threshold again a few milliseconds after every +missed crossing, so a trigger losing 80 % of its edges still fires steadily and +looks healthy. A sparse train — 0000000111000000000000, one short burst in a +long flat run — has nothing to fall back on, so every missed edge is a missed +capture and the yield is a direct measure of how long the FSM was deaf. + +That deafness is what these tests pin down. It is not a bug in itself: a capture +cannot be harvested before the samples after its trigger point exist, so the +trigger is necessarily blind for its own post-trigger window. What must NOT +happen is for edges arriving after that window to be thrown away as well. +*/ + +// pulseTrainSim drives a Hub the way Run() does — ingest on one side, the +// trigger tick on the other — on a simulated clock. +type pulseTrainSim struct { + rateHz float64 + batchSec float64 + pulsePeriod float64 + pulseSamples int + simSec float64 + windowSec float64 + prePercent float64 + holdoffSec float64 + armAt float64 + // When windowChangeAt > 0 the window is switched to windowChangeTo at that + // time and the trigger re-armed, as a user editing the trigger bar would. + windowChangeAt float64 + windowChangeTo float64 +} + +type pulseTrainResult struct { + pulses int // pulse starts presented after the trigger was armed + shots int // captures actually delivered + trigTimes []float64 // the sample time each capture triggered on + + worstCov float64 // smallest fraction of its window a capture spanned + holdDeclined int // captures the zoom hold would not answer for + drawnPulses int // pulses visible in the delivered frames + wantPulses int // pulses those frames' windows really contained + gatedPulses int // pulses that arrived armed but with the fill gate shut +} + +// yield is the fraction of presented pulses that produced a capture. +func (r pulseTrainResult) yield() float64 { + if r.pulses == 0 { + return 0 + } + return float64(r.shots) / float64(r.pulses) +} + +// run executes the simulation and returns what the trigger caught. +func (s pulseTrainSim) run(t *testing.T, key string) pulseTrainResult { + t.Helper() + h := NewHub() + h.rings[key] = newSigRing(ringCapInitial) + h.trigger.SetConfig(trigConfig{ + signalKey: key, edge: "rising", threshold: 0.5, + windowSec: s.windowSec, prePercent: s.prePercent, + mode: "normal", holdoffSec: s.holdoffSec, + }) + + res := pulseTrainResult{worstCov: 1} + nBatch := int(s.rateHz * s.batchSec) + ts := make([]float64, nBatch) + vs := make([]float64, nBatch) + + armed, changed := false, false + for now := 0.0; now < s.simSec; now += s.batchSec { + nPulseStarts := 0 + for i := range ts { + ts[i] = now + float64(i)/s.rateHz + // Position within the current pulse period, in samples. + k := int((ts[i] - math.Floor(ts[i]/s.pulsePeriod)*s.pulsePeriod) * s.rateHz) + if k < s.pulseSamples { + vs[i] = 1 + if k == 0 { + nPulseStarts++ + } + } else { + vs[i] = 0 + } + } + + if !armed && now >= s.armAt { + h.trigger.Arm() + armed = true + } + if s.windowChangeAt > 0 && !changed && now >= s.windowChangeAt { + cfg := h.trigger.Config() + cfg.windowSec = s.windowChangeTo + h.trigger.SetConfig(cfg) + h.trigger.Arm() + changed = true + } + + if armed { + res.pulses += nPulseStarts + if nPulseStarts > 0 && h.trigger.State() == trigArmed { + h.trigger.mu.Lock() + f := h.trigger.fillLocked() + h.trigger.mu.Unlock() + if f < 1 { + res.gatedPulses += nPulseStarts + } + } + } + h.ingest(key, 1, ts, vs) + + // Mirror triggerTick, on the simulated clock. + tick := now + s.batchSec + h.retuneRings(tick) + _, span := h.rings[key].stats() + h.trigger.setBuffered(span, 0, false, true, tick) + + if trigTime, pre, post, ok := h.trigger.dueCapture(tick); ok { + if buf := h.buildTriggerCapture(trigTime, pre, post); buf != nil { + res.shots++ + res.trigTimes = append(res.trigTimes, trigTime) + first, last, _ := decodeCaptureSpan(t, buf, key) + if cov := (last - first) / (pre + post); cov < res.worstCov { + res.worstCov = cov + } + if _, _, ok := h.capture.slice(key, trigTime-pre, trigTime+post); !ok { + res.holdDeclined++ + } + // What the client would actually draw, against what the window + // really contained. A wide window holds several pulses, and a + // frame showing only the one it triggered on has lost the rest + // between the ring, the bucketing and the decimation. + _, fv := decodeCaptureSig(t, buf, key) + res.drawnPulses += countPulses(fv, 0.5) + res.wantPulses += countPulseStarts(trigTime-pre, trigTime+post, + s.pulsePeriod, 1/s.rateHz) + } + h.trigger.markTriggered(tick) + } else if h.trigger.dueRearm(tick) { + h.trigger.rearm() + } + } + return res +} + +// decodeCaptureSig pulls one signal's samples out of a v2 capture frame. +func decodeCaptureSig(t *testing.T, buf []byte, key string) (ts, vs []float64) { + t.Helper() + off := 1 + 8 + 8 + 8 + nSig := int(binary.LittleEndian.Uint32(buf[off:])) + off += 4 + for i := 0; i < nSig; i++ { + kl := int(binary.LittleEndian.Uint16(buf[off:])) + off += 2 + k := string(buf[off : off+kl]) + off += kl + cnt := int(binary.LittleEndian.Uint32(buf[off:])) + off += 4 + if k == key { + ts = make([]float64, cnt) + vs = make([]float64, cnt) + for j := 0; j < cnt; j++ { + ts[j] = math.Float64frombits(binary.LittleEndian.Uint64(buf[off+j*8:])) + vs[j] = math.Float64frombits(binary.LittleEndian.Uint64(buf[off+cnt*8+j*8:])) + } + } + off += cnt * 16 + } + return +} + +// countPulses counts runs of samples at or above thr. +func countPulses(v []float64, thr float64) int { + n, in := 0, false + for _, x := range v { + if x >= thr { + if !in { + n, in = n+1, true + } + } else { + in = false + } + } + return n +} + +// countPulseStarts is how many pulse starts fall inside [t0, t1]. A pulse +// starting within one sample of t1 is not counted: only its first sample is +// inside the window, and the ring's min/max bucket for it may put that sample's +// extremum just past the edge, which is a boundary artefact rather than a loss. +func countPulseStarts(t0, t1, period, dt float64) int { + n := 0 + for k := math.Floor(t0 / period); k*period <= t1; k++ { + if p := k * period; p >= t0 && p < t1-2*dt { + n++ + } + } + return n +} + +// deadTimeSec is how long the FSM is blind after firing at t: it must acquire +// the post-trigger window before the capture can be harvested, and the holdoff +// guards against re-triggering on the same event. Both are measured from the +// trigger point, so they overlap rather than add. +func deadTimeSec(windowSec, prePercent, holdoffSec float64) float64 { + return math.Max(windowSec*(1-prePercent/100), holdoffSec) +} + +// A trigger cannot show two windows at once, so pulses closer together than its +// post-trigger window are necessarily lost. Pulses spaced FURTHER apart than +// that are not: nothing about the acquisition prevents catching every one. +// +// This is the reported failure. The FSM used to go deaf from the trigger point +// until the capture had been harvested (a post-window plus captureMarginSec) +// and the holdoff had then elapsed on top of that, then wait for a fresh edge — +// so the effective spacing was rounded UP to a whole pulse period. At the +// default 1 s window and 0.2 s holdoff the blind stretch came to 1.15 s, which +// is longer than a 1 s pulse period by a hair, and a pulse train at 1 Hz was +// caught at 0.5 Hz. Widening the window made it worse in whole multiples. +func TestSporadicPulsesWiderThanThePostWindowAreAllCaught(t *testing.T) { + const key = "s1:Ch1" + cases := []struct{ window, period float64 }{ + {0.5, 0.5}, // post 0.4 s + {1.0, 1.0}, // post 0.8 s — the case the report was made against + {2.0, 2.0}, // post 1.6 s + {5.0, 5.0}, // post 4.0 s + } + for _, c := range cases { + sim := pulseTrainSim{ + rateHz: 1000, batchSec: 1.0 / 30.0, + pulsePeriod: c.period, pulseSamples: 3, + simSec: 41 * c.period, windowSec: c.window, prePercent: 20, + holdoffSec: autoRearmDelaySec, armAt: c.period, + } + res := sim.run(t, key) + // Two pulses are always in flight rather than caught: the one that lands + // as the trigger arms, and the one still being collected when the run + // ends. + if got := res.yield(); got < 0.94 { + t.Errorf("window %.1f s, pulse every %.1f s: caught %d of %d (%.0f%%); "+ + "the post-trigger window is only %.2f s, so every pulse fits", + c.window, c.period, res.shots, res.pulses, 100*got, + c.window*0.8) + } + } +} + +// The loss that remains must be the loss that has to remain. A capture cannot +// start before the previous one's post-window is acquired, and it can only start +// on a pulse, so consecutive captures are a dead time apart rounded UP to the +// next pulse — never further. Any longer gap means an edge that the acquisition +// no longer needed was thrown away anyway. +// +// The bound is stated as dead + period rather than ceil(dead/period)*period +// because when the two divide exactly, whether the pulse at the boundary counts +// comes down to the last bit of the sample timestamp. Both answers are correct; +// a gap beyond either is not. +func TestSporadicCaptureGapsStayWithinTheDeadTime(t *testing.T) { + const key = "s1:Ch1" + for _, window := range []float64{0.5, 1.0, 2.0, 5.0} { + for _, period := range []float64{0.25, 0.5, 1.0, 2.0} { + sim := pulseTrainSim{ + rateHz: 1000, batchSec: 1.0 / 30.0, + pulsePeriod: period, pulseSamples: 3, + simSec: 60, windowSec: window, prePercent: 20, + holdoffSec: autoRearmDelaySec, armAt: 1.0, + } + res := sim.run(t, key) + dead := deadTimeSec(window, 20, autoRearmDelaySec) + limit := dead + period + 2*sim.batchSec + worst, worstAt := 0.0, 0.0 + for i := 1; i < len(res.trigTimes); i++ { + if g := res.trigTimes[i] - res.trigTimes[i-1]; g > worst { + worst, worstAt = g, res.trigTimes[i-1] + } + } + if worst > limit { + t.Errorf("window %.1f s, pulse every %.2f s: %.2f s between the captures "+ + "at %.2f s and %.2f s; the dead time is only %.2f s, so %.2f s is the most "+ + "that can be missed", + window, period, worst, worstAt, worstAt+worst, dead, limit) + } + t.Logf("window %.1f s, pulse every %.2f s: %d/%d captures (%.0f%%), "+ + "dead time %.2f s, worst gap %.2f s, gated %d", + window, period, res.shots, res.pulses, 100*res.yield(), dead, + worst, res.gatedPulses) + } + } +} + +// Whatever the trigger does catch has to come back whole: a window wide enough +// to hold several pulses must show all of them, at every rate, including the +// rates that force the ring into min/max bucketing. +func TestSporadicCaptureShowsEveryPulseInItsWindow(t *testing.T) { + const key = "s1:Ch1" + for _, rate := range []float64{1000, 200e3} { + for _, window := range []float64{1.0, 2.0, 5.0} { + sim := pulseTrainSim{ + rateHz: rate, batchSec: 1.0 / 30.0, + pulsePeriod: 0.5, pulseSamples: 3, + simSec: 40, windowSec: window, prePercent: 20, + holdoffSec: autoRearmDelaySec, armAt: 1.0, + } + res := sim.run(t, key) + if res.shots == 0 { + t.Fatalf("rate %.0f window %.1f s: no captures at all", rate, window) + } + // wantPulses excludes the pulse straddling each window's far edge, + // whose bucket may place its extremum just past it, so the frames + // may legitimately draw a few more than that — but never fewer. + if res.drawnPulses < res.wantPulses { + t.Errorf("rate %.0f window %.1f s: frames drew %d pulses, their windows held %d", + rate, window, res.drawnPulses, res.wantPulses) + } + if res.worstCov < 0.98 { + t.Errorf("rate %.0f window %.1f s: worst capture spanned %.0f%% of its window", + rate, window, 100*res.worstCov) + } + if res.holdDeclined > 0 { + t.Errorf("rate %.0f window %.1f s: the zoom hold declined %d of %d captures", + rate, window, res.holdDeclined, res.shots) + } + } + } +} + +// Widening the window mid-run is the gesture the report came from. The fill gate +// holds the first shot off until the ring reaches back far enough, which is +// correct; what it must not do is stay shut, nor leave the trigger losing pulses +// once the ring has caught up. +func TestSporadicYieldRecoversAfterAWindowChange(t *testing.T) { + const key = "s1:Ch1" + for _, w := range []float64{1.0, 2.0, 5.0} { + sim := pulseTrainSim{ + rateHz: 200e3, batchSec: 1.0 / 30.0, + pulsePeriod: w, pulseSamples: 3, + simSec: 30 * w, windowSec: 0.2, prePercent: 20, + holdoffSec: autoRearmDelaySec, armAt: 1.0, + windowChangeAt: 10 * w, windowChangeTo: w, + } + res := sim.run(t, key) + // Count only what happened after the change settled. + after, want := 0, 0 + for _, tt := range res.trigTimes { + if tt > 11*w { + after++ + } + } + for p := 11 * w; p < 30*w; p += w { + want++ + } + if float64(after) < 0.9*float64(want) { + t.Errorf("window 0.2 -> %.1f s: %d captures in the %d pulses after the change", + w, after, want) + } + } +} diff --git a/Common/Client/go/wshub/trigger_test.go b/Common/Client/go/wshub/trigger_test.go index cda4758..e610845 100644 --- a/Common/Client/go/wshub/trigger_test.go +++ b/Common/Client/go/wshub/trigger_test.go @@ -297,9 +297,9 @@ func TestCollectingIsBroadcast(t *testing.T) { // later. It forgets any earlier measurement first, so the rate is the one // asked for rather than a blend with it. func setFill(te *triggerEngine, span, growth, now float64) { - te.setBuffered(0, false, now) - te.setBuffered(span-growth, true, now) - te.setBuffered(span, true, now+1) + te.setBuffered(0, 0, false, false, now) + te.setBuffered(span-growth, 0, false, true, now) + te.setBuffered(span, 0, false, true, now+1) } // What has to hold is that the buffer spans the whole window by the time the @@ -417,9 +417,9 @@ func TestForceIgnoresFillGate(t *testing.T) { // interval, so they refresh the span and leave the seeded rate alone. func seedFillNow(te *triggerEngine, span, growth float64) { now := float64(time.Now().UnixNano()) / 1e9 - te.setBuffered(0, false, now-1) - te.setBuffered(span-growth, true, now-1) - te.setBuffered(span, true, now) + te.setBuffered(0, 0, false, false, now-1) + te.setBuffered(span-growth, 0, false, true, now-1) + te.setBuffered(span, 0, false, true, now) } // While it holds off, the trigger looks identical to one that is ignoring @@ -505,3 +505,55 @@ func drainStates(t *testing.T, h *Hub) []map[string]any { } } } + +// A ring whose coverage saturates below the window (measured source rate that +// over-estimates the true one) can never satisfy the full-window need. The +// coverage clamp must open the gate once the ring is full, delivering a short +// capture rather than staying deaf forever. +func TestFillNeedClampedToCoverage(t *testing.T) { + te := newTriggerEngine() + te.SetConfig(trigConfig{signalKey: "s:x", windowSec: 60, prePercent: 20, mode: "normal", holdoffSec: 0.2}) + setFill(te, 50, 0, 100) // ring full at 50 s, no growth + te.mu.Lock() + te.bufCoverage = 50 // the ring can never reach further back + te.mu.Unlock() + + if need := te.fillNeedLocked(); need != 50 { + t.Errorf("need = %v, want 50 (clamped to coverage, not the 60 s window)", need) + } + if f := te.fillLocked(); f < 1 { + t.Errorf("fillLocked = %v, want >= 1: a full ring below the window must still open the gate", f) + } + + // Without the clamp the gate would stay shut forever. + te.mu.Lock() + te.bufCoverage = 0 + te.mu.Unlock() + if f := te.fillLocked(); f >= 1 { + t.Errorf("baseline: fillLocked = %v, want < 1 without a coverage clamp", f) + } +} + +// When the disk archive already spans the window it can back-fill the front of +// a capture, so the gate must only require the ring to have reached the +// pre-window, not the whole window. +func TestFillNeedArchiveLowersToPreWindow(t *testing.T) { + te := newTriggerEngine() + te.SetConfig(trigConfig{signalKey: "s:x", windowSec: 60, prePercent: 20, mode: "normal", holdoffSec: 0.2}) + setFill(te, 30, 0, 100) // ring holds only 30 s, no growth → need 60 without archive + te.mu.Lock() + te.bufArchived = true + te.mu.Unlock() + + if want := 12.0; te.fillNeedLocked() != want { // 60 * 0.20 + t.Errorf("need = %v, want %v (archive lowers to the pre-window)", te.fillNeedLocked(), want) + } + + // A ring holding just the pre-window opens the gate once archived. + te.mu.Lock() + te.bufSpan = 12 + te.mu.Unlock() + if f := te.fillLocked(); f < 1 { + t.Errorf("fillLocked = %v, want >= 1 with pre-window buffered and the archive available", f) + } +}