From ff5ad224470b87dac65fe89dae01b7a4e654e20e Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 13 Aug 2026 10:28:43 +0200 Subject: [PATCH] included jitter correction on client --- Client/udpstreamer/static/app.js | 352 +++++++++++++++--- Client/udpstreamer/static/index.html | 19 +- Client/udpstreamer/static/style.css | 23 ++ Common/Client/go/wshub/hub.go | 297 ++++++++++----- Source/Applications/StreamHub/StreamHub.cpp | 7 + Source/Applications/StreamHub/StreamHub.h | 1 + .../Applications/StreamHub/TriggerEngine.cpp | 26 ++ Source/Applications/StreamHub/TriggerEngine.h | 11 + 8 files changed, 598 insertions(+), 138 deletions(-) diff --git a/Client/udpstreamer/static/app.js b/Client/udpstreamer/static/app.js index 41f10b6..6473e46 100644 --- a/Client/udpstreamer/static/app.js +++ b/Client/udpstreamer/static/app.js @@ -89,6 +89,15 @@ function getVScale(plotId, key) { return sigVScale[vsKey]; } +// Round a raw units-per-division up to the next 1/2/5×10ⁿ step so the Y-axis +// gridlines land on human-readable values. +function niceDiv(x) { + if (!isFinite(x) || x <= 0) return 1; + const p = Math.pow(10, Math.floor(Math.log10(x))); + const m = x / p; + return (m <= 1 ? 1 : m <= 2 ? 2 : m <= 5 ? 5 : 10) * p; +} + function findSignalMeta(key) { const colon = key.indexOf(':'); if (colon < 0) return null; @@ -107,8 +116,8 @@ function resolveVScale(plotId, key, rawY) { if (vs.mode === 'range') { const meta = findSignalMeta(key); if (meta && meta.rangeMin != null && meta.rangeMax != null && meta.rangeMax > meta.rangeMin) { - const divValue = (meta.rangeMax - meta.rangeMin) / 8; - const offset = (meta.rangeMin + meta.rangeMax) / 2; + const divValue = niceDiv((meta.rangeMax - meta.rangeMin) / 8); + const offset = Math.round((meta.rangeMin + meta.rangeMax) / 2 / divValue) * divValue; vs._resolvedDiv = divValue; vs._resolvedOffset = offset; return { divValue, offset, screenPos }; } @@ -120,7 +129,9 @@ function resolveVScale(plotId, key, rawY) { vs._resolvedDiv = divValue; vs._resolvedOffset = offset; return { divValue, offset, screenPos }; } - // Auto: fit data in central 6 of 8 divisions, centered at screenPos + // Auto: fit data in central 6 of 8 divisions, centered at screenPos. + // Both the V/div and the centre offset are snapped so gridlines (and the + // zero line, when in view) fall on round values. let min = Infinity, max = -Infinity; for (let i = 0; i < rawY.length; i++) { const v = rawY[i]; @@ -128,8 +139,8 @@ function resolveVScale(plotId, key, rawY) { } if (!isFinite(min)) { min = -1; max = 1; } if (min === max) { min -= 1; max += 1; } - const divValue = Math.max((max - min) / 6, 1e-30); - const offset = (max + min) / 2; + const divValue = niceDiv(Math.max((max - min) / 6, 1e-30)); + const offset = Math.round((max + min) / 2 / divValue) * divValue; vs._resolvedDiv = divValue; vs._resolvedOffset = offset; return { divValue, offset, screenPos }; } @@ -285,6 +296,12 @@ let _zoomFetchTimer = null; // trig mode → relative seconds from trigger const cursors = { mode: 'off', tA: null, tB: null }; 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 }; // Layout — [label, cssClass, cols, rows] const LAYOUTS = [ @@ -329,7 +346,15 @@ async function resolveHub() { function connectWS() { ws = new WebSocket('ws://' + HUB + '/ws'); ws.binaryType = 'arraybuffer'; - ws.onopen = () => { wsBackoff = 1000; setStatus('orange', 'Connected – waiting for data'); }; + ws.onopen = () => { + wsBackoff = 1000; + setStatus('orange', 'Connected – waiting for data'); + // Restore monotonic TS preference from localStorage. + const monoPref = localStorage.getItem('udpscope.monotonic') === '1'; + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify({ type: 'setMonotonic', enabled: monoPref })); + } + }; ws.onclose = () => { setStatus('red', 'Disconnected (reconnecting…)'); setTimeout(connectWS, wsBackoff); @@ -346,10 +371,26 @@ function connectWS() { else if (msg.type === 'zoom') onZoomReply(msg); else if (msg.type === 'historyZoom') onHistoryZoomReply(msg); else if (msg.type === 'historyInfo') onHistoryInfo(msg); - else if (msg.type === 'triggerState') onTriggerState(msg); + else if (msg.type === 'monotonicState') onMonotonicState(msg); }; } +/* Monotonic timestamp snapping — when enabled, the hub snaps small inter-frame + timestamp deviations (< 5 ms) to the ideal gap, eliminating overlaps/gaps + caused by software-dispatch jitter. */ +function onMonotonicState(msg) { + const cb = document.getElementById('cb-monotonic'); + if (!cb) return; + cb.checked = !!msg.enabled; + localStorage.setItem('udpscope.monotonic', msg.enabled ? '1' : '0'); +} +document.getElementById('cb-monotonic').addEventListener('change', e => { + localStorage.setItem('udpscope.monotonic', e.target.checked ? '1' : '0'); + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify({ type: 'setMonotonic', enabled: e.target.checked })); + } +}); + /* WS zoom request/reply — replaces the Go hub's /api/zoom HTTP endpoint. Resolves with the {key:{t,v}} signals map; rejects on timeout/closure. */ let _zoomReqId = 0; @@ -882,24 +923,41 @@ function makeSeriesPath(key) { /* ════════════════════════════════════════════════════════════════ LTTB Web Worker — offloads decimation off the main thread. - Cache key: "::::" + Cache key: "::::" when zoomed, or + "::rolling" in live mode, where the data + generation is passed alongside instead of being part of the key. On cache-hit → render uses cached {t, v} immediately (stale-while-revalidate). - On cache-miss → render falls back to sync lttb once (first zoom render only), + On cache-miss → render falls back to sync lttb once (first render only), then worker takes over for subsequent updates. ════════════════════════════════════════════════════════════════ */ -const lttbCache = new Map(); // key → {t, v} -const lttbPending = new Set(); // keys currently in-flight +const lttbCache = new Map(); // key → {t, v, gen} +const lttbPending = new Map(); // key → generation currently in flight +// Hard ceiling on cached decimations. Rolling-mode keys are stable (one entry +// per plot), but zoom keys embed the range, so without a cap the map would grow +// for the whole session. +const LTTB_CACHE_MAX = 256; + +// Store a decimation, re-inserting so Map iteration order stays oldest-first. +function lttbCacheStore(key, entry) { + lttbCache.delete(key); + lttbCache.set(key, entry); + while (lttbCache.size > LTTB_CACHE_MAX) { + lttbCache.delete(lttbCache.keys().next().value); + } +} let _lttbWorker = null; try { _lttbWorker = new Worker('lttb-worker.js'); _lttbWorker.onmessage = function({ data: { id, t, v } }) { + const gen = lttbPending.get(id); lttbPending.delete(id); - lttbCache.set(id, { t, v }); - // Invalidate and redraw the owning plot. + lttbCacheStore(id, { t, v, gen }); + // Invalidate and redraw the owning plot. Clearing lastDataGen defeats the + // render loop's no-new-data fast path so this fresh result is actually drawn. const plotId = parseInt(id.split(':')[0], 10); const p = plots.find(q => q.id === plotId); - if (p) { p.needsRedraw = true; } + if (p) { p.needsRedraw = true; p.lastDataGen = -1; } }; _lttbWorker.onerror = e => console.warn('[lttb-worker] error:', e); } catch(e) { @@ -907,14 +965,20 @@ try { } // Submit a LTTB job to the worker (or run sync if worker unavailable). -// Returns cached {t, v} if fresh, null if a worker job was just submitted, -// or a sync result if the worker is unavailable. -function lttbAsync(cacheKey, t, v, threshold) { +// `gen` identifies the input data behind a key that does not itself change with +// the data (rolling mode). A cached entry computed for an older generation is +// still returned — stale-while-revalidate — while a fresh job runs. Callers +// whose key already encodes the input (zoom ranges) pass no generation. +// Returns the cached {t, v} (fresh or stale), or null on the first render. +function lttbAsync(cacheKey, t, v, threshold, gen) { const cached = lttbCache.get(cacheKey); - if (cached) return cached; // cache hit — use immediately + if (cached && cached.gen === gen) return cached; // fresh — nothing to do + // At most one job per key in flight: submitting on every generation change + // would let the worker's message queue grow without bound whenever it cannot + // keep up with the push rate. if (!lttbPending.has(cacheKey)) { - lttbPending.add(cacheKey); + lttbPending.set(cacheKey, gen); if (_lttbWorker) { // Send copies so the main thread retains the originals. const tCopy = new Float64Array(t); @@ -924,12 +988,13 @@ function lttbAsync(cacheKey, t, v, threshold) { } else { // Synchronous fallback (worker unavailable). const result = lttb(t, v, threshold); - lttbCache.set(cacheKey, result); + result.gen = gen; + lttbCacheStore(cacheKey, result); lttbPending.delete(cacheKey); return result; } } - return null; // worker job in-flight — caller should use fallback + return cached || null; // stale entry, or nothing to draw yet } // Evict stale LTTB cache entries for a plot (call when zoom range changes). @@ -938,7 +1003,7 @@ function lttbCacheEvict(plotId) { for (const k of [...lttbCache.keys()]) { if (k.startsWith(prefix)) lttbCache.delete(k); } - for (const k of [...lttbPending]) { + for (const k of [...lttbPending.keys()]) { if (k.startsWith(prefix)) lttbPending.delete(k); } } @@ -1447,6 +1512,51 @@ function drawCursorLines(u, p) { drawLine(cursors.tB, 'rgba(249,226,175,0.85)', 'B'); } +// Convert a normalized (division) y value to the active signal's raw units. +function rulerRawValue(p, yNorm) { + const key = plotActiveSignal[p.id] || (p.traces.length === 1 ? p.traces[0] : null); + const vs = key ? sigVScale[p.id + ':' + key] : null; + if (!vs) return null; + const dv = vs._resolvedDiv != null ? vs._resolvedDiv : (vs.divValue || 1); + const ofs = vs._resolvedOffset != null ? vs._resolvedOffset : (vs.offset || 0); + return (yNorm - (vs.screenPos || 0)) * dv + ofs; +} + +// Draw the horizontal value rulers (called from the draw hook). +function drawRulerLines(u, p) { + if (rulers.mode !== 'on') return; + const { ctx, bbox } = u; + if (!bbox) return; + + const drawLine = (yNorm, color, label) => { + if (yNorm === null) return; + const y = Math.round(u.valToPos(yNorm, 'y', true)); + if (y < bbox.top || y > bbox.top + bbox.height) return; + ctx.save(); + ctx.beginPath(); + ctx.rect(bbox.left, bbox.top, bbox.width, bbox.height); + ctx.clip(); + ctx.strokeStyle = color; + ctx.lineWidth = 1.5; + ctx.setLineDash([5, 4]); + ctx.beginPath(); + ctx.moveTo(bbox.left, y); + ctx.lineTo(bbox.left + bbox.width, y); + ctx.stroke(); + ctx.setLineDash([]); + ctx.fillStyle = color; + ctx.font = 'bold 11px monospace'; + ctx.textAlign = 'left'; + ctx.textBaseline = 'bottom'; + const raw = p ? rulerRawValue(p, yNorm) : null; + ctx.fillText(label + (raw !== null ? ' ' + _fmtVal(raw) : ''), bbox.left + 4, y - 2); + ctx.restore(); + }; + + drawLine(rulers.yA, 'rgba(166,227,161,0.85)', 'Y1'); + drawLine(rulers.yB, 'rgba(243,139,168,0.85)', 'Y2'); +} + // Compute the rolling-window anchor ("newest common timestamp") for a plot. // Returns the min-of-max timestamp across ACTIVE sources contributing traces to p, // so no live source shows a blank right edge. @@ -1563,7 +1673,7 @@ function makeUPlotOpts(p, inTrigMode) { legend: { show: false }, padding: [4, 4, 0, 0], hooks: { - draw: [u => { drawBandSeparators(u, p); drawActiveSeries(u, p); drawOffsetMarkers(u, p); drawCursorLines(u, p); drawSeriesMarkers(u, p); drawTriggerMarker(u, p); }], + draw: [u => { drawBandSeparators(u, p); drawActiveSeries(u, p); drawOffsetMarkers(u, p); drawCursorLines(u, p); drawRulerLines(u, p); drawSeriesMarkers(u, p); drawTriggerMarker(u, p); }], // Two-hook zoom detection: setSelect flags that the NEXT setScale is user-initiated. // uPlot fires setSelect → then immediately setScale (when drag.setScale:true). // All programmatic setScale calls happen without a preceding setSelect, so the @@ -1617,34 +1727,60 @@ function createUPlot(p) { return min + pct * (max - min); } - // Update pointer style based on what's under the mouse + function _rulerAtClientY(clientY) { + 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'; + return null; + } + + function _rulerValFromEvent(e) { + const rect = p.uplot.over.getBoundingClientRect(); + const pct = Math.max(0, Math.min(1, (e.clientY - rect.top) / rect.height)); + const { min, max } = p.uplot.scales.y; + return max - pct * (max - min); + } + + // Update pointer style based on what's under the mouse, and drive the + // time/value hover readout. p.uplot.over.addEventListener('mousemove', e => { - const snap = cursors.mode === 'on' ? _cursorAtClientX(e.clientX) : null; - p.uplot.over.style.cursor = snap ? 'ew-resize' : ''; + const snapX = cursors.mode === 'on' ? _cursorAtClientX(e.clientX) : null; + const snapY = !snapX && rulers.mode === 'on' ? _rulerAtClientY(e.clientY) : null; + p.uplot.over.style.cursor = snapX ? 'ew-resize' : (snapY ? 'ns-resize' : ''); + showHoverReadout(p, e); }); p.uplot.over.addEventListener('mouseleave', () => { p.uplot.over.style.cursor = ''; + hideHoverReadout(); }); // Mousedown: drag an existing cursor (only when mode='on' and mouse is near a cursor line). // If not near a cursor, the event falls through to uPlot for normal zoom/pan behavior. p.uplot.over.addEventListener('mousedown', e => { if (e.button !== 0 || e.shiftKey) return; // shift is pan - if (cursors.mode !== 'on') return; - const target = _cursorAtClientX(e.clientX); - if (!target) return; // not near a cursor — let uPlot handle zoom + const target = cursors.mode === 'on' ? _cursorAtClientX(e.clientX) : null; + const yTarget = !target && rulers.mode === 'on' ? _rulerAtClientY(e.clientY) : null; + if (!target && !yTarget) return; // not near a cursor — let uPlot handle zoom e.stopImmediatePropagation(); // prevent uPlot drag-zoom e.preventDefault(); // Set cursor position immediately on mousedown - if (target === 'A') cursors.tA = _cursorValFromEvent(e); + if (yTarget) { + if (yTarget === 'A') rulers.yA = _rulerValFromEvent(e); + else rulers.yB = _rulerValFromEvent(e); + } else if (target === 'A') cursors.tA = _cursorValFromEvent(e); else cursors.tB = _cursorValFromEvent(e); updateCursorReadout(); cursorsDirty = true; const onMove = ev => { - if (target === 'A') cursors.tA = _cursorValFromEvent(ev); + if (yTarget) { + if (yTarget === 'A') rulers.yA = _rulerValFromEvent(ev); + else rulers.yB = _rulerValFromEvent(ev); + } else if (target === 'A') cursors.tA = _cursorValFromEvent(ev); else cursors.tB = _cursorValFromEvent(ev); updateCursorReadout(); cursorsDirty = true; @@ -1871,10 +2007,12 @@ function buildLiveData(p) { // the full window slice can easily reach 100k–300k pts — far more than uPlot // needs for a 1200px-wide canvas. Always run LTTB via the background worker // (stale-while-revalidate: use cached result; fall back to sync on first render). - // Rolling-mode cache key uses _dataGen so the result refreshes on new data. + // The rolling-mode key is constant per (plot, signal) — the data generation is + // carried separately so the cache holds one entry per plot instead of one per + // push tick, which used to grow without bound for the whole session. const targetPts = Math.max(LTTB_MIN, ((p.uplot ? p.uplot.width : p.div.clientWidth) || 600) * 2); const cacheKey = isRolling - ? `${p.id}:${masterKey}:rolling:${_dataGen}` + ? `${p.id}:${masterKey}:rolling` : `${p.id}:${masterKey}:${t0.toFixed(6)}:${t1.toFixed(6)}:${masterRaw.t.length}`; let sharedT, masterV; if (masterRaw.t.length <= targetPts) { @@ -1882,7 +2020,8 @@ function buildLiveData(p) { sharedT = masterRaw.t; masterV = masterRaw.v; } else { - const cached = lttbAsync(cacheKey, masterRaw.t, masterRaw.v, targetPts); + const cached = lttbAsync(cacheKey, masterRaw.t, masterRaw.v, targetPts, + isRolling ? _dataGen : undefined); let dec; if (cached) { dec = cached; @@ -2094,19 +2233,10 @@ document.getElementById('btn-zoom-fit').addEventListener('click', zoomFit); /* ════════════════════════════════════════════════════════════════ Cursor controls ════════════════════════════════════════════════════════════════ */ -// Show the cursor button only when paused or in trigger-snapshot mode. +// Cursors are always available — in live rolling mode they are pinned to the +// moving viewport by the render loop. function updateCursorBtnVisibility() { - const canUseCursors = globalPause || (trig.enabled && trig.snapshot !== null); - const btn = document.getElementById('btn-cursor'); - btn.style.display = canUseCursors ? '' : 'none'; - if (!canUseCursors && cursors.mode !== 'off') { - cursors.mode = 'off'; - cursors.tA = null; cursors.tB = null; // context changed, clear positions - btn.textContent = 'Cursors'; - btn.classList.remove('active'); - document.getElementById('cursor-readout').classList.remove('visible'); - cursorsDirty = true; - } + document.getElementById('btn-cursor').style.display = ''; } document.getElementById('btn-cursor').addEventListener('click', () => { @@ -2133,6 +2263,18 @@ document.getElementById('btn-cursor').addEventListener('click', () => { cursorsDirty = true; }); +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; + } + updateCursorReadout(); + cursorsDirty = true; +}); + // Format a signal value for the per-plot cursor readout. function fmtVal(v) { if (v === null || v === undefined) return '—'; @@ -2178,11 +2320,80 @@ function updatePlotCursorReadouts() { }); } +/* ─── Hover readout ──────────────────────────────────────────────────────── */ +// Un-normalize a plotted value of trace `key` in plot `p` back to raw units. +function rawFromNorm(p, key, vNorm) { + 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 - (vs.screenPos || 0)) * dv + ofs; +} + +function hideHoverReadout() { + document.getElementById('hover-readout').style.display = 'none'; +} + +// Show the time under the mouse plus every trace's value at that time. +function showHoverReadout(p, e) { + const el = document.getElementById('hover-readout'); + if (!p.uplot || p.traces.length === 0) { el.style.display = 'none'; return; } + + const rect = p.uplot.over.getBoundingClientRect(); + const { min, max } = p.uplot.scales.x; + const pct = Math.max(0, Math.min(1, (e.clientX - rect.left) / rect.width)); + const t = min + pct * (max - min); + const span = Math.abs(max - min); + + const tStr = (trig.enabled && trig.snapshot) ? 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; + const val = vNorm === null ? '—' : _fmtVal(rawFromNorm(p, key, vNorm)); + html += '
' + + '' + escHtml(name) + '' + + '' + escHtml(val) + '
'; + }); + el.innerHTML = html; + el.style.display = 'block'; + + // Keep the tooltip inside the viewport. + const w = el.offsetWidth, h = el.offsetHeight; + let x = e.clientX + 14, y = e.clientY + 14; + if (x + w > window.innerWidth - 4) x = e.clientX - w - 14; + if (y + h > window.innerHeight - 4) y = e.clientY - h - 14; + el.style.left = Math.max(4, x) + 'px'; + 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. +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); + document.getElementById('cur-y1').textContent = 'Y1: ' + fmtVal(vA); + document.getElementById('cur-y2').textContent = 'Y2: ' + fmtVal(vB); + document.getElementById('cur-dy').textContent = + 'ΔY: ' + fmtVal(vA !== null && vB !== null ? vB - vA : null); +} + function updateCursorReadout() { const ro = document.getElementById('cursor-readout'); const active = cursors.mode === 'on'; - ro.classList.toggle('visible', active); + ro.classList.toggle('visible', active || rulers.mode === 'on'); updatePlotCursorReadouts(); + updateRulerReadout(); + ['cur-ta', 'cur-tb', 'cur-dt'].forEach(id => { + document.getElementById(id).style.display = active ? '' : 'none'; + }); if (!active) return; // Use the current visible x-range to pick the display unit. @@ -2286,6 +2497,12 @@ document.getElementById('trig-mode').addEventListener('change', e => { } }); document.getElementById('btn-trig-rearm').addEventListener('click', () => { if (trig.enabled) trigArm(); }); +// Force: capture the window around the newest sample regardless of the threshold. +document.getElementById('btn-trig-force').addEventListener('click', () => { + if (!trig.enabled || !trig.signal) return; + sendTrigConfig(); + wsSend({ type: 'forceTrigger' }); +}); document.getElementById('btn-trig-stop').addEventListener('click', () => { if (!trig.enabled || trig.mode !== 'normal') return; trig.stopped = !trig.stopped; @@ -2943,6 +3160,22 @@ function renderDirtyPlots() { }); } + // Live rolling mode: pin the cursors to the moving viewport so they stay put + // on screen instead of scrolling off the left edge as time advances. + if (cursors.mode === 'on' && !trig.enabled && !globalPause && + plots.some(p => p.uplot && !p.xRange && p.traces.length > 0)) { + if (_cursorAnchorNow !== null && globalPlotNow !== _cursorAnchorNow) { + const shift = globalPlotNow - _cursorAnchorNow; + if (cursors.tA !== null) cursors.tA += shift; + if (cursors.tB !== null) cursors.tB += shift; + cursorsDirty = true; + updateCursorReadout(); + } + _cursorAnchorNow = globalPlotNow; + } else { + _cursorAnchorNow = null; + } + // Fast path: cursor-only redraw (no data rebuild needed) if (cursorsDirty) { cursorsDirty = false; @@ -3204,11 +3437,15 @@ function showVScaleMenu(key, plotId) { const isManual = vs.mode === 'manual'; document.getElementById('vscale-manual-row').style.display = isManual ? 'flex' : 'none'; + document.getElementById('vscale-offset-row').style.display = isManual ? 'flex' : 'none'; document.getElementById('vscale-pos-row').style.display = isManual ? 'flex' : 'none'; - // Pre-fill V/div with resolved or stored value; Position always shows current screenPos. + // Pre-fill V/div and Offset with the resolved or stored values; Position + // always shows the current screenPos. const dv = isManual ? vs.divValue : (vs._resolvedDiv || 1); + const ofs = isManual ? (vs.offset || 0) : (vs._resolvedOffset || 0); document.getElementById('vscale-vdiv').value = dv != null ? parseFloat(dv.toPrecision(4)) : 1; + document.getElementById('vscale-offset').value = parseFloat(ofs.toPrecision(6)); document.getElementById('vscale-pos').value = parseFloat((vs.screenPos || 0).toPrecision(4)); // Type row (Analog/Digital) only shown in mixed mode. @@ -3293,8 +3530,9 @@ function initVScaleMenu() { if (newMode === 'manual' && vs.mode !== 'manual') { // Seed V/div from currently resolved value; screenPos stays as-is. vs.divValue = vs._resolvedDiv || 1; - vs.offset = vs._resolvedOffset || 0; // keep for DC subtraction (internal) + vs.offset = vs._resolvedOffset || 0; // raw value at screen centre document.getElementById('vscale-vdiv').value = parseFloat(vs.divValue.toPrecision(4)); + document.getElementById('vscale-offset').value = parseFloat(vs.offset.toPrecision(6)); document.getElementById('vscale-pos').value = parseFloat((vs.screenPos || 0).toPrecision(4)); } vs.mode = newMode; @@ -3302,6 +3540,7 @@ function initVScaleMenu() { btn.classList.add('active'); const isManual = vs.mode === 'manual'; document.getElementById('vscale-manual-row').style.display = isManual ? 'flex' : 'none'; + document.getElementById('vscale-offset-row').style.display = isManual ? 'flex' : 'none'; document.getElementById('vscale-pos-row').style.display = isManual ? 'flex' : 'none'; refreshPlotForKey(_vsMenuKey); }); @@ -3312,6 +3551,16 @@ function initVScaleMenu() { vs.divValue = Math.max(parseFloat(e.target.value) || 1, 1e-30); refreshPlotForKey(_vsMenuKey); }); + // "Offset" is the raw value shown at screen centre. Unlike Position (which + // is clamped to the ±4 visible divisions) it is unbounded, so a signal can be + // referenced to a level far outside the currently plotted range. + document.getElementById('vscale-offset').addEventListener('input', e => { + if (!_vsMenuKey) return; + const vs = getVScale(_vsMenuPlotId, _vsMenuKey); + const v = parseFloat(e.target.value); + vs.offset = isFinite(v) ? v : 0; + refreshPlotForKey(_vsMenuKey); + }); // "Position (div)" moves the marker and signal together on screen. document.getElementById('vscale-pos').addEventListener('input', e => { if (!_vsMenuKey) return; @@ -3551,6 +3800,11 @@ buildSidebar(); // show "Add Source" section even before WS connection initArrayIdxPicker(); initVScaleMenu(); initSignalMenu(); +// Restore monotonic TS checkbox from localStorage (visual state before WS reply). +{ + const cb = document.getElementById('cb-monotonic'); + if (cb) cb.checked = localStorage.getItem('udpscope.monotonic') === '1'; +} document.getElementById('btn-csv-all').addEventListener('click', exportAllCSV); document.getElementById('btn-stats').addEventListener('click', toggleStats); document.getElementById('btn-stats-close').addEventListener('click', toggleStats); diff --git a/Client/udpstreamer/static/index.html b/Client/udpstreamer/static/index.html index 76150bc..439bc53 100644 --- a/Client/udpstreamer/static/index.html +++ b/Client/udpstreamer/static/index.html @@ -21,6 +21,12 @@ A: —│ B: —│ ΔT: — + Window: - + + +
@@ -83,6 +93,7 @@
IDLE +
@@ -185,6 +196,10 @@
+ + + \ No newline at end of file diff --git a/Client/udpstreamer/static/style.css b/Client/udpstreamer/static/style.css index 670ae30..e6658a3 100644 --- a/Client/udpstreamer/static/style.css +++ b/Client/udpstreamer/static/style.css @@ -52,6 +52,23 @@ html, body { height:100%; background:var(--bg); color:var(--text); #cursor-readout.visible { display:flex; } #cur-ta { color:var(--sky); } #cur-tb { color:var(--yellow); } #cur-dt { color:var(--subtext1); } .cur-sep { color:var(--surface2); } +#ruler-readout { display:inline-flex; align-items:center; gap:8px; } +#cur-y1 { color:var(--green); } #cur-y2 { color:var(--red); } +#cur-dy { color:var(--subtext1); } + +/* Mouse-over time/value tooltip */ +#hover-readout { + position:fixed; z-index:60; pointer-events:none; + background:var(--surface0); border:1px solid var(--surface1); + border-radius:5px; padding:4px 8px; + font-size:11px; font-family:monospace; white-space:nowrap; + box-shadow:0 4px 12px rgba(0,0,0,0.45); +} +#hover-readout .hov-time { color:var(--subtext1); margin-bottom:3px; } +#hover-readout .hov-row { display:flex; align-items:center; gap:6px; } +#hover-readout .hov-dot { width:8px; height:8px; border-radius:50%; flex-shrink:0; } +#hover-readout .hov-name { color:var(--subtext0); } +#hover-readout .hov-val { color:var(--text); margin-left:auto; padding-left:10px; } .topbar-vsep { width:1px; height:22px; background:var(--surface0); flex-shrink:0; margin:0 2px; } #layout-btns { display:flex; gap:2px; align-items:center; flex-shrink:0; } @@ -74,6 +91,12 @@ button.ctrl-btn.trig-active { background:rgba(203,166,247,0.15); border-color:va button.ctrl-btn.cursor-a { border-color:var(--sky); color:var(--sky); } button.ctrl-btn.cursor-b { border-color:var(--yellow); color:var(--yellow); } button.ctrl-btn.resume-btn { border-color:var(--teal); color:var(--teal); } +label.ctrl-check { + display:flex; align-items:center; gap:4px; flex-shrink:0; + font-size:12px; color:var(--subtext0); cursor:pointer; white-space:nowrap; +} +label.ctrl-check input { margin:0; cursor:pointer; accent-color:var(--accent); } +label.ctrl-check:has(input:checked) { color:var(--accent); } /* ── Trigger bar ──────────────────────────────────────────────── */ #trigbar { diff --git a/Common/Client/go/wshub/hub.go b/Common/Client/go/wshub/hub.go index 2f03b25..833ffa3 100644 --- a/Common/Client/go/wshub/hub.go +++ b/Common/Client/go/wshub/hub.go @@ -107,7 +107,18 @@ func (c *wsClient) readPump() { case c.hub.commandCh <- hubCmd{op: "wsSaveSources"}: default: } + case "setMonotonic": + enabled, _ := env["enabled"].(bool) + select { + case c.hub.commandCh <- hubCmd{op: "setMonotonic", enabled: enabled}: + default: + } + case "zoom": + c.hub.handleWSZoom(c, env) default: + if c.hub.handleTriggerCommand(t, env) { + break + } // Unrecognized message type — forward to DebugCh select { case c.hub.DebugCh <- msg: @@ -182,6 +193,14 @@ type sourceHubState struct { // per signal name. Used by the default (TimeModePacket, n>1) path to estimate // per-element dt when only one packet arrives in a 30 Hz tick. lastPktNs map[string]int64 + + // Monotonic timestamp snapping state (all accessed from Run() goroutine): + // lastFrameMeasured — uncorrected measured anchor of the previous frame. + // lastFrameEndT — corrected anchor after snapping. + // gapEMA — exponential moving average of the measured inter-frame gap. + lastFrameMeasured map[string]float64 + lastFrameEndT map[string]float64 + gapEMA map[string]float64 } // taggedSample is a DataSample annotated with its source ID. @@ -192,7 +211,7 @@ type taggedSample struct { // hubCmd carries a command to the Run() goroutine. type hubCmd struct { - op string // "addSource","removeSource","setSourceState","updateConfig", + op string // "addSource","removeSource","setSourceState","updateConfig", // "wsAddSource","wsRemoveSource","wsSaveSources" sourceID string label string @@ -201,6 +220,7 @@ type hubCmd struct { sigs []udpsprotocol.SignalInfo multicastGroup string dataPort int + enabled bool // "setMonotonic" toggle } // Hub is the central broker between UDP clients and WebSocket clients. @@ -224,21 +244,17 @@ type Hub struct { ringsMu sync.RWMutex rings map[string]*sigRing // "sourceId:signalKey" → ring - // lastZoomAt tracks the last time a zoom request was served. - // Ring buffer writes are skipped when no zoom has been requested - // in the last 10 s, saving substantial CPU on LTTB + ring writes. - lastZoomAt time.Time - zoomAtMu sync.Mutex - statsMu sync.RWMutex statsMap map[string]*SourceStat - // onClientConnect, if set, is called each time a new WebSocket client - // registers. The callback receives a send function that delivers a message - // directly to that client. It is invoked synchronously from Run(), so it - // must not block. + // trigger is the hub-side trigger FSM driving the oscilloscope capture mode. + trigger *triggerEngine onClientConnectMu sync.RWMutex onClientConnect func(send func([]byte)) + + // monotonicTS, when true, snaps small inter-frame timestamp deviations + // (< monotonicTolerance) to the ideal gap to eliminate jitter. + monotonicTS bool } // NewHub creates an initialised Hub. @@ -253,6 +269,7 @@ func NewHub() *Hub { DebugCh: make(chan []byte, 256), rings: make(map[string]*sigRing), statsMap: make(map[string]*SourceStat), + trigger: newTriggerEngine(), } } @@ -278,44 +295,9 @@ func (h *Hub) getRing(key string) *sigRing { return rb } -// shouldWriteRing returns true if zoom was requested within the last 10 seconds. -func (h *Hub) shouldWriteRing() bool { - h.zoomAtMu.Lock() - ok := time.Since(h.lastZoomAt) < 10*time.Second - h.zoomAtMu.Unlock() - return ok -} - -// HandleZoom serves GET /api/zoom?... It also records the access time -// so the ring buffer knows zoom is active and worth populating. -func (h *Hub) HandleZoom(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 n int - if nStr := q.Get("n"); nStr == "" { - n = 2400 - } else { - n, _ = strconv.Atoi(nStr) - if n <= 0 { - n = 1 << 30 // no decimation - } else if n < 10 { - n = 2400 - } - } - - if n > 0 { - h.zoomAtMu.Lock() - h.lastZoomAt = time.Now() - h.zoomAtMu.Unlock() - } - - keys := strings.Split(q.Get("signals"), ",") - +// zoomSlice extracts [t0, t1] from the full-resolution rings for the named +// signals, decimating each to at most n points. +func (h *Hub) zoomSlice(t0, t1 float64, keys []string, n int) map[string]sigData { h.ringsMu.RLock() refs := make(map[string]*sigRing, len(keys)) for _, k := range keys { @@ -338,11 +320,69 @@ func (h *Hub) HandleZoom(w http.ResponseWriter, r *http.Request) { dt, dv := lttbDecimate(rt, rv, n) result[k] = sigData{T: dt, V: dv} } + return result +} + +// zoomPoints normalises the client's requested point budget: absent → 2400, +// non-positive → every sample in the range, implausibly small → 2400. +func zoomPoints(n int, present bool) int { + switch { + case !present: + return 2400 + case n <= 0: + return 1 << 30 // no decimation + case n < 10: + return 2400 + } + return n +} + +// handleWSZoom answers a browser {"type":"zoom","reqId":..,"t0":..,"t1":.., +// "n":..,"signals":"a,b"} request, unicasting {"type":"zoom","reqId":.., +// "signals":{...}} back to the requesting client. This is the path the web SPA +// actually uses; /api/zoom is the equivalent HTTP entry point. +func (h *Hub) handleWSZoom(c *wsClient, env map[string]interface{}) { + t0, ok0 := env["t0"].(float64) + t1, ok1 := env["t1"].(float64) + if !ok0 || !ok1 || t1 <= t0 { + return + } + nF, nOK := env["n"].(float64) + n := zoomPoints(int(nF), nOK) + sigCSV, _ := env["signals"].(string) + + reply, err := json.Marshal(map[string]any{ + "type": "zoom", + "reqId": env["reqId"], + "signals": h.zoomSlice(t0, t1, strings.Split(sigCSV, ","), n), + }) + if err != nil { + log.Printf("hub: ws zoom encode: %v", err) + return + } + select { + case c.send <- wsMessage{websocket.TextMessage, reply}: + default: + } +} + +// HandleZoom serves GET /api/zoom?... +func (h *Hub) HandleZoom(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 + } + nStr := q.Get("n") + nVal, _ := strconv.Atoi(nStr) + n := zoomPoints(nVal, nStr != "") w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(map[string]any{ "type": "zoom", - "signals": result, + "signals": h.zoomSlice(t0, t1, strings.Split(q.Get("signals"), ","), n), }); err != nil { log.Printf("hub: zoom encode: %v", err) } @@ -455,13 +495,28 @@ func (h *Hub) Run() { h.clients[c] = true // Send current state to the new client. if sourcesMsg != nil { - select { case c.send <- wsMessage{websocket.TextMessage, sourcesMsg}: default: } + select { + case c.send <- wsMessage{websocket.TextMessage, sourcesMsg}: + default: + } } for _, src := range sourcesMap { if src.configJS != nil { - select { case c.send <- wsMessage{websocket.TextMessage, src.configJS}: default: } + select { + case c.send <- wsMessage{websocket.TextMessage, src.configJS}: + default: + } } } + select { + case c.send <- wsMessage{websocket.TextMessage, h.trigger.stateMsg()}: + default: + } + monoMsg, _ := json.Marshal(map[string]any{"type": "monotonicState", "enabled": h.monotonicTS}) + select { + case c.send <- wsMessage{websocket.TextMessage, monoMsg}: + default: + } // Notify the application layer so it can replay any persistent state // (e.g., MARTe2 connection status, forced/traced signals). h.onClientConnectMu.RLock() @@ -469,7 +524,10 @@ func (h *Hub) Run() { h.onClientConnectMu.RUnlock() if fn != nil { fn(func(msg []byte) { - select { case c.send <- wsMessage{websocket.TextMessage, msg}: default: } + select { + case c.send <- wsMessage{websocket.TextMessage, msg}: + default: + } }) } @@ -481,19 +539,25 @@ func (h *Hub) Run() { case msg := <-h.broadcastCh: for c := range h.clients { - select { case c.send <- wsMessage{websocket.TextMessage, msg}: default: } + select { + case c.send <- wsMessage{websocket.TextMessage, msg}: + default: + } } case cmd := <-h.commandCh: switch cmd.op { case "addSource": sourcesMap[cmd.sourceID] = &sourceHubState{ - id: cmd.sourceID, - label: cmd.label, - addr: cmd.addr, - connState: "connecting", - timeSigCalib: make(map[string]float64), - lastPktNs: make(map[string]int64), + id: cmd.sourceID, + label: cmd.label, + addr: cmd.addr, + connState: "connecting", + timeSigCalib: make(map[string]float64), + lastPktNs: make(map[string]int64), + lastFrameEndT: make(map[string]float64), + lastFrameMeasured: make(map[string]float64), + gapEMA: make(map[string]float64), } h.statsMu.Lock() h.statsMap[cmd.sourceID] = &SourceStat{} @@ -529,6 +593,7 @@ func (h *Hub) Run() { } src.signals = cmd.sigs src.configSeq++ + src.lastFrameEndT = make(map[string]float64) cfgMsg, err := json.Marshal(map[string]any{ "type": "config", "sourceId": cmd.sourceID, @@ -581,6 +646,10 @@ func (h *Hub) Run() { log.Printf("hub: save sources: %v", err) } } + case "setMonotonic": + h.monotonicTS = cmd.enabled + monoMsg, _ := json.Marshal(map[string]any{"type": "monotonicState", "enabled": h.monotonicTS}) + h.broadcast(monoMsg) } case ts := <-h.dataCh: @@ -607,6 +676,7 @@ func (h *Hub) Run() { } } } + h.triggerTick() case <-statsTicker.C: h.statsMu.RLock() @@ -640,11 +710,26 @@ func writeFloat64s(buf []byte, off int, f []float64) int { // ─── Data serialisation ─────────────────────────────────────────────────────── +// maxPushPoints bounds the live push only. The zoom rings deliberately store +// every sample: decimating on the way in would cap the resolution a zoom can +// ever recover, and the browser already decimates for display. const maxPushPoints = 50 -const maxRingPoints = 20_000 + +// Zoom ring depth, in samples per signal (16 bytes each). ringCapTemporal +// holds 6 s of a 1 MSps waveform; ringCapScalar holds 100 000 packets. const ringCapTemporal = 6_000_000 const ringCapScalar = 100_000 +// monotonicTolerance is the maximum inter-frame timestamp deviation (seconds) +// treated as jitter and snapped to the ideal gap. Larger deviations are +// preserved as genuine discontinuities (missing frames, rate changes). +const monotonicTolerance = 0.005 // 5 ms + +// monotonicEMAAlpha is the smoothing factor for the inter-frame gap EMA. +// 0.01 gives a time constant of ~100 frames (~1 s at 100 Hz): fast enough to +// track real rate changes, slow enough to average out per-frame jitter. +const monotonicEMAAlpha = 0.01 + // lttbDecimate reduces (tIn, vIn) to at most threshold representative points // using the Largest-Triangle-Three-Buckets algorithm. func lttbDecimate(tIn, vIn []float64, threshold int) ([]float64, []float64) { @@ -667,10 +752,13 @@ func lttbDecimate(tIn, vIn []float64, threshold int) ([]float64, []float64) { } avgT, avgV, cnt := 0.0, 0.0, 0 for j := avgS; j < avgE; j++ { - avgT += tIn[j]; avgV += vIn[j]; cnt++ + avgT += tIn[j] + avgV += vIn[j] + cnt++ } if cnt > 0 { - avgT /= float64(cnt); avgV /= float64(cnt) + avgT /= float64(cnt) + avgV /= float64(cnt) } rS := int(float64(i)*every) + 1 rE := int(float64(i+1)*every) + 1 @@ -682,7 +770,8 @@ func lttbDecimate(tIn, vIn []float64, threshold int) ([]float64, []float64) { for j := rS; j < rE; j++ { area := math.Abs((aT-avgT)*(vIn[j]-aV) - (aT-tIn[j])*(avgV-aV)) if area > maxArea { - maxArea = area; next = j + maxArea = area + next = j } } outT[i+1], outV[i+1] = tIn[next], vIn[next] @@ -710,11 +799,13 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []udpsp if src.configSeq != src.configSeqAtCalib { src.configSeqAtCalib = src.configSeq src.timeSigCalib = make(map[string]float64) + src.lastFrameEndT = make(map[string]float64) + src.lastFrameMeasured = make(map[string]float64) + src.gapEMA = make(map[string]float64) } sigs := src.signals pfx := src.id + ":" - writeRing := h.shouldWriteRing() type pairBuf struct { t, v []float64 @@ -766,6 +857,25 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []udpsp anchorTime = float64(s.WallTime.UnixNano()) / 1e9 anchorIsFirstSample = false } + if h.monotonicTS && dt > 0 { + nominalGap := float64(n) * dt + measuredAnchor := anchorTime + if prevMeasured, ok := src.lastFrameMeasured[sig.Name]; ok { + measuredGap := measuredAnchor - prevMeasured + prevEMA, hasEMA := src.gapEMA[sig.Name] + if !hasEMA { + prevEMA = nominalGap + } + src.gapEMA[sig.Name] = prevEMA*(1-monotonicEMAAlpha) + measuredGap*monotonicEMAAlpha + smoothedGap := src.gapEMA[sig.Name] + deviation := math.Abs(measuredGap - smoothedGap) + if deviation > 0 && deviation < monotonicTolerance { + anchorTime = src.lastFrameEndT[sig.Name] + smoothedGap + } + } + src.lastFrameMeasured[sig.Name] = measuredAnchor + src.lastFrameEndT[sig.Name] = anchorTime + } for k := 0; k < n; k++ { var t float64 if anchorIsFirstSample { @@ -777,12 +887,10 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []udpsp allV = append(allV, vals[k]) } } - if writeRing { - ringT, ringV := lttbDecimate(allT, allV, maxRingPoints) - if rb := h.getRing(pfx + sig.Name); rb != nil { - rb.write(ringT, ringV) - } + if rb := h.getRing(pfx + sig.Name); rb != nil { + rb.write(allT, allV) } + h.trigger.feed(pfx+sig.Name, n, allT, allV) decimT, decimV := lttbDecimate(allT, allV, maxPushPoints) pairs[sig.Name] = pairBuf{t: decimT, v: decimV} @@ -825,12 +933,10 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []udpsp allV = append(allV, vals[k]) } } - if writeRing { - ringT, ringV := lttbDecimate(allT, allV, maxRingPoints) - if rb := h.getRing(pfx + sig.Name); rb != nil { - rb.write(ringT, ringV) - } + if rb := h.getRing(pfx + sig.Name); rb != nil { + rb.write(allT, allV) } + h.trigger.feed(pfx+sig.Name, n, allT, allV) decimT, decimV := lttbDecimate(allT, allV, maxPushPoints) pairs[sig.Name] = pairBuf{t: decimT, v: decimV} @@ -845,25 +951,22 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []udpsp ts = append(ts, float64(s.WallTime.UnixNano())/1e9) vs = append(vs, vals[0]) } - if writeRing { - if rb := h.getRing(pfx + sig.Name); rb != nil { - rb.write(ts, vs) - } + if rb := h.getRing(pfx + sig.Name); rb != nil { + rb.write(ts, vs) } + h.trigger.feed(pfx+sig.Name, 1, ts, vs) pairs[sig.Name] = pairBuf{t: ts, v: vs} default: // n > 1, TimeModePacket: C++ sends samplingRate=0 so we interpolate // per-element timestamps from wall-clock differences between packets. // - // Three fixes vs the naïve approach: + // Two fixes vs the naïve approach: // 1. Use src.lastPktNs[name] for the single-packet case so dt is // estimated from the actual inter-packet gap, not 1/n. // 2. Send all n elements to the browser without LTTB so sinusoidal // waveforms are not degraded (packets arrive at ≤30 Hz, bandwidth // is trivially acceptable). - // 3. Always write the ring buffer regardless of shouldWriteRing() so - // the first zoom request immediately returns full-resolution data. allT := make([]float64, 0, len(batch)*n) allV := make([]float64, 0, len(batch)*n) for bi, s := range batch { @@ -876,19 +979,38 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []udpsp var dtSec float64 if bi+1 < len(batch) { // Two consecutive packets in this tick → exact dt. - dtSec = (float64(batch[bi+1].WallTime.UnixNano())-float64(wallNs))/1e9/float64(n) + dtSec = (float64(batch[bi+1].WallTime.UnixNano()) - float64(wallNs)) / 1e9 / float64(n) } else if bi > 0 { // Last of multiple packets → use diff from previous. - dtSec = (float64(wallNs)-float64(batch[bi-1].WallTime.UnixNano()))/1e9/float64(n) + dtSec = (float64(wallNs) - float64(batch[bi-1].WallTime.UnixNano())) / 1e9 / float64(n) } else if prevNs, ok2 := src.lastPktNs[sig.Name]; ok2 && prevNs > 0 && wallNs > prevNs { // Single packet this tick → gap from the previous tick's packet. - dtSec = (float64(wallNs)-float64(prevNs))/1e9/float64(n) + dtSec = (float64(wallNs) - float64(prevNs)) / 1e9 / float64(n) } else { // Truly first packet ever — inter-packet timing unknown. // Skip to avoid poisoning the ring with wrongly-spaced timestamps; // lastPktNs will be recorded below so the next packet uses correct dt. continue } + if h.monotonicTS && dtSec > 0 { + nominalGap := float64(n) * dtSec + measuredStart := wallSec + if prevMeasured, ok := src.lastFrameMeasured[sig.Name]; ok { + measuredGap := measuredStart - prevMeasured + prevEMA, hasEMA := src.gapEMA[sig.Name] + if !hasEMA { + prevEMA = nominalGap + } + src.gapEMA[sig.Name] = prevEMA*(1-monotonicEMAAlpha) + measuredGap*monotonicEMAAlpha + smoothedGap := src.gapEMA[sig.Name] + deviation := math.Abs(measuredGap - smoothedGap) + if deviation > 0 && deviation < monotonicTolerance { + wallSec = src.lastFrameEndT[sig.Name] + smoothedGap + } + } + src.lastFrameMeasured[sig.Name] = measuredStart + src.lastFrameEndT[sig.Name] = wallSec + } for j := 0; j < n; j++ { allT = append(allT, wallSec+float64(j)*dtSec) allV = append(allV, vals[j]) @@ -898,11 +1020,10 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []udpsp src.lastPktNs[sig.Name] = batch[len(batch)-1].WallTime.UnixNano() } if len(allT) > 0 { - // Ring: always populate (fix 3), LTTB only if it actually reduces size. - ringT, ringV := lttbDecimate(allT, allV, maxRingPoints) if rb := h.getRing(pfx + sig.Name); rb != nil { - rb.write(ringT, ringV) + rb.write(allT, allV) } + h.trigger.feed(pfx+sig.Name, n, allT, allV) // Live push: send all points without LTTB (fix 2). pairs[sig.Name] = pairBuf{t: allT, v: allV} } diff --git a/Source/Applications/StreamHub/StreamHub.cpp b/Source/Applications/StreamHub/StreamHub.cpp index 3f7ca4a..087eff3 100644 --- a/Source/Applications/StreamHub/StreamHub.cpp +++ b/Source/Applications/StreamHub/StreamHub.cpp @@ -750,6 +750,7 @@ void StreamHub::OnWSCommand(const char *json, uint32 /*len*/, uint32 slotIdx) { else if (strcmp(type, "rearm") == 0) { HandleRearm(); } else if (strcmp(type, "trigStop") == 0) { HandleTrigStop(json); } else if (strcmp(type, "setTrigger") == 0) { HandleSetTrigger(json); } + else if (strcmp(type, "forceTrigger") == 0) { HandleForceTrigger(); } else if (strcmp(type, "zoom") == 0) { HandleZoom(json, slotIdx); } else if (strcmp(type, "historyZoom") == 0) { HandleHistoryZoom(json, slotIdx); } else if (strcmp(type, "historyInfo") == 0) { HandleHistoryInfo(slotIdx); } @@ -1017,6 +1018,12 @@ void StreamHub::HandleRearm() { HandleArm(); } +void StreamHub::HandleForceTrigger() { + rearmPending_ = false; + (void) trigger_.Force(); + BroadcastTriggerState(); +} + void StreamHub::HandleTrigStop(const char *json) { /* {"type":"trigStop","stopped":bool} — absent "stopped" toggles. */ bool stopped = !trigger_.GetStopped(); diff --git a/Source/Applications/StreamHub/StreamHub.h b/Source/Applications/StreamHub/StreamHub.h index 340a603..5524854 100644 --- a/Source/Applications/StreamHub/StreamHub.h +++ b/Source/Applications/StreamHub/StreamHub.h @@ -140,6 +140,7 @@ private: void HandleRearm(); void HandleTrigStop(const char *json); void HandleSetTrigger(const char *json); + void HandleForceTrigger(); void HandleZoom(const char *json, uint32 slotIdx); void HandleHistoryZoom(const char *json, uint32 slotIdx); void HandleHistoryInfo(uint32 slotIdx); diff --git a/Source/Applications/StreamHub/TriggerEngine.cpp b/Source/Applications/StreamHub/TriggerEngine.cpp index 6583699..6ee0738 100644 --- a/Source/Applications/StreamHub/TriggerEngine.cpp +++ b/Source/Applications/StreamHub/TriggerEngine.cpp @@ -14,6 +14,8 @@ TriggerEngine::TriggerEngine() stopped_(false), prevValue_(0.0), prevValid_(false), + lastTime_(0.0), + lastTimeValid_(false), trigTime_(0.0), firedPreSec_(0.0), firedPostSec_(0.0), @@ -82,6 +84,11 @@ bool TriggerEngine::GetStopped() const { void TriggerEngine::CheckSample(float64 t, float64 v) { (void) mutex_.FastLock(); + /* Track the newest watched timestamp in every state so Force() has a + * reference time to latch the capture window around. */ + lastTime_ = t; + lastTimeValid_ = true; + if (state_ != kTrigArmed) { mutex_.FastUnLock(); return; @@ -122,6 +129,25 @@ void TriggerEngine::CheckSample(float64 t, float64 v) { mutex_.FastUnLock(); } +bool TriggerEngine::Force() { + (void) mutex_.FastLock(); + + bool ok = lastTimeValid_ && (state_ != kTrigCollecting); + if (ok) { + state_ = kTrigCollecting; + trigTime_ = lastTime_; + firedPreSec_ = config_.windowSec * config_.prePercent / 100.0; + firedPostSec_ = config_.windowSec - firedPreSec_; + firedValid_ = true; + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information, + "TriggerEngine: forced at t=%.6f (pre=%.4fs post=%.4fs)", + trigTime_, firedPreSec_, firedPostSec_); + } + + mutex_.FastUnLock(); + return ok; +} + TrigState TriggerEngine::GetState() const { (void) mutex_.FastLock(); TrigState ret = state_; diff --git a/Source/Applications/StreamHub/TriggerEngine.h b/Source/Applications/StreamHub/TriggerEngine.h index b46d8d0..6df1dbf 100644 --- a/Source/Applications/StreamHub/TriggerEngine.h +++ b/Source/Applications/StreamHub/TriggerEngine.h @@ -106,6 +106,15 @@ public: */ void CheckSample(float64 t, float64 v); + /** + * @brief Fire the trigger unconditionally at the most recent sample time of + * the watched signal, latching the pre/post window exactly as CheckSample + * does. Any state except COLLECTING → COLLECTING. + * @return false when no sample has been seen yet, or a capture is already + * being collected. + */ + bool Force(); + /** @return Current FSM state. */ TrigState GetState() const; @@ -127,6 +136,8 @@ private: bool stopped_; float64 prevValue_; ///< Last sample (edge detection) bool prevValid_; ///< First-sample guard in ARMED state + float64 lastTime_; ///< Timestamp of the newest watched sample + bool lastTimeValid_;///< true once a watched sample has been seen float64 trigTime_; ///< Latched trigger time (Unix s) float64 firedPreSec_; ///< Window pre-part latched at fire time float64 firedPostSec_; ///< Window post-part latched at fire time