diff --git a/Client/WebUI/hub.go b/Client/WebUI/hub.go index 67fa7d6..0fc40b8 100644 --- a/Client/WebUI/hub.go +++ b/Client/WebUI/hub.go @@ -82,9 +82,14 @@ func (c *wsClient) readPump() { case "addSource": label, _ := env["label"].(string) addr, _ := env["addr"].(string) + mcastGroup, _ := env["multicastGroup"].(string) + dataPortF, _ := env["dataPort"].(float64) if addr != "" { select { - case c.hub.commandCh <- hubCmd{op: "wsAddSource", label: label, addr: addr}: + case c.hub.commandCh <- hubCmd{ + op: "wsAddSource", label: label, addr: addr, + multicastGroup: mcastGroup, dataPort: int(dataPortF), + }: default: } } @@ -139,11 +144,13 @@ type taggedSample struct { type hubCmd struct { op string // "addSource","removeSource","setSourceState","updateConfig", // "wsAddSource","wsRemoveSource","wsSaveSources" - sourceID string - label string - addr string - state string - sigs []SignalInfo + sourceID string + label string + addr string + state string + sigs []SignalInfo + multicastGroup string + dataPort int } // Hub is the central broker between UDP clients and WebSocket clients. @@ -474,7 +481,9 @@ func (h *Hub) Run() { case "wsAddSource": if h.sm != nil { - go func(label, addr string) { h.sm.Add(label, addr, "", 0) }(cmd.label, cmd.addr) + go func(label, addr, mcastGroup string, dataPort int) { + h.sm.Add(label, addr, mcastGroup, dataPort) + }(cmd.label, cmd.addr, cmd.multicastGroup, cmd.dataPort) } case "wsRemoveSource": @@ -718,9 +727,10 @@ func (h *Hub) buildDataMessageForSource(src *sourceHubState, batch []DataSample) decimT, decimV := lttbDecimate(allT, allV, maxPushPoints) out[pfx+sig.Name] = sigData{T: decimT, V: decimV} - case n > 1 && sig.TimeMode == TimeModeFullArray: - // The time signal has the same N elements as the data signal. + case sig.TimeMode == TimeModeFullArray: // Each element pair (timeSig[k], dataSig[k]) is one (t, v) sample. + // This handles both standard N-element FullArray signals and + // Accumulate-mode scalars (n=1) auto-assigned FullArray time mode. hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs) var timeSigName string timerToSec := 1e-6 @@ -918,7 +928,9 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []DataS decimT, decimV := lttbDecimate(allT, allV, maxPushPoints) pairs[sig.Name] = pairBuf{t: decimT, v: decimV} - case n > 1 && sig.TimeMode == TimeModeFullArray: + case sig.TimeMode == TimeModeFullArray: + // Handles both standard N-element FullArray signals and + // Accumulate-mode scalars (n=1) with auto-assigned FullArray time mode. hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs) var timeSigName string timerToSec := 1e-6 @@ -957,9 +969,11 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []DataS allV = append(allV, vals[k]) } } - ringT, ringV := lttbDecimate(allT, allV, maxRingPoints) - if rb := h.getRing(pfx + sig.Name); rb != nil { - rb.write(ringT, ringV) + if writeRing { + ringT, ringV := lttbDecimate(allT, allV, maxRingPoints) + if rb := h.getRing(pfx + sig.Name); rb != nil { + rb.write(ringT, ringV) + } } decimT, decimV := lttbDecimate(allT, allV, maxPushPoints) pairs[sig.Name] = pairBuf{t: decimT, v: decimV} diff --git a/Client/WebUI/main.go b/Client/WebUI/main.go index f3aa5fa..6781b27 100644 --- a/Client/WebUI/main.go +++ b/Client/WebUI/main.go @@ -43,8 +43,8 @@ func main() { } } for _, arg := range sourceArgs { - label, addr := ParseSourceArg(arg) - sm.Add(label, addr, "", 0) + label, addr, mcastGroup, dataPort := ParseSourceArgFull(arg) + sm.Add(label, addr, mcastGroup, dataPort) } sub, err := fs.Sub(staticFiles, "static") diff --git a/Client/WebUI/protocol.go b/Client/WebUI/protocol.go index a364482..b46d5ca 100644 --- a/Client/WebUI/protocol.go +++ b/Client/WebUI/protocol.go @@ -34,6 +34,11 @@ const ( TimeModeFullArray uint8 = 1 // TimeSignal has same N elements; not expanded here TimeModeFirstSample uint8 = 2 // TimeSignal scalar = time of element [0] TimeModeLastSample uint8 = 3 // TimeSignal scalar = time of element [N-1] + + // PublishMode values – must match UDPStreamerPublishMode enum in UDPStreamer.h + PublishModeStrict uint8 = 0 // one packet per Synchronise() call + PublishModeAccumulate uint8 = 1 // variable batch; DATA has [8 HRT][4 numSamples][signals...] + PublishModeDecimate uint8 = 2 // one packet every Ratio calls ) // ─── Packet header (17 bytes, little-endian, packed) ───────────────────────── @@ -216,16 +221,17 @@ func nullTermString(b []byte) string { // ─── CONFIG payload parser ──────────────────────────────────────────────────── // ParseConfig decodes a fully-reassembled CONFIG payload. -func ParseConfig(payload []byte) ([]SignalInfo, error) { +// Returns the signal list, the publishing mode byte (PublishMode*), and any error. +func ParseConfig(payload []byte) ([]SignalInfo, uint8, error) { if len(payload) < 4 { - return nil, fmt.Errorf("config payload too short") + return nil, 0, fmt.Errorf("config payload too short") } numSigs := binary.LittleEndian.Uint32(payload[0:4]) offset := 4 sigs := make([]SignalInfo, 0, numSigs) for i := uint32(0); i < numSigs; i++ { if offset+SigDescSize > len(payload) { - return nil, fmt.Errorf("config payload truncated at signal %d", i) + return nil, 0, fmt.Errorf("config payload truncated at signal %d", i) } raw := payload[offset : offset+SigDescSize] si := SignalInfo{ @@ -245,7 +251,12 @@ func ParseConfig(payload []byte) ([]SignalInfo, error) { sigs = append(sigs, si) offset += SigDescSize } - return sigs, nil + // Trailing publish-mode byte (added after signal descriptors). + publishMode := PublishModeStrict + if offset < len(payload) { + publishMode = payload[offset] + } + return sigs, publishMode, nil } // ─── DATA payload parser ────────────────────────────────────────────────────── @@ -257,48 +268,116 @@ type DataSample struct { Values map[string][]float64 // key = signal name, value = []float64 with NumElements entries } -// ParseData decodes a fully-reassembled DATA payload using the provided signal config. -// arrivalTime is the wall-clock time at which the packet was received. -func ParseData(payload []byte, sigs []SignalInfo, arrivalTime time.Time) (DataSample, error) { +// parseElems reads n elements for sig from payload at offset, advancing offset. +// Returns the slice of float64 values and the new offset. +func parseElems(payload []byte, offset, n int, sig SignalInfo) ([]float64, int, error) { + elems := make([]float64, n) + if sig.QuantType == QuantNone { + sz := rawTypeSize(sig.TypeCode) + needed := n * sz + if offset+needed > len(payload) { + return nil, offset, fmt.Errorf("data payload truncated for signal %q", sig.Name) + } + for i := 0; i < n; i++ { + elems[i] = readRawElement(payload, offset+i*sz, sig.TypeCode) + } + offset += needed + } else { + sz := quantSize(sig.QuantType) + needed := n * sz + if offset+needed > len(payload) { + return nil, offset, fmt.Errorf("data payload truncated (quant) for signal %q", sig.Name) + } + for i := 0; i < n; i++ { + var raw uint16 + if sz == 1 { + raw = uint16(payload[offset+i]) + } else { + raw = binary.LittleEndian.Uint16(payload[offset+i*2:]) + } + elems[i] = dequantise(sig.QuantType, raw, sig.RangeMin, sig.RangeMax) + } + offset += needed + } + return elems, offset, nil +} + +// ParseData decodes a fully-reassembled DATA payload using the provided signal config +// and publishing mode. arrivalTime is the wall-clock time at which the packet arrived. +// +// For PublishModeAccumulate the payload format is: +// +// [8 HRT][4 numSamples][for each signal: accumulated scalars → numSamples elems; arrays → NumElements elems] +// +// The function returns one DataSample per accumulated snapshot so the hub can +// process each slot independently with its own timestamp. +func ParseData(payload []byte, sigs []SignalInfo, publishMode uint8, arrivalTime time.Time) ([]DataSample, error) { if len(payload) < 8 { - return DataSample{}, fmt.Errorf("data payload too short") + return nil, fmt.Errorf("data payload too short") } hrt := binary.LittleEndian.Uint64(payload[0:8]) offset := 8 + if publishMode == PublishModeAccumulate { + if len(payload) < 12 { + return nil, fmt.Errorf("accumulate data payload too short (missing numSamples)") + } + numSamples := int(binary.LittleEndian.Uint32(payload[8:12])) + offset = 12 + if numSamples == 0 { + return []DataSample{}, nil + } + + // Parse per-signal data blocks (all slots for a signal are contiguous). + accumVals := make(map[string][]float64, len(sigs)) // scalars: numSamples values + fixedVals := make(map[string][]float64, len(sigs)) // arrays: NumElements values + + for _, sig := range sigs { + n := sig.NumElements() + if n == 1 { + // Accumulated scalar: read numSamples back-to-back elements. + elems, newOff, err := parseElems(payload, offset, numSamples, sig) + if err != nil { + return nil, err + } + offset = newOff + accumVals[sig.Name] = elems + } else { + // Fixed array (non-accumulated): one set of NumElements values. + elems, newOff, err := parseElems(payload, offset, n, sig) + if err != nil { + return nil, err + } + offset = newOff + fixedVals[sig.Name] = elems + } + } + + // Build one DataSample per slot. + samples := make([]DataSample, numSamples) + for k := 0; k < numSamples; k++ { + vals := make(map[string][]float64, len(sigs)) + for sigName, av := range accumVals { + vals[sigName] = []float64{av[k]} + } + for sigName, fv := range fixedVals { + vals[sigName] = fv // shared read-only reference; hub does not modify + } + samples[k] = DataSample{HRTTimestamp: hrt, WallTime: arrivalTime, Values: vals} + } + return samples, nil + } + + // Strict / Decimate: single snapshot, one element set per signal. vals := make(map[string][]float64, len(sigs)) for _, sig := range sigs { n := sig.NumElements() - elems := make([]float64, n) - - if sig.QuantType == QuantNone { - sz := rawTypeSize(sig.TypeCode) - needed := n * sz - if offset+needed > len(payload) { - return DataSample{}, fmt.Errorf("data payload truncated for signal %q", sig.Name) - } - for i := 0; i < n; i++ { - elems[i] = readRawElement(payload, offset+i*sz, sig.TypeCode) - } - offset += needed - } else { - sz := quantSize(sig.QuantType) - needed := n * sz - if offset+needed > len(payload) { - return DataSample{}, fmt.Errorf("data payload truncated (quant) for signal %q", sig.Name) - } - for i := 0; i < n; i++ { - var raw uint16 - if sz == 1 { - raw = uint16(payload[offset+i]) - } else { - raw = binary.LittleEndian.Uint16(payload[offset+i*2:]) - } - elems[i] = dequantise(sig.QuantType, raw, sig.RangeMin, sig.RangeMax) - } - offset += needed + elems, newOff, err := parseElems(payload, offset, n, sig) + if err != nil { + return nil, err } + offset = newOff vals[sig.Name] = elems } - return DataSample{HRTTimestamp: hrt, WallTime: arrivalTime, Values: vals}, nil + return []DataSample{{HRTTimestamp: hrt, WallTime: arrivalTime, Values: vals}}, nil } diff --git a/Client/WebUI/sources.go b/Client/WebUI/sources.go index 121b1e3..3209fb2 100644 --- a/Client/WebUI/sources.go +++ b/Client/WebUI/sources.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "os" + "strconv" "strings" "sync" "sync/atomic" @@ -135,9 +136,33 @@ func (sm *SourceManager) Load(path string) error { // ParseSourceArg parses "label@host:port" or "host:port". func ParseSourceArg(s string) (label, addr string) { - s = strings.TrimSpace(s) - if idx := strings.Index(s, "@"); idx >= 0 { - return strings.TrimSpace(s[:idx]), strings.TrimSpace(s[idx+1:]) - } - return "", s + label, addr, _, _ = ParseSourceArgFull(s) + return +} + +// ParseSourceArgFull parses "[label@]host:port[/multicastGroup:dataPort]". +// Examples: +// +// "Streamer@127.0.0.1:44500/239.0.0.1:44503" → label="Streamer", addr="127.0.0.1:44500", group="239.0.0.1", port=44503 +// "127.0.0.1:44501" → label="", addr="127.0.0.1:44501", group="", port=0 +func ParseSourceArgFull(s string) (label, addr, multicastGroup string, dataPort int) { + s = strings.TrimSpace(s) + rest := s + if idx := strings.Index(s, "@"); idx >= 0 { + label = strings.TrimSpace(s[:idx]) + rest = strings.TrimSpace(s[idx+1:]) + } + if idx := strings.Index(rest, "/"); idx >= 0 { + addr = strings.TrimSpace(rest[:idx]) + mcastPart := strings.TrimSpace(rest[idx+1:]) + if lastColon := strings.LastIndex(mcastPart, ":"); lastColon >= 0 { + multicastGroup = strings.TrimSpace(mcastPart[:lastColon]) + dataPort, _ = strconv.Atoi(strings.TrimSpace(mcastPart[lastColon+1:])) + } else { + multicastGroup = mcastPart + } + } else { + addr = rest + } + return } diff --git a/Client/WebUI/static/app.js b/Client/WebUI/static/app.js index 3cde343..362abae 100644 --- a/Client/WebUI/static/app.js +++ b/Client/WebUI/static/app.js @@ -39,6 +39,12 @@ function getSigStyle(key) { if (!sigStyle[key]) sigStyle[key] = { color: getTraceColor(key), width: 1.5, dash: 'solid', marker: 'none', markerSize: 4 }; return sigStyle[key]; } + +// Per-signal vertical scale state: key → {mode, divValue, offset, _resolvedDiv, _resolvedOffset} +const sigVScale = {}; +// Active signal per plot: plotId → key +const plotActiveSignal = {}; + function setSigStyle(key, updates) { const s = getSigStyle(key); Object.assign(s, updates); @@ -53,6 +59,121 @@ function setSigStyle(key, updates) { plots.forEach(p => { if (p.traces.includes(key)) { createUPlot(p); p.needsRedraw = true; } }); } +/* ─── VScale helpers ─────────────────────────────────────────────────────── */ +// vsKey: compound key "plotId:signalKey" so same signal in different plots is independent. +function getVScale(plotId, key) { + const vsKey = plotId + ':' + key; + if (!sigVScale[vsKey]) sigVScale[vsKey] = { mode: 'auto', divValue: 1, offset: 0, screenPos: 0, _resolvedDiv: null, _resolvedOffset: null }; + return sigVScale[vsKey]; +} + +function findSignalMeta(key) { + const colon = key.indexOf(':'); + if (colon < 0) return null; + const src = sourcesMap[key.slice(0, colon)]; + if (!src) return null; + return src.signals.find(s => s.name === key.slice(colon + 1)) || null; +} + +// Resolve the effective {divValue, offset, screenPos} for a signal given its raw data array. +// y_norm = (y_raw - offset) / divValue + screenPos +// divValue: units per division offset: raw value at screen center screenPos: divisions from center +// Also caches the resolved values in vs._resolvedDiv/_resolvedOffset for Y-axis label use. +function resolveVScale(plotId, key, rawY) { + const vs = getVScale(plotId, key); + const screenPos = vs.screenPos || 0; + 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; + vs._resolvedDiv = divValue; vs._resolvedOffset = offset; + return { divValue, offset, screenPos }; + } + // Fall through to auto if no range + } + if (vs.mode === 'manual') { + const divValue = Math.max(vs.divValue || 1, 1e-30); + const offset = vs.offset != null ? vs.offset : 0; + vs._resolvedDiv = divValue; vs._resolvedOffset = offset; + return { divValue, offset, screenPos }; + } + // Auto: fit data in central 6 of 8 divisions, centered at screenPos + let min = Infinity, max = -Infinity; + for (let i = 0; i < rawY.length; i++) { + const v = rawY[i]; + if (v != null && isFinite(v)) { if (v < min) min = v; if (v > max) max = v; } + } + 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; + vs._resolvedDiv = divValue; vs._resolvedOffset = offset; + return { divValue, offset, screenPos }; +} + +// Apply vscale normalization to a list of raw Y arrays (one per trace in p.traces). +// Returns normalized arrays where y_norm = (y_raw - offset) / divValue + screenPos. +function applyVScaleNorm(p, yArrays) { + return yArrays.map((rawY, ki) => { + const key = p.traces[ki]; + const { divValue, offset, screenPos } = resolveVScale(p.id, key, rawY); + const out = new Float64Array(rawY.length); + for (let i = 0; i < rawY.length; i++) { + const v = rawY[i]; + out[i] = (v == null || !isFinite(v)) ? NaN : (v - offset) / divValue + screenPos; + } + return out; + }); +} + +// Set the active (Y-axis-labelled) signal for a plot and update badge highlights. +function setActiveSig(plotId, key) { + if (key === null || key === undefined) { + delete plotActiveSignal[plotId]; + } else { + plotActiveSignal[plotId] = key; + } + const c = document.getElementById('badges-' + plotId); + if (c) c.querySelectorAll('.sig-badge').forEach(b => + b.classList.toggle('sig-badge-active', key != null && b.dataset.key === key)); + const p = plots.find(q => q.id === plotId); + if (p && p.uplot) p.uplot.redraw(false); +} + +// Mark plots containing key dirty and refresh badge vscale text. +function refreshPlotForKey(key) { + plots.forEach(p => { + if (p.traces.includes(key)) { + p.needsRedraw = true; + _updateBadgeVScaleInfo(p.id, key); + } + }); +} + +// Format a numeric value concisely for badge/axis display. +function _fmtVal(v) { + if (v == null || !isFinite(v)) return '?'; + const abs = Math.abs(v); + if (abs === 0) return '0'; + if (abs >= 1e4 || abs < 1e-3) return v.toExponential(1); + return parseFloat(v.toPrecision(3)).toString(); +} + +// Refresh the vscale info text inside a badge. +function _updateBadgeVScaleInfo(plotId, key) { + const c = document.getElementById('badges-' + plotId); if (!c) return; + const b = c.querySelector('[data-key="' + CSS.escape(key) + '"]'); if (!b) return; + const infoEl = b.querySelector('.vscale-info'); if (!infoEl) return; + const vs = sigVScale[plotId + ':' + key]; + if (!vs) { infoEl.textContent = ''; return; } + const divValue = vs._resolvedDiv || vs.divValue || 1; + const sp = vs.screenPos || 0; + let txt = _fmtVal(divValue) + '/div'; + if (sp !== 0) txt += ' ' + (sp >= 0 ? '+' : '') + sp.toFixed(1) + 'div'; + infoEl.textContent = txt; +} + // Sync: shared uPlot cursor crosshair across all live plots const LIVE_SYNC = uPlot.sync('live'); const TRIG_SYNC = uPlot.sync('trig'); @@ -692,32 +813,83 @@ function buildDataFromFetched(p, fetchedSignals, targetPts) { if (!sd || !sd.t.length) { yArrays.push(new Float64Array(sharedT.length)); continue; } yArrays.push(resampleLinear(sd.t, sd.v, sharedT)); } - return [sharedT, ...yArrays]; + return [sharedT, ...applyVScaleNorm(p, yArrays)]; } /* ════════════════════════════════════════════════════════════════ uPlot helpers ════════════════════════════════════════════════════════════════ */ -// Format Unix seconds → HH:MM:SS.mmm (used for live x-axis ticks) -function fmtLiveTick(u, vals) { - return vals.map(v => { - if (v == null) return ''; - const d = new Date(v * 1000); - const hh = String(d.getHours()).padStart(2, '0'); - const mm = String(d.getMinutes()).padStart(2, '0'); - const ss = String(d.getSeconds()).padStart(2, '0'); +// ─── Time formatting helpers ────────────────────────────────────────────────── + +// Returns the span (in seconds) of the currently visible x-axis across all plots. +// Falls back to windowSec when no uPlot instances exist yet. +function currentXSpan() { + for (const p of plots) { + if (p.uplot) { + const s = p.uplot.scales.x; + if (s && s.min != null && s.max != null) return Math.abs(s.max - s.min); + } + } + return windowSec; +} + +// Format a signed duration (seconds) auto-selecting s / ms / µs / ns based on +// refSpan (e.g. the visible x-range or the value itself). +// sign = '+' prefix only when showSign is true (default false for ΔT display). +function fmtDuration(sec, refSpan, showSign) { + const abs = Math.abs(sec); + const sign = showSign ? (sec < 0 ? '−' : '+') : (sec < 0 ? '−' : ''); + if (refSpan < 1e-6) { // nanosecond range + return sign + (abs * 1e9).toFixed(1) + ' ns'; + } else if (refSpan < 1e-3) { // microsecond range + return sign + (abs * 1e6).toFixed(3) + ' µs'; + } else if (refSpan < 1) { // millisecond range + return sign + (abs * 1e3).toFixed(3) + ' ms'; + } else { // second range + return sign + abs.toFixed(6) + ' s'; + } +} + +// Format a Unix-seconds timestamp → HH:MM:SS.fraction +// The number of sub-second digits adapts to the visible x-range span. +function fmtLiveTime(v, span) { + const d = new Date(v * 1000); + const hh = String(d.getHours()).padStart(2, '0'); + const mm = String(d.getMinutes()).padStart(2, '0'); + const ss = String(d.getSeconds()).padStart(2, '0'); + const frac = v - Math.floor(v); // sub-second part, full float64 precision + if (span < 1e-6) { + // show 9 decimal places (ns precision) + const ns = Math.round(frac * 1e9); + return hh + ':' + mm + ':' + ss + '.' + String(ns).padStart(9, '0'); + } else if (span < 1e-3) { + // show 6 decimal places (µs precision) + const us = Math.round(frac * 1e6); + return hh + ':' + mm + ':' + ss + '.' + String(us).padStart(6, '0'); + } else if (span < 1) { + // show 3 decimal places (ms precision) — tick labels only show ms const ms = String(d.getMilliseconds()).padStart(3, '0'); return hh + ':' + mm + ':' + ss + '.' + ms; - }); + } else { + const ms = String(d.getMilliseconds()).padStart(3, '0'); + return hh + ':' + mm + ':' + ss + '.' + ms; + } } -// Format relative seconds → ±Xms or ±Xs (used for trigger x-axis ticks) + +// Format Unix seconds → HH:MM:SS.fraction (used for live x-axis ticks) +// Precision adapts to the visible x-range (via u.scales.x.{min,max}). +function fmtLiveTick(u, vals) { + const span = (u.scales.x && u.scales.x.max != null) + ? Math.abs(u.scales.x.max - u.scales.x.min) : windowSec; + return vals.map(v => v == null ? '' : fmtLiveTime(v, span)); +} + +// Format relative seconds → auto-scaled unit (used for trigger x-axis ticks) +// Unit (s/ms/µs/ns) is determined by the visible x-range span. function fmtTrigTick(u, vals) { - return vals.map(v => { - if (v == null) return ''; - const abs = Math.abs(v), sign = v < 0 ? '-' : '+'; - if (abs < 1) return sign + (abs * 1000).toFixed(1) + 'ms'; - return sign + abs.toFixed(3) + 's'; - }); + const span = (u.scales.x && u.scales.x.max != null) + ? Math.abs(u.scales.x.max - u.scales.x.min) : 1; + return vals.map(v => v == null ? '' : fmtDuration(v, span, true)); } // Draw the trigger marker: dashed vertical line at t=0, plus a horizontal threshold @@ -745,7 +917,16 @@ function drawTriggerMarker(u, p) { ctx.fillText('T', px + 3, bbox.top + 2); // Horizontal threshold line — only on plots that contain the trigger signal if (p && trig.signal && p.traces.includes(trig.signal)) { - const y = u.valToPos(trig.threshold, 'y', true); + // Normalize the raw threshold to this plot's vscale for the trigger signal. + const tvs = p ? sigVScale[p.id + ':' + trig.signal] : null; + let threshNorm = trig.threshold; + if (tvs) { + const dv = tvs._resolvedDiv || tvs.divValue || 1; + const ofs = tvs._resolvedOffset != null ? tvs._resolvedOffset : (tvs.offset || 0); + const sp = tvs.screenPos || 0; + threshNorm = (trig.threshold - ofs) / dv + sp; + } + const y = u.valToPos(threshNorm, 'y', true); if (y >= bbox.top && y <= bbox.top + bbox.height) { const py = Math.round(y); ctx.strokeStyle = 'rgba(203,166,247,0.45)'; @@ -778,6 +959,72 @@ function interpAtTime(u, si, t) { return v0 + (t - t0) / (t1 - t0) * (v1 - v0); } +// Redraw the active signal's line on top of all series with a wider stroke, so it +// visually appears in the foreground regardless of series draw order. +function drawActiveSeries(u, p) { + if (!u.bbox) return; + const activeKey = plotActiveSignal[p.id]; + if (!activeKey) return; + const idx = p.traces.indexOf(activeKey); + if (idx < 0) return; + const xs = u.data[0]; + const ys = u.data[idx + 1]; // +1 because index 0 is time + if (!xs || !ys) return; + const style = getSigStyle(activeKey); + const dpr = window.devicePixelRatio || 1; + const { ctx } = u; + ctx.save(); + ctx.strokeStyle = style.color; + ctx.lineWidth = style.width * 2 * dpr; + ctx.lineJoin = 'round'; + ctx.lineCap = 'round'; + ctx.beginPath(); + let started = false; + for (let i = 0; i < xs.length; i++) { + const yv = ys[i]; + if (yv == null || !isFinite(yv)) { started = false; continue; } + const xPx = u.valToPos(xs[i], 'x', true); + const yPx = u.valToPos(yv, 'y', true); + if (!started) { ctx.moveTo(xPx, yPx); started = true; } + else ctx.lineTo(xPx, yPx); + } + ctx.stroke(); + ctx.restore(); +} + +// Draw offset position markers (right-pointing triangles) on the left edge of the plot +// for each signal. Active signal marker is larger and outlined in white. +function drawOffsetMarkers(u, p) { + if (!u.bbox) return; + const { ctx, bbox } = u; + const dpr = window.devicePixelRatio || 1; + + p.traces.forEach(key => { + const vs = sigVScale[p.id + ':' + key]; + const screenPos = vs ? (vs.screenPos || 0) : 0; + const yCtr = u.valToPos(screenPos, 'y', true); + const mH = (plotActiveSignal[p.id] === key ? 7 : 5) * dpr; + const mW = (plotActiveSignal[p.id] === key ? 10 : 7) * dpr; + if (yCtr < bbox.top - mH * 2 || yCtr > bbox.top + bbox.height + mH * 2) return; + const isActive = plotActiveSignal[p.id] === key; + ctx.save(); + ctx.fillStyle = getSigStyle(key).color; + // Right-pointing triangle: tip at left edge of plot area, body extends left into Y-axis area + ctx.beginPath(); + ctx.moveTo(bbox.left + dpr, yCtr); + ctx.lineTo(bbox.left - mW + dpr, yCtr - mH); + ctx.lineTo(bbox.left - mW + dpr, yCtr + mH); + ctx.closePath(); + ctx.fill(); + if (isActive) { + ctx.strokeStyle = 'rgba(255,255,255,0.75)'; + ctx.lineWidth = dpr; + ctx.stroke(); + } + ctx.restore(); + }); +} + // Draw custom marker shapes (square, cross, diamond) for all series in a plot. // Circle markers are handled natively by uPlot's points option. function drawSeriesMarkers(u, p) { @@ -855,10 +1102,20 @@ function drawCursorLines(u, p) { if (p) { const DSZ = 5; // diamond half-size in px p.traces.forEach((key, idx) => { - const v = interpAtTime(u, idx + 1, val); - if (v === null) return; - const cy = u.valToPos(v, 'y', true); + const vNorm = interpAtTime(u, idx + 1, val); + 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 + screenPos → y_raw = (y_norm - screenPos) * divValue + offset + const vs = sigVScale[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); + const sp = vs.screenPos || 0; + vReal = (vNorm - sp) * dv + ofs; + } const tc = getSigStyle(key).color; // Diamond marker at intersection ctx.fillStyle = tc; @@ -871,13 +1128,13 @@ function drawCursorLines(u, p) { ctx.lineTo(px - DSZ, cy); ctx.closePath(); ctx.fill(); - // Value text next to diamond - const str = Math.abs(v) >= 10000 ? v.toExponential(2) : parseFloat(v.toPrecision(4)).toString(); + // Value text next to diamond (real units) + const str = Math.abs(vReal) >= 10000 ? vReal.toExponential(2) : parseFloat(vReal.toPrecision(4)).toString(); ctx.fillStyle = tc; ctx.font = '11px monospace'; const currentAlign = ctx.textAlign; - ctx.textAlign = "left"; // horizontal alignment - ctx.textBaseline = "middle"; + ctx.textAlign = 'left'; + ctx.textBaseline = 'middle'; ctx.fillText(str, px + DSZ + 4, cy); ctx.textAlign = currentAlign; }); @@ -955,20 +1212,41 @@ function makeUPlotOpts(p, inTrigMode) { } return { time: false, auto: false, min: xMin, max: xMax }; })(), - y: { auto: true }, + y: { auto: false, min: -4.5, max: 4.5 }, }, series: seriesArr, axes: [ { stroke: '#7f849c', grid: { stroke: '#313244', width: 1 }, ticks: { stroke: '#313244', width: 1 }, - values: xVals, size: 36, space: 90 + values: xVals, size: 36, + // Always produce exactly 10 horizontal divisions (11 evenly-spaced tick lines). + splits: (u, _ai, sMin, sMax) => { + const n = 10, span = sMax - sMin; + if (span === 0) return [sMin]; + return Array.from({ length: n + 1 }, (_, i) => sMin + span * i / n); + }, + }, + { + stroke: '#7f849c', grid: { stroke: '#313244', width: 1 }, ticks: { stroke: '#313244', width: 1 }, size: 60, + // Fixed 9 splits at integer divisions [-4..4] matching the normalized Y scale. + splits: () => [-4, -3, -2, -1, 0, 1, 2, 3, 4], + // Labels show real-unit values of the active signal for the plot. + // y_norm = (y_raw - offset) / divValue + screenPos → y_raw = (y_norm - screenPos) * divValue + offset + values: (u, vals) => { + const activeKey = plotActiveSignal[p.id]; + const vs = activeKey ? sigVScale[p.id + ':' + activeKey] : null; + if (!vs) return vals.map(v => v == null ? '' : v.toFixed(1)); + const divValue = vs._resolvedDiv || vs.divValue || 1; + const offset = vs._resolvedOffset != null ? vs._resolvedOffset : (vs.offset || 0); + const screenPos = vs.screenPos || 0; + return vals.map(v => v == null ? '' : _fmtVal((v - screenPos) * divValue + offset)); + }, }, - { stroke: '#7f849c', grid: { stroke: '#313244', width: 1 }, ticks: { stroke: '#313244', width: 1 }, size: 50 }, ], legend: { show: false }, padding: [4, 4, 0, 0], hooks: { - draw: [u => { drawCursorLines(u, p); drawSeriesMarkers(u, p); drawTriggerMarker(u, p); }], + draw: [u => { drawActiveSeries(u, p); drawOffsetMarkers(u, p); drawCursorLines(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 @@ -1062,6 +1340,59 @@ function createUPlot(p) { document.addEventListener('mouseup', onUp); }, true); // capture:true so we fire before uPlot's own handlers + // ── Offset marker drag ───────────────────────────────────────────────────── + // Detect mousedown near the left edge of the plot area (marker triangle zone). + // Dragging moves the marker AND the signal together by changing screenPos. + p.div.addEventListener('mousedown', e => { + if (e.button !== 0 || !p.uplot || !p.uplot.bbox) return; + const canvas = p.uplot.ctx.canvas; + const rect = canvas.getBoundingClientRect(); + const dpr = window.devicePixelRatio || 1; + const plotLeftCss = rect.left + p.uplot.bbox.left / dpr; + const markerZone = 12; // CSS px hit area to left/right of plot edge + + if (e.clientX > plotLeftCss + 3 || e.clientX < plotLeftCss - markerZone) return; + + // Find which marker was hit (closest to screenPos canvas position per signal). + let hitKey = null, hitDist = Infinity; + p.traces.forEach(key => { + const vs = sigVScale[p.id + ':' + key]; + const screenPos = vs ? (vs.screenPos || 0) : 0; + const yDev = p.uplot.valToPos(screenPos, 'y', true); + const yCss = rect.top + yDev / dpr; + const dist = Math.abs(e.clientY - yCss); + if (dist < 14 && dist < hitDist) { hitDist = dist; hitKey = key; } + }); + if (!hitKey) return; + + e.preventDefault(); + e.stopPropagation(); + setActiveSig(p.id, hitKey); + + const vs = getVScale(p.id, hitKey); + const startY = e.clientY; + const startScreenPos = vs.screenPos || 0; + const overRect = p.uplot.over.getBoundingClientRect(); + + const onMove = ev => { + const dy = ev.clientY - startY; // positive = down in canvas = lower y_norm + // Y scale spans 9 divisions over plot height; drag up → higher screenPos. + const dNorm = -dy / overRect.height * 9; + vs.screenPos = Math.max(-4, Math.min(4, startScreenPos + dNorm)); + // Keep "Position" input in sync if the vscale menu is open for this signal. + if (_vsMenuKey === hitKey) { + document.getElementById('vscale-pos').value = parseFloat(vs.screenPos.toPrecision(4)); + } + refreshPlotForKey(hitKey); + }; + const onUp = () => { + document.removeEventListener('mousemove', onMove); + document.removeEventListener('mouseup', onUp); + }; + document.addEventListener('mousemove', onMove); + document.addEventListener('mouseup', onUp); + }, true); + // Pan support: Shift+left-drag pans the current view (synced across all plots). // Works in both zoomed mode (xRange set) and rolling mode (freezes the window first). let _panActive = false, _panAnchorX = 0, _panAnchorMin = 0, _panAnchorMax = 0; @@ -1191,7 +1522,9 @@ function buildLiveData(p) { if (p.xRange) { const zd = zoomData[p.id]; if (zd && Math.abs(zd.t0 - t0) < 1e-9 && Math.abs(zd.t1 - t1) < 1e-9) { - return buildDataFromFetched(p, zd.signals, Math.max(LTTB_MIN, ((p.uplot ? p.uplot.width : p.div.clientWidth) || 600) * 2)); + const fetched = buildDataFromFetched(p, zd.signals, Math.max(LTTB_MIN, ((p.uplot ? p.uplot.width : p.div.clientWidth) || 600) * 2)); + // Only use server data if it actually has samples; otherwise fall through to local buffer. + if (fetched[0] && fetched[0].length > 0) return fetched; } } @@ -1249,7 +1582,7 @@ function buildLiveData(p) { yArrays.push(resampleLinear(sl.t, sl.v, sharedT)); } - return [sharedT, ...yArrays]; + return [sharedT, ...applyVScaleNorm(p, yArrays)]; } function buildTrigData(p) { @@ -1301,7 +1634,7 @@ function buildTrigData(p) { yArrays.push(resampleLinear(relT, sl.v, sharedT)); } - return [sharedT, ...yArrays]; + return [sharedT, ...applyVScaleNorm(p, yArrays)]; } /* ════════════════════════════════════════════════════════════════ @@ -1486,26 +1819,24 @@ function updateCursorReadout() { ro.classList.toggle('visible', active); if (!active) return; - // Format depends on mode: live = HH:MM:SS.mmm, trigger = ±Xms + // Use the current visible x-range to pick the display unit. + const span = currentXSpan(); + + // Format a cursor position: trigger mode → signed relative duration; + // live mode → absolute wall time with span-appropriate precision. const fmt = v => { if (v === null) return '—'; - if (trig.enabled && trig.snapshot) { - const abs = Math.abs(v), sign = v < 0 ? '-' : '+'; - return abs < 1 ? sign + (abs * 1000).toFixed(3) + 'ms' : sign + abs.toFixed(6) + 's'; - } - const d = new Date(v * 1000); - return String(d.getHours()).padStart(2, '0') + ':' - + String(d.getMinutes()).padStart(2, '0') + ':' - + String(d.getSeconds()).padStart(2, '0') + '.' - + String(d.getMilliseconds()).padStart(3, '0'); + if (trig.enabled && trig.snapshot) return fmtDuration(v, span, true); + return fmtLiveTime(v, span); }; + document.getElementById('cur-ta').textContent = 'A: ' + fmt(cursors.tA); document.getElementById('cur-tb').textContent = 'B: ' + fmt(cursors.tB); + if (cursors.tA !== null && cursors.tB !== null) { - const dt = cursors.tB - cursors.tA, abs = Math.abs(dt); - const s = dt >= 0 ? '+' : '-'; - const str = abs < 1 ? s + (abs * 1000).toFixed(3) + 'ms' : s + abs.toFixed(6) + 's'; - document.getElementById('cur-dt').textContent = 'ΔT: ' + str; + const dt = cursors.tB - cursors.tA; + // ΔT auto-scales by its own magnitude for precision regardless of x-range. + document.getElementById('cur-dt').textContent = 'ΔT: ' + fmtDuration(dt, Math.abs(dt), true); } else { document.getElementById('cur-dt').textContent = 'ΔT: —'; } @@ -1549,7 +1880,24 @@ function openTrigBar(open) { } document.getElementById('btn-trigger').addEventListener('click', () => openTrigBar(!trig.enabled)); document.getElementById('trig-signal').addEventListener('change', e => { - trig.signal = e.target.value; trig.prevVal = null; + const val = e.target.value; + if (!val) { trig.signal = ''; trigDisarm(); return; } + // Array signal: ask for element index via picker dialog. + const meta = findSignalMeta(val); + const n = meta ? numElements(meta) : 1; + if (meta && !isTemporal(meta) && n > 1) { + showArrayIdxPicker(val, n, idx => { + trig.signal = val + '[' + idx + ']'; trig.prevVal = null; + if (trig.enabled) trigArm(); + }, () => { + // Cancelled: revert selection to current trig.signal base or empty. + const sel = document.getElementById('trig-signal'); + const base = trig.signal ? trig.signal.replace(/\[\d+\]$/, '') : ''; + sel.value = base || ''; + }); + return; + } + trig.signal = val; trig.prevVal = null; if (trig.enabled && trig.signal) trigArm(); else if (!trig.signal) trigDisarm(); }); document.getElementById('trig-edge').addEventListener('change', e => { trig.edge = e.target.value; if (trig.armed) trig.prevVal = null; }); @@ -1581,28 +1929,30 @@ document.getElementById('btn-trig-stop').addEventListener('click', () => { Trigger signal selector ════════════════════════════════════════════════════════════════ */ function buildTrigSignalSelect() { - const sel = document.getElementById('trig-signal'), cur = sel.value; + const sel = document.getElementById('trig-signal'); + // Preserve the currently active trig.signal (may include an array index like "[3]"). + const curBase = trig.signal ? trig.signal.replace(/\[\d+\]$/, '') : ''; sel.innerHTML = ''; Object.values(sourcesMap).forEach(src => { const prefix = src.id + ':'; const srcLabel = src.label || src.addr || src.id; (src.signals || []).forEach(sig => { const n = numElements(sig); + const key = prefix + sig.name; + const o = document.createElement('option'); + o.value = key; if (isTemporal(sig) || n === 1) { - const key = prefix + sig.name; - const o = document.createElement('option'); - o.value = key; o.textContent = srcLabel + ': ' + sig.name; sel.appendChild(o); + o.textContent = srcLabel + ': ' + sig.name; } else { - for (let i = 0; i < n; i++) { - const key = prefix + sig.name + '[' + i + ']'; - const o = document.createElement('option'); - o.value = key; o.textContent = srcLabel + ': ' + sig.name + '[' + i + ']'; sel.appendChild(o); - } + // Array: single entry; user chooses index via dialog on selection. + o.textContent = srcLabel + ': ' + sig.name + ' [0…' + (n - 1) + ']'; } + sel.appendChild(o); }); }); - if (cur && [...sel.options].some(o => o.value === cur)) sel.value = cur; - trig.signal = sel.value; + // Restore selection: match base key so array element "sig[3]" selects "sig" option. + if (curBase && [...sel.options].some(o => o.value === curBase)) sel.value = curBase; + // Do NOT overwrite trig.signal here — an array element selection must be preserved. } /* ════════════════════════════════════════════════════════════════ @@ -1919,6 +2269,7 @@ function addPlot() {
Plot ${id}
+
Drop signals here
⚡ Collecting…
@@ -1955,6 +2306,12 @@ function removeTraceFrom(plotId, signalKey) { const p = plots.find(p => p.id === plotId); if (!p) return; p.traces = p.traces.filter(t => t !== signalKey); removeBadge(plotId, signalKey); + // If the removed trace was active, close toolbar and pick a new active signal. + if (plotActiveSignal[plotId] === signalKey) { + if (_vsMenuPlotId === plotId) hideVScaleMenu(); + const newActive = p.traces[0] || null; + if (newActive) setActiveSig(plotId, newActive); else delete plotActiveSignal[plotId]; + } createUPlot(p); p.needsRedraw = true; if (!p.traces.length) document.querySelector('#hint-' + plotId).style.display = ''; @@ -1966,17 +2323,46 @@ function addBadge(plotId, key) { const color = getSigStyle(key).color; const badge = document.createElement('span'); badge.className = 'sig-badge'; badge.dataset.key = key; + const dot = document.createElement('span'); dot.className = 'trace-dot'; dot.style.background = color; + + // Show signal name without the "sourceId:" prefix. + const displayName = key.includes(':') ? key.split(':').slice(1).join(':') : key; + const nameSpan = document.createElement('span'); nameSpan.textContent = displayName; + + // Small vscale info text (V/div + offset when not in auto mode). + const infoSpan = document.createElement('span'); infoSpan.className = 'vscale-info'; + const x = document.createElement('span'); x.className = 'sig-badge-x'; x.title = 'Remove'; x.textContent = '×'; - x.addEventListener('click', () => removeTraceFrom(plotId, key)); + x.addEventListener('click', e => { e.stopPropagation(); removeTraceFrom(plotId, key); }); + + // Left-click: select + show vscale toolbar; click same signal again to deselect. + badge.addEventListener('click', e => { + if (e.target === x) return; + const isActive = plotActiveSignal[plotId] === key; + const toolbarVisible = _vsMenuKey === key && _vsMenuPlotId === plotId; + if (isActive && toolbarVisible) { + setActiveSig(plotId, null); + hideVScaleMenu(); + } else { + setActiveSig(plotId, key); + showVScaleMenu(key, plotId); + } + }); + // Right-click: open signal style (color/width/…) menu. badge.addEventListener('contextmenu', e => { e.preventDefault(); showSignalMenu(key, plotId, e.clientX, e.clientY); }); - // Show signal name without the "sourceId:" prefix in the badge label. - const displayName = key.includes(':') ? key.split(':').slice(1).join(':') : key; - badge.appendChild(dot); badge.appendChild(document.createTextNode(displayName)); badge.appendChild(x); + + badge.appendChild(dot); + badge.appendChild(nameSpan); + badge.appendChild(infoSpan); + badge.appendChild(x); c.appendChild(badge); + + // Auto-activate the first signal added to this plot. + if (!plotActiveSignal[plotId]) setActiveSig(plotId, key); } function removeBadge(plotId, key) { const c = document.getElementById('badges-' + plotId); if (!c) return; @@ -2167,9 +2553,12 @@ function onSources(msg) { if (statsOpen) _refreshStatsSelector(); } -function addSourceWS(label, addr) { +function addSourceWS(label, addr, multicastGroup, dataPort) { if (ws && ws.readyState === WebSocket.OPEN) { - ws.send(JSON.stringify({ type: 'addSource', label, addr })); + const msg = { type: 'addSource', label, addr }; + if (multicastGroup) { msg.multicastGroup = multicastGroup; } + if (dataPort) { msg.dataPort = dataPort; } + ws.send(JSON.stringify(msg)); } } @@ -2204,12 +2593,23 @@ function makeAddSourceSection() { labelInput.className = 'add-src-input'; labelInput.type = 'text'; labelInput.placeholder = 'label (optional)'; + const mcastInput = document.createElement('input'); + mcastInput.className = 'add-src-input'; mcastInput.type = 'text'; + mcastInput.placeholder = 'multicast group (e.g. 239.0.0.1, optional)'; + + const dataPortInput = document.createElement('input'); + dataPortInput.className = 'add-src-input'; dataPortInput.type = 'number'; + dataPortInput.placeholder = 'data port (multicast only)'; + dataPortInput.min = '1'; dataPortInput.max = '65535'; + const addBtn = document.createElement('button'); addBtn.className = 'add-src-btn'; addBtn.textContent = 'Connect'; addBtn.addEventListener('click', () => { const addr = addrInput.value.trim(); if (!addr) return; - addSourceWS(labelInput.value.trim(), addr); - addrInput.value = ''; labelInput.value = ''; + const mcastGroup = mcastInput.value.trim(); + const dataPort = dataPortInput.value ? parseInt(dataPortInput.value, 10) : 0; + addSourceWS(labelInput.value.trim(), addr, mcastGroup, dataPort); + addrInput.value = ''; labelInput.value = ''; mcastInput.value = ''; dataPortInput.value = ''; }); addrInput.addEventListener('keydown', e => { if (e.key === 'Enter') addBtn.click(); }); @@ -2218,7 +2618,7 @@ function makeAddSourceSection() { saveBtn.textContent = 'Save list'; saveBtn.title = 'Save source list to file'; saveBtn.addEventListener('click', saveSourcesWS); - body.append(addrInput, labelInput, addBtn, saveBtn); + body.append(addrInput, labelInput, mcastInput, dataPortInput, addBtn, saveBtn); section.append(title, body); title.addEventListener('click', () => { @@ -2236,6 +2636,148 @@ function escHtml(s) { return String(s).replace(/&/g, '&').replace(//g, '>').replace(/"/g, '"'); } +/* ════════════════════════════════════════════════════════════════ + VScale menu (left-click on badge) + ════════════════════════════════════════════════════════════════ */ +let _vsMenuKey = null, _vsMenuPlotId = null; + +function showVScaleMenu(key, plotId) { + hideSignalMenu(); + // If the toolbar was open for a different plot, hide that bar first. + if (_vsMenuPlotId != null && _vsMenuPlotId !== plotId) { + const oldBar = document.getElementById('vstb-' + _vsMenuPlotId); + if (oldBar) oldBar.style.display = 'none'; + } + _vsMenuKey = key; _vsMenuPlotId = plotId; + + const menu = document.getElementById('vscale-menu'); + const vs = getVScale(_vsMenuPlotId, key); + const displayName = key.includes(':') ? key.split(':').slice(1).join(':') : key; + document.getElementById('vscale-menu-key').textContent = displayName; + + document.querySelectorAll('#vscale-mode-btns .ctx-btn').forEach(btn => + btn.classList.toggle('active', btn.dataset.mode === vs.mode)); + + // Disable Range button when signal has no defined range. + const rangeBtn = document.querySelector('#vscale-mode-btns [data-mode="range"]'); + if (rangeBtn) { + const meta = findSignalMeta(key); + const hasRange = meta && meta.rangeMin != null && meta.rangeMax != null; + rangeBtn.disabled = !hasRange; + rangeBtn.title = hasRange ? '' : 'No range defined for this signal'; + } + + const isManual = vs.mode === 'manual'; + document.getElementById('vscale-manual-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. + const dv = isManual ? vs.divValue : (vs._resolvedDiv || 1); + document.getElementById('vscale-vdiv').value = dv != null ? parseFloat(dv.toPrecision(4)) : 1; + document.getElementById('vscale-pos').value = parseFloat((vs.screenPos || 0).toPrecision(4)); + + // Move the toolbar div into this plot's vscale bar. + const bar = document.getElementById('vstb-' + plotId); + if (bar) { + bar.appendChild(menu); + bar.style.display = 'block'; + } + menu.style.display = 'block'; +} + +function hideVScaleMenu() { + const menu = document.getElementById('vscale-menu'); + // Return the menu element to body so it is detached from any plot card. + menu.style.display = 'none'; + document.body.appendChild(menu); + // Hide the vscale bar of the previously active plot. + if (_vsMenuPlotId != null) { + const bar = document.getElementById('vstb-' + _vsMenuPlotId); + if (bar) bar.style.display = 'none'; + } + _vsMenuKey = null; _vsMenuPlotId = null; +} + +/* ─── Array index picker ─────────────────────────────────────────────────── */ +let _aipOnConfirm = null, _aipOnCancel = null, _aipMaxIdx = 0; + +function showArrayIdxPicker(sigKey, n, onConfirm, onCancel) { + _aipOnConfirm = onConfirm; _aipOnCancel = onCancel; _aipMaxIdx = n - 1; + const menu = document.getElementById('array-idx-picker'); + const displayName = sigKey.includes(':') ? sigKey.split(':').slice(1).join(':') : sigKey; + document.getElementById('aip-sig').textContent = displayName; + document.getElementById('aip-range').textContent = '(0 – ' + (n - 1) + ')'; + const idxInput = document.getElementById('aip-idx'); + idxInput.max = n - 1; idxInput.value = 0; + menu.style.display = 'block'; + // Centre the picker on screen. + const mw = menu.offsetWidth || 220, mh = menu.offsetHeight || 120; + menu.style.left = Math.round((window.innerWidth - mw) / 2) + 'px'; + menu.style.top = Math.round((window.innerHeight - mh) / 3) + 'px'; + idxInput.focus(); idxInput.select(); +} + +function _aipConfirm() { + const idxInput = document.getElementById('aip-idx'); + const idx = Math.max(0, Math.min(_aipMaxIdx, parseInt(idxInput.value, 10) || 0)); + document.getElementById('array-idx-picker').style.display = 'none'; + if (_aipOnConfirm) _aipOnConfirm(idx); + _aipOnConfirm = _aipOnCancel = null; +} + +function _aipCancel() { + document.getElementById('array-idx-picker').style.display = 'none'; + if (_aipOnCancel) _aipOnCancel(); + _aipOnConfirm = _aipOnCancel = null; +} + +function initArrayIdxPicker() { + document.getElementById('aip-ok').addEventListener('click', _aipConfirm); + document.getElementById('aip-cancel').addEventListener('click', _aipCancel); + document.getElementById('aip-idx').addEventListener('keydown', e => { + if (e.key === 'Enter') _aipConfirm(); + if (e.key === 'Escape') _aipCancel(); + }); +} + +function initVScaleMenu() { + document.querySelectorAll('#vscale-mode-btns .ctx-btn').forEach(btn => { + btn.addEventListener('click', () => { + if (!_vsMenuKey) return; + const vs = getVScale(_vsMenuPlotId, _vsMenuKey); + const newMode = btn.dataset.mode; + 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) + document.getElementById('vscale-vdiv').value = parseFloat(vs.divValue.toPrecision(4)); + document.getElementById('vscale-pos').value = parseFloat((vs.screenPos || 0).toPrecision(4)); + } + vs.mode = newMode; + document.querySelectorAll('#vscale-mode-btns .ctx-btn').forEach(b => b.classList.remove('active')); + btn.classList.add('active'); + const isManual = vs.mode === 'manual'; + document.getElementById('vscale-manual-row').style.display = isManual ? 'flex' : 'none'; + document.getElementById('vscale-pos-row').style.display = isManual ? 'flex' : 'none'; + refreshPlotForKey(_vsMenuKey); + }); + }); + document.getElementById('vscale-vdiv').addEventListener('input', e => { + if (!_vsMenuKey) return; + const vs = getVScale(_vsMenuPlotId, _vsMenuKey); + vs.divValue = Math.max(parseFloat(e.target.value) || 1, 1e-30); + refreshPlotForKey(_vsMenuKey); + }); + // "Position (div)" moves the marker and signal together on screen. + document.getElementById('vscale-pos').addEventListener('input', e => { + if (!_vsMenuKey) return; + const vs = getVScale(_vsMenuPlotId, _vsMenuKey); + vs.screenPos = Math.max(-4, Math.min(4, parseFloat(e.target.value) || 0)); + refreshPlotForKey(_vsMenuKey); + }); + document.getElementById('btn-vscale-close').addEventListener('click', hideVScaleMenu); +} + /* ════════════════════════════════════════════════════════════════ Signal style context menu ════════════════════════════════════════════════════════════════ */ @@ -2307,7 +2849,7 @@ function initSignalMenu() { document.addEventListener('click', e => { if (!e.target.closest('#sig-ctx-menu') && !e.target.closest('.sig-badge')) hideSignalMenu(); }); - document.addEventListener('keydown', e => { if (e.key === 'Escape') hideSignalMenu(); }); + document.addEventListener('keydown', e => { if (e.key === 'Escape') { hideSignalMenu(); hideVScaleMenu(); } }); } /* ════════════════════════════════════════════════════════════════ @@ -2449,6 +2991,8 @@ setInterval(() => { if (statsOpen) renderStats(); }, 1000); buildLayoutMenu(); applyLayout('l1x1'); buildSidebar(); // show "Add Source" section even before WS connection +initArrayIdxPicker(); +initVScaleMenu(); initSignalMenu(); document.getElementById('btn-csv-all').addEventListener('click', exportAllCSV); document.getElementById('btn-stats').addEventListener('click', toggleStats); diff --git a/Client/WebUI/static/index.html b/Client/WebUI/static/index.html index f7d6709..03ba106 100644 --- a/Client/WebUI/static/index.html +++ b/Client/WebUI/static/index.html @@ -158,6 +158,39 @@ 4px
+ + + + \ No newline at end of file diff --git a/Client/WebUI/static/style.css b/Client/WebUI/static/style.css index d5bc57d..40f02af 100644 --- a/Client/WebUI/static/style.css +++ b/Client/WebUI/static/style.css @@ -311,6 +311,48 @@ input[type=range].trig-range::-webkit-slider-thumb { } .ctx-range { width:80px; } .ctx-range-val { font-size:11px; color:var(--mauve); min-width:26px; } +.ctx-num { + width:90px; background:var(--surface0); border:1px solid var(--surface1); border-radius:4px; + color:var(--text); font-size:11px; padding:2px 6px; +} +.ctx-num:focus { outline:none; border-color:var(--accent); } +.ctx-btn:disabled { opacity:0.35; cursor:not-allowed; border-color:var(--surface1); } + +/* ── Array index picker ─────────────────────────────────────────── */ +#array-idx-picker { + position:fixed; z-index:300; + background:var(--mantle); border:1px solid var(--surface1); border-radius:var(--radius); + box-shadow:0 8px 24px rgba(0,0,0,0.6); padding:10px; min-width:200px; +} + +/* ── VScale toolbar (embedded in plot card) ──────────────────────── */ +#vscale-menu { + flex-shrink:0; background:rgba(17,17,27,0.92); border-top:1px solid var(--surface1); + padding:3px 8px; +} +.vstb-header { + display:flex; align-items:center; gap:8px; flex-wrap:nowrap; overflow-x:auto; + scrollbar-width:none; +} +.vstb-header::-webkit-scrollbar { display:none; } +.vstb-label { font-size:11px; color:var(--subtext0); white-space:nowrap; flex-shrink:0; } +.vstb-lbl { font-size:10px; color:var(--overlay0); white-space:nowrap; } +.vstb-close { + margin-left:auto; flex-shrink:0; + background:transparent; border:none; color:var(--overlay0); + cursor:pointer; font-size:11px; padding:0 3px; line-height:1; + transition:color var(--transition); +} +.vstb-close:hover { color:var(--red); } +.plot-vscale-bar { display:none; } + +/* ── Badge vscale info & active state ───────────────────────────── */ +.vscale-info { + font-size:9px; color:var(--overlay0); font-family:monospace; margin-left:2px; white-space:nowrap; +} +.sig-badge { cursor:pointer; } +.sig-badge-active { outline:1px solid rgba(255,255,255,0.35); background:rgba(88,91,112,0.9); } +.sig-badge-active .vscale-info { color:var(--subtext0); } /* ── Source groups ────────────────────────────────────────────── */ .source-group { margin-bottom:2px; } diff --git a/Client/WebUI/udpclient.go b/Client/WebUI/udpclient.go index 929706c..0a55027 100644 --- a/Client/WebUI/udpclient.go +++ b/Client/WebUI/udpclient.go @@ -107,6 +107,7 @@ func (u *UDPClient) runSession() error { reassembler := NewReassembler(2 * time.Second) buf := make([]byte, readBufSize) var currentSigs []SignalInfo + var currentPublishMode uint8 for { conn.SetReadDeadline(time.Now().Add(silenceTimeout)) @@ -142,13 +143,14 @@ func (u *UDPClient) runSession() error { switch hdr.Type { case PktConfig: - sigs, err := ParseConfig(complete) + sigs, pm, err := ParseConfig(complete) if err != nil { log.Printf("[%s] udp: parse config: %v", u.sourceID, err) continue } currentSigs = sigs - log.Printf("[%s] udp: received CONFIG (%d signals)", u.sourceID, len(sigs)) + currentPublishMode = pm + log.Printf("[%s] udp: received CONFIG (%d signals, publishMode=%d)", u.sourceID, len(sigs), pm) u.hub.SetSourceState(u.sourceID, "connected") u.hub.UpdateConfigForSource(u.sourceID, sigs) @@ -156,12 +158,14 @@ func (u *UDPClient) runSession() error { if len(currentSigs) == 0 { continue } - sample, err := ParseData(complete, currentSigs, arrivalTime) + samples, err := ParseData(complete, currentSigs, currentPublishMode, arrivalTime) if err != nil { log.Printf("[%s] udp: parse data: %v", u.sourceID, err) continue } - u.hub.PushDataForSource(u.sourceID, sample) + for _, s := range samples { + u.hub.PushDataForSource(u.sourceID, s) + } case PktACK: log.Printf("[%s] udp: received ACK (counter=%d)", u.sourceID, hdr.Counter) @@ -224,11 +228,11 @@ func (u *UDPClient) runMulticastSession() error { return err } } - currentSigs, err := ParseConfig(cfgPayload) + currentSigs, currentPublishMode, err := ParseConfig(cfgPayload) if err != nil { return err } - log.Printf("[%s] tcp: received CONFIG (%d signals)", u.sourceID, len(currentSigs)) + log.Printf("[%s] tcp: received CONFIG (%d signals, publishMode=%d)", u.sourceID, len(currentSigs), currentPublishMode) u.hub.SetSourceState(u.sourceID, "connected") u.hub.UpdateConfigForSource(u.sourceID, currentSigs) @@ -318,12 +322,14 @@ func (u *UDPClient) runMulticastSession() error { if len(currentSigs) == 0 { continue } - sample, parseErr := ParseData(complete, currentSigs, arrivalTime) + samples, parseErr := ParseData(complete, currentSigs, currentPublishMode, arrivalTime) if parseErr != nil { log.Printf("[%s] multicast: parse data: %v", u.sourceID, parseErr) continue } - u.hub.PushDataForSource(u.sourceID, sample) + for _, s := range samples { + u.hub.PushDataForSource(u.sourceID, s) + } } select { diff --git a/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp b/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp index d2102cb..c630833 100644 --- a/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp +++ b/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp @@ -127,6 +127,18 @@ UDPStreamer::UDPStreamer() : syncTimestamp = 0u; clientConnected = false; packetCounter = 0u; + maxBatchCount = 0u; + singleCycleWireBytes = 0u; + fixedWireBytes = 0u; + lastPublishTs = 0u; + accumBuffer = NULL_PTR(uint8 *); + accumTimestamps = NULL_PTR(uint64 *); + accumFill = 0u; + readyTimestamps = NULL_PTR(uint64 *); + scratchTimestamps = NULL_PTR(uint64 *); + readyFill = 0u; + decimateRatio = 1u; + decimateCounter = 0u; if (!dataSem.Create()) { REPORT_ERROR(ErrorManagement::FatalError, "Could not create EventSem."); @@ -157,6 +169,23 @@ UDPStreamer::~UDPStreamer() { (void) clientSocket.Close(); } + HeapI *heapAccum = GlobalObjectsDatabase::Instance()->GetStandardHeap(); + if (accumBuffer != NULL_PTR(uint8 *)) { + heapAccum->Free(reinterpret_cast(accumBuffer)); + } + if (accumTimestamps != NULL_PTR(uint64 *)) { + delete[] accumTimestamps; + accumTimestamps = NULL_PTR(uint64 *); + } + if (readyTimestamps != NULL_PTR(uint64 *)) { + delete[] readyTimestamps; + readyTimestamps = NULL_PTR(uint64 *); + } + if (scratchTimestamps != NULL_PTR(uint64 *)) { + delete[] scratchTimestamps; + scratchTimestamps = NULL_PTR(uint64 *); + } + /* Multicast-mode cleanup */ if (tcpClient != NULL_PTR(BasicTCPSocket *)) { (void) tcpClient->Close(); @@ -249,34 +278,59 @@ bool UDPStreamer::Initialise(StructuredDataI &data) { if ((publishStr.Size() == 0u) || (publishStr == "Strict")) { publishMode = UDPStreamerPublishStrict; } - else if (publishStr == "Auto") { - publishMode = UDPStreamerPublishAuto; + else if (publishStr == "Accumulate") { + publishMode = UDPStreamerPublishAccumulate; + } + else if (publishStr == "Decimate") { + publishMode = UDPStreamerPublishDecimate; } else { REPORT_ERROR(ErrorManagement::ParametersError, - "Unknown PublishingMode '%s'. Allowed: Strict|Auto.", + "Unknown PublishingMode '%s'. Allowed: Strict|Accumulate|Decimate.", publishStr.Buffer()); ok = false; } } - if (ok && (publishMode == UDPStreamerPublishAuto)) { + if (ok && (publishMode == UDPStreamerPublishAccumulate)) { + /* MinRefreshRate controls the time-based flush: flush when + * (now - lastPublishTs) >= flushPeriodTicks, or when adding one more + * sample would overflow MaxPayloadSize. Whichever fires first. */ if (!data.Read("MinRefreshRate", minRefreshRate) || (minRefreshRate <= 0.0)) { REPORT_ERROR(ErrorManagement::ParametersError, - "MinRefreshRate > 0 is required when PublishingMode = Auto."); + "MinRefreshRate > 0 is required when PublishingMode = Accumulate."); ok = false; } else { - /* Pre-compute the HRT tick count for one flush interval */ - float64 hrtFreq = static_cast(HighResolutionTimer::Frequency()); + float64 hrtFreq = static_cast(HighResolutionTimer::Frequency()); flushPeriodTicks = static_cast(hrtFreq / minRefreshRate); REPORT_ERROR(ErrorManagement::Information, - "Auto mode: MinRefreshRate=%.1f Hz, flushPeriodTicks=%llu.", + "Accumulate mode: MinRefreshRate=%.1f Hz, flushPeriodTicks=%llu.", minRefreshRate, static_cast(flushPeriodTicks)); } } + if (ok && (publishMode == UDPStreamerPublishDecimate)) { + /* Ratio: send 1 packet every Ratio Synchronise() calls. */ + uint32 ratio = 0u; + if (!data.Read("Ratio", ratio) || (ratio == 0u)) { + REPORT_ERROR(ErrorManagement::ParametersError, + "Ratio >= 1 is required when PublishingMode = Decimate."); + ok = false; + } + else { + decimateRatio = ratio; + if (decimateRatio == 1u) { + REPORT_ERROR(ErrorManagement::Warning, + "Decimate mode with Ratio=1 is equivalent to Strict mode."); + } + REPORT_ERROR(ErrorManagement::Information, + "Decimate mode: Ratio=%u (1 packet per %u RT cycle(s)).", + decimateRatio, decimateRatio); + } + } + if (ok) { StreamString mcastStr = ""; (void) data.Read("MulticastGroup", mcastStr); @@ -537,6 +591,9 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) { /* --- Pass 4: validate time-signal dimensions and compute wire sizes --- */ for (uint32 i = 0u; i < numSigs && ok; i++) { + /* Initialise accumulated flag: false until pass 5 may flip it */ + signalInfos[i].accumulated = false; + /* Compute wire byte size per element */ uint32 elemWireBytes = 0u; switch (signalInfos[i].quantType) { @@ -586,6 +643,99 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) { } } + /* --- Pass 5: Accumulate mode setup --- + * + * Scalars (numElements == 1) are tagged accumulated = true and auto-assigned + * a FullArray time reference if a primary time signal exists. numCols / numRows + * are left at 1 — the actual per-packet element count is determined at runtime + * and transmitted as a 4-byte numSamples field in the DATA payload header. + * + * Compute singleCycleWireBytes (accumulated signals) and fixedWireBytes + * (non-accumulated arrays that travel once per packet from the most-recent slot). + * Override totalWireBytes to the maximum possible DATA payload for wireBuffer + * allocation: 12 + maxBatchCount × singleCycleWireBytes + fixedWireBytes. + */ + if (ok && (publishMode == UDPStreamerPublishAccumulate)) { + + /* Find primary time signal: prefer Unit="us"/"ns", fall back to first integer scalar */ + uint32 primaryTsIdx = UDPS_NO_TIME_SIGNAL; + for (uint32 i = 0u; i < numSigs && (primaryTsIdx == UDPS_NO_TIME_SIGNAL); i++) { + if (signalInfos[i].numElements == 1u) { + if ((signalInfos[i].unit == "us") || (signalInfos[i].unit == "ns")) { + primaryTsIdx = i; + } + } + } + if (primaryTsIdx == UDPS_NO_TIME_SIGNAL) { + for (uint32 i = 0u; i < numSigs && (primaryTsIdx == UDPS_NO_TIME_SIGNAL); i++) { + if (signalInfos[i].numElements == 1u) { + TypeDescriptor td = signalInfos[i].type; + if ((td == UnsignedInteger32Bit) || (td == UnsignedInteger64Bit) || + (td == SignedInteger32Bit) || (td == SignedInteger64Bit)) { + primaryTsIdx = i; + } + } + } + } + if (primaryTsIdx != UDPS_NO_TIME_SIGNAL) { + REPORT_ERROR(ErrorManagement::Information, + "Accumulate: primary time signal '%s' (idx=%u).", + signalInfos[primaryTsIdx].name.Buffer(), primaryTsIdx); + } + + /* Partition signals into accumulated (scalars) and fixed (arrays). + * Auto-assign FullArray time mode for scalars that had PacketTime. */ + singleCycleWireBytes = 0u; + fixedWireBytes = 0u; + for (uint32 i = 0u; i < numSigs; i++) { + if (signalInfos[i].numElements == 1u) { + signalInfos[i].accumulated = true; + singleCycleWireBytes += signalInfos[i].wireByteSize; /* = srcByteSize for 1 elem */ + /* Auto-assign time reference for non-primary, non-time scalars */ + if ((i != primaryTsIdx) && (primaryTsIdx != UDPS_NO_TIME_SIGNAL) && + (signalInfos[i].timeMode == UDPStreamerTimePacket)) { + signalInfos[i].timeMode = UDPStreamerTimeFullArray; + signalInfos[i].timeSignalIdx = primaryTsIdx; + } + } + else { + /* Non-scalar: not accumulated; wire size already computed in pass 4 */ + fixedWireBytes += signalInfos[i].wireByteSize; + } + } + + if (singleCycleWireBytes == 0u) { + REPORT_ERROR(ErrorManagement::ParametersError, + "Accumulate mode: no scalar signals found to accumulate."); + ok = false; + } + + if (ok) { + /* DATA payload: [8 HRT][4 numSamples][numSamples × singleCycle][fixed] */ + static const uint32 ACCUM_HEADER = UDPS_TIMESTAMP_BYTES + 4u; /* 12 bytes */ + if ((ACCUM_HEADER + singleCycleWireBytes + fixedWireBytes) > maxPayloadSize) { + REPORT_ERROR(ErrorManagement::ParametersError, + "Accumulate mode: even a single sample (%u B) exceeds " + "MaxPayloadSize (%u B).", + ACCUM_HEADER + singleCycleWireBytes + fixedWireBytes, + maxPayloadSize); + ok = false; + } + } + + if (ok) { + static const uint32 ACCUM_HEADER = UDPS_TIMESTAMP_BYTES + 4u; + maxBatchCount = (maxPayloadSize - ACCUM_HEADER - fixedWireBytes) / singleCycleWireBytes; + /* Override totalWireBytes: size of the largest possible DATA payload */ + totalWireBytes = ACCUM_HEADER + maxBatchCount * singleCycleWireBytes + fixedWireBytes; + REPORT_ERROR(ErrorManagement::Information, + "Accumulate mode: singleCycleWireBytes=%u, fixedWireBytes=%u, " + "maxBatchCount=%u, maxPayloadSize=%u, totalWireBytes=%u.", + singleCycleWireBytes, fixedWireBytes, + maxBatchCount, maxPayloadSize, totalWireBytes); + } + } + return ok; } @@ -602,21 +752,25 @@ bool UDPStreamer::AllocateMemory() { HeapI *heap = GlobalObjectsDatabase::Instance()->GetStandardHeap(); + /* In Accumulate mode, readyBuffer / scratchBuffer hold maxBatchCount consecutive + * snapshots instead of a single one. */ + uint32 readyBufSize = (maxBatchCount > 0u) ? (maxBatchCount * totalSrcBytes) : totalSrcBytes; + /* readyBuffer: copy of signal memory shared with background thread */ - readyBuffer = reinterpret_cast(heap->Malloc(totalSrcBytes)); + readyBuffer = reinterpret_cast(heap->Malloc(readyBufSize)); if (readyBuffer == NULL_PTR(uint8 *)) { REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate readyBuffer."); return false; } - (void) MemoryOperationsHelper::Set(readyBuffer, 0, totalSrcBytes); + (void) MemoryOperationsHelper::Set(readyBuffer, 0, readyBufSize); /* scratchBuffer: background-thread-private copy for serialization */ - scratchBuffer = reinterpret_cast(heap->Malloc(totalSrcBytes)); + scratchBuffer = reinterpret_cast(heap->Malloc(readyBufSize)); if (scratchBuffer == NULL_PTR(uint8 *)) { REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate scratchBuffer."); return false; } - (void) MemoryOperationsHelper::Set(scratchBuffer, 0, totalSrcBytes); + (void) MemoryOperationsHelper::Set(scratchBuffer, 0, readyBufSize); /* wireBuffer: serialized/quantized payload for transmission */ wireBuffer = reinterpret_cast(heap->Malloc(totalWireBytes)); @@ -635,6 +789,44 @@ bool UDPStreamer::AllocateMemory() { } } + /* --- Accumulate-mode extra buffers --- */ + if (maxBatchCount > 0u) { + /* Linear fill buffer: RT thread writes one snapshot per slot (0..maxBatchCount-1) */ + uint32 accumBufSize = maxBatchCount * totalSrcBytes; + accumBuffer = reinterpret_cast(heap->Malloc(accumBufSize)); + if (accumBuffer == NULL_PTR(uint8 *)) { + REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate accumBuffer."); + return false; + } + (void) MemoryOperationsHelper::Set(accumBuffer, 0, accumBufSize); + + /* Per-slot HRT timestamp arrays */ + accumTimestamps = new uint64[maxBatchCount]; + readyTimestamps = new uint64[maxBatchCount]; + scratchTimestamps = new uint64[maxBatchCount]; + if ((accumTimestamps == NULL_PTR(uint64 *)) || + (readyTimestamps == NULL_PTR(uint64 *)) || + (scratchTimestamps == NULL_PTR(uint64 *))) { + REPORT_ERROR(ErrorManagement::FatalError, + "Could not allocate timestamp arrays."); + return false; + } + uint32 tsBytes = maxBatchCount * static_cast(sizeof(uint64)); + (void) MemoryOperationsHelper::Set( + reinterpret_cast(accumTimestamps), 0, tsBytes); + (void) MemoryOperationsHelper::Set( + reinterpret_cast(readyTimestamps), 0, tsBytes); + (void) MemoryOperationsHelper::Set( + reinterpret_cast(scratchTimestamps), 0, tsBytes); + + accumFill = 0u; + readyFill = 0u; + + REPORT_ERROR(ErrorManagement::Information, + "Accumulate buffers: maxBatchCount=%u, accumBufSize=%u B, readyBufSize=%u B.", + maxBatchCount, accumBufSize, readyBufSize); + } + return true; } @@ -707,6 +899,14 @@ bool UDPStreamer::PrepareNextState(const char8 *const currentStateName, } } + /* Initialise the flush timestamp so the first Accumulate flush is deferred + * until MinRefreshRate elapses (not immediately on the first Synchronise). */ + if (ok && (publishMode == UDPStreamerPublishAccumulate)) { + lastPublishTs = HighResolutionTimer::Counter(); + accumFill = 0u; + readyFill = 0u; + } + /* Start the background thread (idempotent; shared by both modes) */ if (ok && (executor.GetStatus() == EmbeddedThreadI::OffState)) { executor.SetName(GetName()); @@ -724,17 +924,76 @@ bool UDPStreamer::PrepareNextState(const char8 *const currentStateName, } bool UDPStreamer::Synchronise() { - /* Capture timestamp as early as possible */ + /* Capture HRT timestamp as early as possible. */ uint64 ts = HighResolutionTimer::Counter(); - /* RT-safe copy of signal memory → readyBuffer */ - bufMutex.FastLock(TTInfiniteWait); - (void) MemoryOperationsHelper::Copy(readyBuffer, memory, totalSrcBytes); - syncTimestamp = ts; - bufMutex.FastUnLock(); + if (publishMode == UDPStreamerPublishAccumulate) { + /* --- Accumulate path --- + * + * Append this snapshot to the linear accumulation buffer, then check + * the two flush conditions (from the user spec): + * + * (a) size: accumulate_size + next_sample_size >= MaxPayloadSize + * (adding one more would overflow the UDP datagram) + * (b) time: expected_next_cycle_time - lastPublishTs >= flushPeriodTicks + * approximated as: ts - lastPublishTs >= flushPeriodTicks + * + * When either fires, the completed batch is promoted to readyBuffer / + * readyTimestamps and dataSem is posted. The background thread sends + * the ready batch without any additional timer check. */ + bufMutex.FastLock(TTInfiniteWait); + uint8 *slot = accumBuffer + (accumFill * totalSrcBytes); + (void) MemoryOperationsHelper::Copy(slot, memory, totalSrcBytes); + accumTimestamps[accumFill] = ts; + accumFill++; + uint32 filled = accumFill; + bufMutex.FastUnLock(); - /* Wake the background sender thread */ - (void) dataSem.Post(); + /* Check flush conditions (volatile read of lastPublishTs is safe on x86). */ + static const uint32 ACCUM_HEADER = UDPS_TIMESTAMP_BYTES + 4u; /* 12 bytes */ + uint32 curPayload = ACCUM_HEADER + filled * singleCycleWireBytes + fixedWireBytes; + uint32 nextPayload = curPayload + singleCycleWireBytes; + bool sizeCondition = (nextPayload >= maxPayloadSize); + bool timeCondition = ((ts - lastPublishTs) >= flushPeriodTicks); + + if (sizeCondition || timeCondition) { + bufMutex.FastLock(TTInfiniteWait); + (void) MemoryOperationsHelper::Copy( + readyBuffer, accumBuffer, filled * totalSrcBytes); + (void) MemoryOperationsHelper::Copy( + reinterpret_cast(readyTimestamps), + reinterpret_cast(accumTimestamps), + filled * static_cast(sizeof(uint64))); + readyFill = filled; + accumFill = 0u; + bufMutex.FastUnLock(); + + /* Reset the time-based deadline (volatile write). */ + lastPublishTs = ts; + (void) dataSem.Post(); + } + } + else if (publishMode == UDPStreamerPublishDecimate) { + /* --- Decimate path --- + * Post dataSem only every decimateRatio calls. */ + decimateCounter++; + if (decimateCounter >= decimateRatio) { + decimateCounter = 0u; + bufMutex.FastLock(TTInfiniteWait); + (void) MemoryOperationsHelper::Copy(readyBuffer, memory, totalSrcBytes); + syncTimestamp = ts; + bufMutex.FastUnLock(); + (void) dataSem.Post(); + } + } + else { + /* --- Strict path: post every call --- */ + bufMutex.FastLock(TTInfiniteWait); + (void) MemoryOperationsHelper::Copy(readyBuffer, memory, totalSrcBytes); + syncTimestamp = ts; + bufMutex.FastUnLock(); + (void) dataSem.Post(); + } return true; } @@ -742,17 +1001,13 @@ bool UDPStreamer::Synchronise() { ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) { ErrorManagement::ErrorType ret = ErrorManagement::NoError; - /* nextFlushTick: HRT counter target for the next Auto-mode flush. - * Declared static so it persists across Execute() calls (the framework - * calls Execute() in a tight loop for the MainStage). */ - static uint64 nextFlushTick = 0u; - if (info.GetStage() == ExecutionInfo::StartupStage) { - nextFlushTick = HighResolutionTimer::Counter(); + const char8 *modeStr = "Strict"; + if (publishMode == UDPStreamerPublishAccumulate) { modeStr = "Accumulate"; } + else if (publishMode == UDPStreamerPublishDecimate) { modeStr = "Decimate"; } REPORT_ERROR(ErrorManagement::Information, "UDPStreamer background thread started (port %u, mode %s).", - static_cast(port), - (publishMode == UDPStreamerPublishAuto) ? "Auto" : "Strict"); + static_cast(port), modeStr); } if (info.GetStage() == ExecutionInfo::MainStage) { @@ -856,43 +1111,54 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) { } if (dataReady && clientConnected) { - /* In Auto mode, only flush when the HRT flush interval has elapsed. - * This reduces send syscalls and network load by (rtHz / minRefreshRate)x - * without any changes to the RT thread or the wire protocol. - * The most-recent signal values (captured in readyBuffer) are used. */ - bool shouldSend = true; - if (publishMode == UDPStreamerPublishAuto) { - uint64 now = HighResolutionTimer::Counter(); - if (now < nextFlushTick) { - shouldSend = false; + /* Synchronise() already gates posting dataSem to the correct rate + * (size/time for Accumulate, every-Nth for Decimate, every call for + * Strict). Execute() just sends whatever is in the ready buffers. */ + if (publishMode == UDPStreamerPublishAccumulate) { + /* --- Accumulate batch send --- */ + uint32 fill = 0u; + bufMutex.FastLock(TTInfiniteWait); + fill = readyFill; + if (fill > 0u) { + (void) MemoryOperationsHelper::Copy( + scratchBuffer, readyBuffer, fill * totalSrcBytes); + (void) MemoryOperationsHelper::Copy( + reinterpret_cast(scratchTimestamps), + reinterpret_cast(readyTimestamps), + fill * static_cast(sizeof(uint64))); } - else { - /* Advance deadline by one full period (keeps phase-locked). */ - nextFlushTick += flushPeriodTicks; - /* Guard against clock drift: if we're already more than one - * period behind, reset to avoid a burst of back-to-back sends. */ - if (nextFlushTick < now) { - nextFlushTick = now + flushPeriodTicks; + bufMutex.FastUnLock(); + + if (fill > 0u) { + SerializeAccumulated(scratchBuffer, scratchTimestamps, fill); + uint32 sendBytes = UDPS_TIMESTAMP_BYTES + 4u + + fill * singleCycleWireBytes + fixedWireBytes; + packetCounter++; + if (!SendFragmented(UDPS_TYPE_DATA, packetCounter, + wireBuffer, sendBytes)) { + REPORT_ERROR(ErrorManagement::Warning, + "Failed to send Accumulate DATA packet (counter=%u).", + packetCounter); } } } - - if (shouldSend) { - /* Copy readyBuffer → scratchBuffer under brief spinlock */ + else { + /* --- Single-snapshot send (Strict or Decimate) --- */ uint64 ts = 0u; bufMutex.FastLock(TTInfiniteWait); - (void) MemoryOperationsHelper::Copy(scratchBuffer, readyBuffer, totalSrcBytes); + (void) MemoryOperationsHelper::Copy( + scratchBuffer, readyBuffer, totalSrcBytes); ts = syncTimestamp; bufMutex.FastUnLock(); - /* Serialize signal data into wireBuffer */ QuantizeAndSerialize(scratchBuffer, ts); - /* Send (fragmented if needed) */ packetCounter++; - if (!SendFragmented(UDPS_TYPE_DATA, packetCounter, wireBuffer, totalWireBytes)) { + if (!SendFragmented(UDPS_TYPE_DATA, packetCounter, + wireBuffer, totalWireBytes)) { REPORT_ERROR(ErrorManagement::Warning, - "Failed to send DATA packet (counter=%u).", packetCounter); + "Failed to send DATA packet (counter=%u).", + packetCounter); } } } @@ -919,6 +1185,152 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) { return ret; } +void UDPStreamer::SerializeAccumulated(const uint8 *src, + const uint64 *timestamps, + uint32 numSamples) { + /* Wire layout (Accumulate mode DATA payload): + * [8 bytes] : HRT of slot 0 (oldest sample) + * [4 bytes] : numSamples (uint32, little-endian) + * for each signal: + * if accumulated : numSamples elements (one per slot) + * if non-accumulated (array): one copy from the most-recent slot + */ + uint8 *dst = wireBuffer; + + /* 8-byte packet-level HRT timestamp = timestamp of the first (oldest) sample */ + (void) MemoryOperationsHelper::Copy(dst, ×tamps[0u], UDPS_TIMESTAMP_BYTES); + dst += UDPS_TIMESTAMP_BYTES; + + /* 4-byte sample count */ + (void) MemoryOperationsHelper::Copy(dst, &numSamples, 4u); + dst += 4u; + + for (uint32 i = 0u; i < numSigs; i++) { + if (signalInfos[i].accumulated) { + /* Scalar: pack one value from each slot in order */ + uint32 elemSrcBytes = signalInfos[i].srcByteSize; /* bytes for one element */ + + for (uint32 k = 0u; k < numSamples; k++) { + const uint8 *slotSrc = src + (k * totalSrcBytes) + signalInfos[i].bufferOffset; + + if (signalInfos[i].quantType == UDPStreamerQuantNone) { + (void) MemoryOperationsHelper::Copy(dst, slotSrc, elemSrcBytes); + dst += elemSrcBytes; + } + else { + float64 rawVal = 0.0; + if (signalInfos[i].type == Float32Bit) { + float32 f32 = 0.0f; + (void) MemoryOperationsHelper::Copy(&f32, slotSrc, 4u); + rawVal = static_cast(f32); + } + else { + (void) MemoryOperationsHelper::Copy(&rawVal, slotSrc, 8u); + } + float64 rMin = signalInfos[i].rangeMin; + float64 rRange = signalInfos[i].rangeMax - rMin; + if (rRange == 0.0) { rRange = 1.0; } + float64 norm = (rawVal - rMin) / rRange; + if (norm < 0.0) { norm = 0.0; } + if (norm > 1.0) { norm = 1.0; } + switch (signalInfos[i].quantType) { + case UDPStreamerQuantUint8: { + uint8 q = static_cast(norm * 255.0); + *dst = q; dst += 1u; + break; + } + case UDPStreamerQuantInt8: { + int8 q = static_cast((norm * 254.0) - 127.0); + (void) MemoryOperationsHelper::Copy(dst, &q, 1u); + dst += 1u; + break; + } + case UDPStreamerQuantUint16: { + uint16 q = static_cast(norm * 65535.0); + (void) MemoryOperationsHelper::Copy(dst, &q, 2u); + dst += 2u; + break; + } + case UDPStreamerQuantInt16: { + int16 q = static_cast((norm * 65534.0) - 32767.0); + (void) MemoryOperationsHelper::Copy(dst, &q, 2u); + dst += 2u; + break; + } + default: { + (void) MemoryOperationsHelper::Copy(dst, slotSrc, elemSrcBytes); + dst += elemSrcBytes; + break; + } + } + } + } + } + else { + /* Non-accumulated array: send from the most-recent slot */ + const uint8 *slotSrc = src + ((numSamples - 1u) * totalSrcBytes) + + signalInfos[i].bufferOffset; + + if (signalInfos[i].quantType == UDPStreamerQuantNone) { + (void) MemoryOperationsHelper::Copy(dst, slotSrc, signalInfos[i].srcByteSize); + dst += signalInfos[i].srcByteSize; + } + else { + float64 rMin = signalInfos[i].rangeMin; + float64 rRange = signalInfos[i].rangeMax - rMin; + if (rRange == 0.0) { rRange = 1.0; } + bool isSrcFloat32 = (signalInfos[i].type == Float32Bit); + uint32 nelems = signalInfos[i].numElements; + const uint8 *s = slotSrc; + + for (uint32 e = 0u; e < nelems; e++) { + float64 rawVal = 0.0; + if (isSrcFloat32) { + float32 f32 = 0.0f; + (void) MemoryOperationsHelper::Copy(&f32, s, 4u); + rawVal = static_cast(f32); + s += 4u; + } + else { + (void) MemoryOperationsHelper::Copy(&rawVal, s, 8u); + s += 8u; + } + float64 norm = (rawVal - rMin) / rRange; + if (norm < 0.0) { norm = 0.0; } + if (norm > 1.0) { norm = 1.0; } + switch (signalInfos[i].quantType) { + case UDPStreamerQuantUint8: { + uint8 q = static_cast(norm * 255.0); + *dst = q; dst += 1u; + break; + } + case UDPStreamerQuantInt8: { + int8 q = static_cast((norm * 254.0) - 127.0); + (void) MemoryOperationsHelper::Copy(dst, &q, 1u); + dst += 1u; + break; + } + case UDPStreamerQuantUint16: { + uint16 q = static_cast(norm * 65535.0); + (void) MemoryOperationsHelper::Copy(dst, &q, 2u); + dst += 2u; + break; + } + case UDPStreamerQuantInt16: { + int16 q = static_cast((norm * 65534.0) - 32767.0); + (void) MemoryOperationsHelper::Copy(dst, &q, 2u); + dst += 2u; + break; + } + default: + break; + } + } + } + } + } +} + void UDPStreamer::HandleClientCommand(const uint8 *buf, uint32 size) { if (size < static_cast(sizeof(UDPSPacketHeader))) { return; @@ -956,7 +1368,7 @@ void UDPStreamer::HandleClientCommand(const uint8 *buf, uint32 size) { static_cast(src.GetPort())); /* Send CONFIG packet */ - uint32 configBufSize = 4u + (numSigs * UDPS_SIGNAL_DESC_SIZE) + 32u; + uint32 configBufSize = 4u + (numSigs * UDPS_SIGNAL_DESC_SIZE) + 32u + 1u; HeapI *heap = GlobalObjectsDatabase::Instance()->GetStandardHeap(); uint8 *cfgBuf = reinterpret_cast(heap->Malloc(configBufSize)); if (cfgBuf != NULL_PTR(uint8 *)) { @@ -1082,6 +1494,13 @@ bool UDPStreamer::BuildConfigPayload(uint8 *buf, payloadSize += UDPS_SIGNAL_DESC_SIZE; } + /* 1 byte: publishing mode (so clients can parse DATA payloads correctly) */ + if ((payloadSize + 1u) > bufSize) { + return false; + } + buf[payloadSize] = static_cast(publishMode); + payloadSize += 1u; + return true; } diff --git a/Source/Components/DataSources/UDPStreamer/UDPStreamer.h b/Source/Components/DataSources/UDPStreamer/UDPStreamer.h index c8fc21b..e62d81a 100644 --- a/Source/Components/DataSources/UDPStreamer/UDPStreamer.h +++ b/Source/Components/DataSources/UDPStreamer/UDPStreamer.h @@ -50,13 +50,17 @@ namespace MARTe { /** * @brief Publishing mode for the background sender thread. * - * - Strict: send one UDP packet on every Synchronise() call (legacy behaviour). - * - Auto: buffer successive ticks and only flush when the HRT-based flush - * interval has elapsed. Controlled by MinRefreshRate (Hz). + * - Strict: send one packet every Synchronise() call (default). + * - Accumulate: buffer N successive snapshots and flush either when the next + * snapshot would exceed MaxPayloadSize or when 1/MinRefreshRate + * has elapsed. Scalars expand to arrays; the first scalar with + * Unit="us"/"ns" is auto-promoted as the FullArray time reference. + * - Decimate: send one packet every Ratio Synchronise() calls. */ typedef enum { - UDPStreamerPublishStrict = 0u, /**< Send on every RT cycle (default) */ - UDPStreamerPublishAuto = 1u /**< Rate-limited: flush at MinRefreshRate Hz */ + UDPStreamerPublishStrict = 0u, /**< Send on every RT cycle (default) */ + UDPStreamerPublishAccumulate = 1u, /**< Accumulate until size/time limit; then flush */ + UDPStreamerPublishDecimate = 2u /**< Send 1 packet every Ratio cycles */ } UDPStreamerPublishMode; /** @@ -100,6 +104,7 @@ struct UDPStreamerSignalInfo { uint32 srcByteSize; /**< Bytes in MARTe2 memory */ uint32 wireByteSize; /**< Bytes on the wire (may differ when quantized) */ uint32 bufferOffset; /**< Byte offset in the flat MemoryDataSourceI memory buffer */ + bool accumulated; /**< True when this scalar was expanded to flushCount elements in Auto accumulation mode */ }; /** @@ -145,59 +150,134 @@ static const uint8 UDPS_TYPECODE_FLOAT64 = 9u; static const uint8 UDPS_TYPECODE_UNKNOWN = 255u; /** - * @brief A DataSource that streams MARTe2 signals to a single UDP client. + * @brief A DataSource that streams MARTe2 signals to UDP clients. * * @details This output DataSource accepts signals from GAMs and forwards them - * asynchronously to a connected UDP client. A dedicated background thread handles - * all network I/O so that the real-time thread is only blocked for a fast spinlock + * asynchronously over the network. A dedicated background thread handles all + * network I/O so that the real-time thread is only blocked by a fast spinlock * and a memcpy during Synchronise(). * - * Protocol: - * - Client sends CONNECT → server replies with CONFIG describing all signals. - * - Server sends DATA packets (fragmented if needed) on every RT cycle. - * - Client sends ACK packets (optional, for monitoring loss). - * - Client sends DISCONNECT to terminate the session. + * Two operating modes are selected by the presence or absence of MulticastGroup: * - * Signal configuration syntax: + * @par Unicast mode (default — no MulticastGroup) + * The server opens a single UDP socket on Port. The client initiates the session + * by sending a CONNECT packet to that port. The server replies with a CONFIG + * packet on the same socket and subsequently sends DATA packets directly to the + * client's address. Any number of clients may connect sequentially (one at a + * time); a new CONNECT evicts the previous client. + * + * @par Multicast mode (MulticastGroup specified) + * The server opens a TCP listener on Port for control traffic and a UDP socket + * aimed at MulticastGroup:DataPort for data traffic. The client: + * 1. Connects to Port via TCP and sends a CONNECT packet. + * 2. Receives the CONFIG packet over TCP. + * 3. Joins the multicast group (MulticastGroup:DataPort) to receive DATA packets. + * Multiple clients may receive data simultaneously by joining the same group. + * A new TCP CONNECT evicts the previous client from the control channel; the + * multicast data stream continues regardless. + * + * @par Packet protocol (both modes) + * - Client → Server: CONNECT (initiates session), DISCONNECT (terminates session), + * ACK (optional, for packet-loss monitoring). + * - Server → Client: CONFIG (signal metadata), DATA (signal values, possibly + * fragmented into multiple datagrams if payload exceeds MaxPayloadSize). + * + * @par Top-level configuration parameters + * | Parameter | Type | Default | Description | + * |-----------------|---------|---------|-------------| + * | Port | uint16 | 44500 | TCP control port (multicast) or UDP server port (unicast). Values ≤ 1024 produce a warning. | + * | MulticastGroup | string | *(absent)* | **Enables multicast mode.** IPv4 multicast address, e.g. `"239.0.0.1"`. Must be in 224.0.0.0/4. Absent or empty = unicast. | + * | DataPort | uint16 | Port+1 | UDP port for multicast DATA datagrams. Ignored in unicast mode. Must be non-zero and differ from Port. | + * | MaxPayloadSize | uint32 | 1400 | Maximum bytes of signal payload per UDP datagram (excluding the 17-byte header). Larger signals are fragmented. | + * | PublishingMode | string | Strict | `Strict`: send one packet every Synchronise() call. `Auto`: rate-limited; flush only when MinRefreshRate interval has elapsed. | + * | MinRefreshRate | float64 | — | Required when PublishingMode = Auto. Flush frequency in Hz (e.g. 120.0). | + * | MaxBatchSize | uint32 | 1 | Optional when PublishingMode = Auto. Number of RT cycles to accumulate before flushing one packet. Scalar signals are expanded to arrays of MaxBatchSize elements; the first scalar with Unit="us" or "ns" is auto-promoted as the per-sample FullArray timestamp reference for all other scalars. When omitted or 1, the most-recent single value is sent at MinRefreshRate. | + * | CPUMask | uint32 | 0xFFFFFFFF | CPU affinity bitmask for the background thread. | + * | StackSize | uint32 | (MARTe2 default) | Stack size in bytes for the background thread. | + * + * @par Per-signal configuration parameters + * | Parameter | Type | Default | Description | + * |------------------|---------|-------------|-------------| + * | Type | string | — | MARTe2 type name (uint8, int16, float32, float64, …). Mandatory. | + * | NumberOfDimensions | uint8 | 0 | 0 = scalar, 1 = array, 2 = matrix. | + * | NumberOfElements | uint32 | 1 | Total element count (rows × cols for matrices). | + * | Unit | string | "" | Physical unit label, e.g. `"Pa"` or `"m/s"`. Transmitted in CONFIG for display purposes. | + * | RangeMin | float64 | 0.0 | Minimum of the physical range. Required when QuantizedType is not `none`. | + * | RangeMax | float64 | 1.0 | Maximum of the physical range. Required when QuantizedType is not `none`. | + * | QuantizedType | string | none | Wire quantization for float signals: `none` \| `uint8` \| `int8` \| `uint16` \| `int16`. Reduces wire bandwidth at the cost of precision. Only valid for float32/float64 signals. | + * | TimeMode | string | PacketTime | How the signal's time axis is encoded (see below). | + * | TimeSignal | string | — | Name of the signal that carries timestamps. Required when TimeMode ≠ PacketTime. | + * | SamplingRate | float64 | — | Signal sampling rate in Hz. Required when TimeMode = FirstSample or LastSample. | + * + * @par TimeMode values + * | Value | Description | + * |--------------|-------------| + * | PacketTime | Uses the HRT counter captured at Synchronise() time as the single packet timestamp. No dedicated time signal needed. | + * | FullArray | TimeSignal has the same NumberOfElements as this signal; one timestamp per element. | + * | FirstSample | TimeSignal is a scalar = timestamp of the first element; subsequent elements are spaced by 1/SamplingRate. | + * | LastSample | TimeSignal is a scalar = timestamp of the last element; elements are spaced backwards by 1/SamplingRate. | + * + * @par Example — unicast mode * 
- * +UDPStreamer1 = {
- *     Class = UDPStreamer
- *     Port = 44500              // Optional (default 44500)
- *     MaxPayloadSize = 1400     // Optional (default 1400); max payload bytes per UDP datagram
- *     CPUMask = 0x2             // Optional, affinity for background thread
- *     StackSize = 1048576       // Optional, stack size for background thread
- *     PublishingMode = "Auto"   // Optional: "Strict" (default) | "Auto"
- *     MinRefreshRate = 120      // Optional (Hz); only used when PublishingMode = "Auto"
+ * +Streamer = {
+ *     Class            = UDPStreamer
+ *     Port             = 44500
+ *     MaxPayloadSize   = 1400
+ *     PublishingMode   = "Strict"
  *     Signals = {
  *         Time = {
  *             Type = uint64
  *         }
  *         Pressure = {
- *             Type = float32
- *             NumberOfDimensions = 1
- *             NumberOfElements = 100
- *             Unit = "Pa"             // Optional
- *             RangeMin = 0.0          // Optional (required for quantization)
- *             RangeMax = 1000000.0    // Optional (required for quantization)
- *             QuantizedType = uint16  // Optional: none|uint8|int8|uint16|int16
- *             TimeMode = LastSample   // Optional: PacketTime|FullArray|FirstSample|LastSample
- *             TimeSignal = Time       // Required when TimeMode != PacketTime
- *             SamplingRate = 10000.0  // Required when TimeMode = FirstSample or LastSample
+ *             Type                = float32
+ *             NumberOfDimensions  = 1
+ *             NumberOfElements    = 100
+ *             Unit                = "Pa"
+ *             RangeMin            = 0.0
+ *             RangeMax            = 1000000.0
+ *             QuantizedType       = uint16
+ *             TimeMode            = LastSample
+ *             TimeSignal          = Time
+ *             SamplingRate        = 10000.0
  *         }
  *         Temperature = {
- *             Type = float64
- *             Unit = "K"
+ *             Type     = float64
+ *             Unit     = "degC"
  *             TimeMode = PacketTime
  *         }
  *     }
  * }
  *
* - * Notes: - * - QuantizedType is only valid for float32/float64 signals. - * - TimeMode = PacketTime uses the HRT counter captured in Synchronise(). - * - TimeMode = FullArray requires TimeSignal to have the same NumberOfElements. - * - TimeMode = FirstSample/LastSample requires a scalar TimeSignal and SamplingRate. + * @par Example — multicast mode + * 
+ * +Streamer = {
+ *     Class            = UDPStreamer
+ *     Port             = 44500           // TCP control port
+ *     MulticastGroup   = "239.0.0.1"     // Enables multicast mode
+ *     DataPort         = 44501           // UDP data port (default: Port+1)
+ *     MaxPayloadSize   = 1400
+ *     PublishingMode   = "Auto"
+ *     MinRefreshRate   = 60
+ *     Signals = {
+ *         Time = {
+ *             Type = uint64
+ *         }
+ *         Voltage = {
+ *             Type                = float32
+ *             NumberOfDimensions  = 1
+ *             NumberOfElements    = 1000
+ *             Unit                = "V"
+ *             RangeMin            = -10.0
+ *             RangeMax            = 10.0
+ *             QuantizedType       = int16
+ *             TimeMode            = FirstSample
+ *             TimeSignal          = Time
+ *             SamplingRate        = 100000.0
+ *         }
+ *     }
+ * }
+ *
*/ class UDPStreamer : public MemoryDataSourceI, public EmbeddedServiceMethodBinderI { public: @@ -214,8 +294,11 @@ public: virtual ~UDPStreamer(); /** - * @brief Parses top-level configuration parameters (Port, MaxPayloadSize, CPUMask, StackSize). - * @return true if all mandatory parameters are valid. + * @brief Parses top-level configuration parameters. + * @details Reads Port, MaxPayloadSize, CPUMask, StackSize, PublishingMode, + * MinRefreshRate, MulticastGroup, and DataPort from the configuration node. + * Sets useMulticast and dataPort when MulticastGroup is present and valid. + * @return true if all mandatory parameters are valid and consistent. */ virtual bool Initialise(StructuredDataI &data); @@ -312,6 +395,14 @@ private: */ void HandleTCPConnect(const uint8 *buf, uint32 size); + /** + * @brief Serializes an accumulated batch into wireBuffer. + * @param src Flat buffer [numSamples × totalSrcBytes], slot 0 = oldest. + * @param timestamps HRT counters per slot; timestamps[0] is the packet timestamp. + * @param numSamples Actual number of filled slots to serialize. + */ + void SerializeAccumulated(const uint8 *src, const uint64 *timestamps, uint32 numSamples); + /** * @brief Maps a MARTe2 TypeDescriptor to the UDPS_TYPECODE_* constants. */ @@ -325,6 +416,20 @@ private: UDPStreamerPublishMode publishMode; /**< Strict or Auto publishing mode */ float64 minRefreshRate; /**< Minimum flush rate (Hz) for Auto mode */ uint64 flushPeriodTicks; /**< HRT ticks per flush interval (computed from minRefreshRate) */ + /* Accumulate mode — dynamic batch parameters */ + uint32 maxBatchCount; /**< Max snapshots that fit in MaxPayloadSize (Accumulate) */ + uint32 singleCycleWireBytes; /**< Wire bytes for all accumulated signals per snapshot */ + uint32 fixedWireBytes; /**< Wire bytes for non-accumulated signals (arrays, once per packet) */ + volatile uint64 lastPublishTs; /**< HRT counter of last successful flush (Accumulate mode) */ + uint8 *accumBuffer; /**< Heap: [maxBatchCount × totalSrcBytes] linear fill */ + uint64 *accumTimestamps; /**< Heap: [maxBatchCount] HRT counter per snapshot */ + uint32 accumFill; /**< Slots filled in accumBuffer (0..maxBatchCount) */ + uint64 *readyTimestamps; /**< Heap: [maxBatchCount] HRT for completed ready batch */ + uint64 *scratchTimestamps; /**< Heap: [maxBatchCount] background-thread local copy */ + uint32 readyFill; /**< Snapshot count in the ready batch */ + /* Decimate mode */ + uint32 decimateRatio; /**< Send 1 packet every decimateRatio Synchronise() calls */ + uint32 decimateCounter; /**< Current decimate cycle counter */ /* Signal metadata */ uint32 numSigs; /**< Number of signals */ diff --git a/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp b/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp index ea58a64..279d868 100644 --- a/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp +++ b/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp @@ -236,7 +236,7 @@ bool UDPStreamerTest::TestInitialise_AutoMode_Valid() { UDPStreamer ds; ConfigurationDatabase cdb; cdb.Write("Port", 44502u); - cdb.Write("PublishingMode", "Auto"); + cdb.Write("PublishingMode", "Accumulate"); cdb.Write("MinRefreshRate", 120.0); cdb.CreateRelative("Signals"); cdb.MoveToRoot(); @@ -248,7 +248,7 @@ bool UDPStreamerTest::TestInitialise_AutoMode_MissingRefreshRate() { UDPStreamer ds; ConfigurationDatabase cdb; cdb.Write("Port", 44503u); - cdb.Write("PublishingMode", "Auto"); + cdb.Write("PublishingMode", "Accumulate"); /* MinRefreshRate intentionally omitted */ cdb.CreateRelative("Signals"); cdb.MoveToRoot(); diff --git a/Test/MARTeApp/TestApp.cfg b/Test/MARTeApp/TestApp.cfg index bea93d6..2768b6a 100644 --- a/Test/MARTeApp/TestApp.cfg +++ b/Test/MARTeApp/TestApp.cfg @@ -3,7 +3,9 @@ * * Three independent RT threads demonstrate all four UDPStreamer time modes: * - * Thread1 – "Streamer" port 44500 (scalar signals, PacketTime) + * Thread1 – "Streamer" port 44500 TCP control / 44503 UDP multicast 239.0.0.1 + * (scalar signals, auto-accumulated to arrays[10], multicast mode, + * 100 Hz flush = 10 samples per packet, 1 kHz source) * Counter – uint32 cycle counter * Time – uint32 time in microseconds (LinuxTimer, 1 kHz) * Sine1 – float32, 1 Hz, quantised to uint16 on wire (PacketTime) @@ -413,10 +415,18 @@ $TestApp = { } // ── Streamer: scalar signals, PacketTime (port 44500) ───────────────── + // Multicast mode: clients connect via TCP on port 44500 to receive the + // CONFIG packet, then join 239.0.0.1:44503 to receive DATA datagrams. + // Auto publishing ensures data is sent every RT cycle (1 kHz = 1 ms + // temporal resolution) but never faster, so the rate is bounded. +Streamer = { Class = UDPStreamer Port = 44500 + MulticastGroup = "239.0.0.1" + DataPort = 44503 MaxPayloadSize = 1400 + PublishingMode = "Accumulate" + MinRefreshRate = 100 Signals = { Counter = { Type = uint32 diff --git a/Test/MARTeApp/run.sh b/Test/MARTeApp/run.sh index 1a87e62..9408775 100755 --- a/Test/MARTeApp/run.sh +++ b/Test/MARTeApp/run.sh @@ -167,7 +167,7 @@ WEBUI_PID="" if [ "${START_WEBUI}" -eq 1 ]; then echo "==> Starting WebUI on http://localhost:8080..." "${WEBUI_BIN}" \ - --source "Streamer@127.0.0.1:44500" \ + --source "Streamer@127.0.0.1:44500/239.0.0.1:44503" \ --source "FastStreamer@127.0.0.1:44501" \ --source "FullArrStreamer@127.0.0.1:44502" \ --listen :8080 &