diff --git a/Client/WebUI/hub.go b/Client/WebUI/hub.go index cddde21..1a3a751 100644 --- a/Client/WebUI/hub.go +++ b/Client/WebUI/hub.go @@ -3,8 +3,8 @@ package main import ( "encoding/json" "log" + "math" "net/http" - "sync" "time" "github.com/gorilla/websocket" @@ -59,7 +59,6 @@ func (c *wsClient) readPump() { if err != nil { break } - // Handle client messages (ping, setWindow, etc.) – currently just log. var env map[string]interface{} if json.Unmarshal(msg, &env) == nil { if t, ok := env["type"].(string); ok { @@ -70,6 +69,28 @@ func (c *wsClient) readPump() { case c.send <- resp: default: } + case "addSource": + label, _ := env["label"].(string) + addr, _ := env["addr"].(string) + if addr != "" { + select { + case c.hub.commandCh <- hubCmd{op: "wsAddSource", label: label, addr: addr}: + default: + } + } + case "removeSource": + id, _ := env["id"].(string) + if id != "" { + select { + case c.hub.commandCh <- hubCmd{op: "wsRemoveSource", sourceID: id}: + default: + } + } + case "saveSources": + select { + case c.hub.commandCh <- hubCmd{op: "wsSaveSources"}: + default: + } } } } @@ -82,80 +103,109 @@ func (c *wsClient) readPump() { var upgrader = websocket.Upgrader{ ReadBufferSize: 4096, WriteBufferSize: 64 * 1024, - CheckOrigin: func(r *http.Request) bool { return true }, + CheckOrigin: func(r *http.Request) bool { return true }, } -// Hub is the central data broker between the UDP client and WebSocket clients. -type Hub struct { - mu sync.RWMutex - signals []SignalInfo - configJS []byte // cached JSON config message - configSeq uint64 // incremented on every UpdateConfig call +// sourceHubState holds all data for one active data source. +// Only accessed from the Run() goroutine. +type sourceHubState struct { + id, label, addr, connState string + signals []SignalInfo + configJS []byte + // Time-signal calibration — only accessed from Run() goroutine. + timeSigCalib map[string]float64 + configSeq uint64 + configSeqAtCalib uint64 +} + +// taggedSample is a DataSample annotated with its source ID. +type taggedSample struct { + sourceID string + sample DataSample +} + +// hubCmd carries a command to the Run() goroutine. +type hubCmd struct { + op string // "addSource","removeSource","setSourceState","updateConfig", + // "wsAddSource","wsRemoveSource","wsSaveSources" + sourceID string + label string + addr string + state string + sigs []SignalInfo +} + +// Hub is the central broker between UDP clients and WebSocket clients. +// All map state is accessed exclusively from the Run() goroutine. +type Hub struct { clients map[*wsClient]bool register chan *wsClient unregister chan *wsClient - broadcastCh chan []byte // all sends go through Run() to avoid races on c.send + broadcastCh chan []byte + dataCh chan taggedSample + commandCh chan hubCmd - dataCh chan DataSample // incoming samples from UDP goroutine - - // Time-signal calibration: only accessed from the Run() goroutine. - // For temporal-array signals (TimeMode=FirstSample/LastSample), the - // MARTe2 Time signal value (uint32 microseconds from start) is used as - // the timing anchor. We calibrate a per-signal wall-clock offset once - // on the first data packet so that subsequent packets are stamped purely - // from the embedded timer value → perfect continuity, no jitter gaps. - timeSigCalib map[string]float64 // key=time-signal name, value=wallTime-timerSecs offset - configSeqAtCalib uint64 // configSeq value when timeSigCalib was last reset + sm *SourceManager // set after construction; used for WS-initiated source changes } // NewHub creates an initialised Hub. func NewHub() *Hub { return &Hub{ - clients: make(map[*wsClient]bool), - register: make(chan *wsClient, 8), - unregister: make(chan *wsClient, 8), - broadcastCh: make(chan []byte, 64), - dataCh: make(chan DataSample, 256), - timeSigCalib: make(map[string]float64), + clients: make(map[*wsClient]bool), + register: make(chan *wsClient, 8), + unregister: make(chan *wsClient, 8), + broadcastCh: make(chan []byte, 64), + dataCh: make(chan taggedSample, 256), + commandCh: make(chan hubCmd, 64), } } -// UpdateConfig stores a new signal config and broadcasts it to all WS clients. -func (h *Hub) UpdateConfig(sigs []SignalInfo) { - msg, err := json.Marshal(map[string]interface{}{ - "type": "config", - "signals": sigs, - }) - if err != nil { - log.Printf("hub: marshal config: %v", err) - return - } - h.mu.Lock() - h.signals = sigs - h.configJS = msg - h.configSeq++ - h.mu.Unlock() - - h.broadcast(msg) -} - -// PushData enqueues a data sample for broadcasting to WebSocket clients. -func (h *Hub) PushData(s DataSample) { +// AddSource notifies the Hub that a new source has been registered. +func (h *Hub) AddSource(id, label, addr string) { select { - case h.dataCh <- s: + case h.commandCh <- hubCmd{op: "addSource", sourceID: id, label: label, addr: addr}: + default: + } +} + +// RemoveSource notifies the Hub that a source has been removed. +func (h *Hub) RemoveSource(id string) { + select { + case h.commandCh <- hubCmd{op: "removeSource", sourceID: id}: + default: + } +} + +// SetSourceState updates the connection state of a source. +func (h *Hub) SetSourceState(id, state string) { + select { + case h.commandCh <- hubCmd{op: "setSourceState", sourceID: id, state: state}: + default: + } +} + +// UpdateConfigForSource stores a new signal config for a source and broadcasts it. +func (h *Hub) UpdateConfigForSource(sourceID string, sigs []SignalInfo) { + select { + case h.commandCh <- hubCmd{op: "updateConfig", sourceID: sourceID, sigs: sigs}: + default: + } +} + +// PushDataForSource enqueues a data sample from a specific source. +func (h *Hub) PushDataForSource(sourceID string, s DataSample) { + select { + case h.dataCh <- taggedSample{sourceID: sourceID, sample: s}: default: - // Drop if buffer full to avoid blocking the UDP goroutine. } } // broadcast enqueues a message for delivery to all WebSocket clients. -// All actual sends happen inside Run() to avoid concurrent access to c.send. func (h *Hub) broadcast(msg []byte) { select { case h.broadcastCh <- msg: default: - // Drop if the broadcast queue is full. } } @@ -166,135 +216,244 @@ func (h *Hub) HandleWebSocket(w http.ResponseWriter, r *http.Request) { log.Printf("ws upgrade: %v", err) return } - c := &wsClient{ - hub: h, - conn: conn, - send: make(chan []byte, 64), - } + c := &wsClient{hub: h, conn: conn, send: make(chan []byte, 64)} h.register <- c - go c.writePump() go c.readPump() } -// Run is the hub's main goroutine. It must be started with go hub.Run(). +// buildSourcesMsg serialises the current source list as a JSON "sources" message. +func buildSourcesMsg(sm map[string]*sourceHubState) []byte { + type srcInfo struct { + ID string `json:"id"` + Label string `json:"label"` + Addr string `json:"addr"` + State string `json:"state"` + } + list := make([]srcInfo, 0, len(sm)) + for _, src := range sm { + list = append(list, srcInfo{ID: src.id, Label: src.label, Addr: src.addr, State: src.connState}) + } + msg, _ := json.Marshal(map[string]interface{}{"type": "sources", "sources": list}) + return msg +} + +// Run is the hub's main goroutine. Must be started with go hub.Run(). func (h *Hub) Run() { - // Batch data at ≤30 Hz. ticker := time.NewTicker(time.Second / 30) defer ticker.Stop() - // Accumulate samples between ticks. - pending := make([]DataSample, 0, 64) + sourcesMap := make(map[string]*sourceHubState) + var sourcesMsg []byte + + // pending[sourceID] accumulates samples between 30 Hz ticks. + pending := make(map[string][]DataSample) + + rebuildSources := func() { + sourcesMsg = buildSourcesMsg(sourcesMap) + h.broadcast(sourcesMsg) + } for { select { case c := <-h.register: - h.mu.Lock() h.clients[c] = true - cfg := h.configJS - h.mu.Unlock() - // Send current config immediately if we have one. - if cfg != nil { - select { - case c.send <- cfg: - default: + // Send current state to the new client. + if sourcesMsg != nil { + select { case c.send <- sourcesMsg: default: } + } + for _, src := range sourcesMap { + if src.configJS != nil { + select { case c.send <- src.configJS: default: } } } case c := <-h.unregister: - h.mu.Lock() if _, ok := h.clients[c]; ok { delete(h.clients, c) close(c.send) } - h.mu.Unlock() case msg := <-h.broadcastCh: - h.mu.RLock() for c := range h.clients { - select { - case c.send <- msg: - default: + select { case c.send <- msg: default: } + } + + case cmd := <-h.commandCh: + switch cmd.op { + case "addSource": + sourcesMap[cmd.sourceID] = &sourceHubState{ + id: cmd.sourceID, + label: cmd.label, + addr: cmd.addr, + connState: "connecting", + timeSigCalib: make(map[string]float64), + } + rebuildSources() + + case "removeSource": + delete(sourcesMap, cmd.sourceID) + delete(pending, cmd.sourceID) + rebuildSources() + + case "setSourceState": + if src, ok := sourcesMap[cmd.sourceID]; ok { + src.connState = cmd.state + rebuildSources() + } + + case "updateConfig": + src, ok := sourcesMap[cmd.sourceID] + if !ok { + continue + } + src.signals = cmd.sigs + src.configSeq++ + cfgMsg, err := json.Marshal(map[string]interface{}{ + "type": "config", + "sourceId": cmd.sourceID, + "signals": cmd.sigs, + }) + if err != nil { + log.Printf("hub: marshal config: %v", err) + continue + } + src.configJS = cfgMsg + h.broadcast(cfgMsg) + + case "wsAddSource": + if h.sm != nil { + go func(label, addr string) { h.sm.Add(label, addr) }(cmd.label, cmd.addr) + } + + case "wsRemoveSource": + if h.sm != nil { + go func(id string) { h.sm.Remove(id) }(cmd.sourceID) + } + + case "wsSaveSources": + if h.sm != nil { + if err := h.sm.Save(); err != nil { + log.Printf("hub: save sources: %v", err) + } } } - h.mu.RUnlock() - case s := <-h.dataCh: - pending = append(pending, s) + case ts := <-h.dataCh: + pending[ts.sourceID] = append(pending[ts.sourceID], ts.sample) case <-ticker.C: - if len(pending) == 0 { - continue - } - h.mu.RLock() - sigs := h.signals - noClients := len(h.clients) == 0 - h.mu.RUnlock() - - if noClients || len(sigs) == 0 { - pending = pending[:0] - continue - } - - msg := h.buildDataMessage(pending, sigs) - pending = pending[:0] - if msg != nil { - h.broadcast(msg) + for srcID, samples := range pending { + if len(samples) == 0 { + continue + } + src, ok := sourcesMap[srcID] + if !ok || len(src.signals) == 0 || len(h.clients) == 0 { + pending[srcID] = pending[srcID][:0] + continue + } + msg := h.buildDataMessageForSource(src, samples) + pending[srcID] = pending[srcID][:0] + if msg != nil { + h.broadcast(msg) + } } } } } -// maxBatchPoints is the maximum number of points sent per signal per 30 Hz display tick. -// For temporal-array signals (e.g. 1 MSps packed as 1000 samples/packet), the expanded -// sample stream is decimated to this limit before transmission to WebSocket clients. -const maxBatchPoints = 2000 +// ─── Data serialisation ─────────────────────────────────────────────────────── -// sigData carries one signal's worth of time+value pairs in a single batch message. +// maxScalarPoints caps scalar/spatial-array signals per 30 Hz tick. +// At typical cycle rates (≤10 kHz) a tick accumulates at most ~333 samples, +// so this cap is almost never hit. +const maxScalarPoints = 2000 + +// maxTemporalPoints caps temporal-array (packed-burst) signals per 30 Hz tick. +// Raised significantly vs scalars because temporal arrays carry high-frequency +// waveforms: at 5 Msps / 5 kHz update rate a tick produces ~167 k samples; +// sending 20 k points limits the wire to ~320 KB/ch/tick while giving a +// minimum visible Δt of ≈ 1.6 µs (vs ≈16 µs with the old 2 k cap). +const maxTemporalPoints = 20000 + +// lttbDecimate reduces (tIn, vIn) to at most threshold representative points +// using the Largest-Triangle-Three-Buckets algorithm. +// Returns the original slices unchanged when len(tIn) ≤ threshold. +func lttbDecimate(tIn, vIn []float64, threshold int) ([]float64, []float64) { + n := len(tIn) + if n <= threshold || threshold < 3 { + return tIn, vIn + } + outT := make([]float64, threshold) + outV := make([]float64, threshold) + outT[0], outV[0] = tIn[0], vIn[0] + outT[threshold-1], outV[threshold-1] = tIn[n-1], vIn[n-1] + + every := float64(n-2) / float64(threshold-2) + a := 0 + for i := 0; i < threshold-2; i++ { + // Centroid of the next bucket + avgS := int(float64(i+1)*every) + 1 + avgE := int(float64(i+2)*every) + 1 + if avgE > n { + avgE = n + } + avgT, avgV, cnt := 0.0, 0.0, 0 + for j := avgS; j < avgE; j++ { + avgT += tIn[j]; avgV += vIn[j]; cnt++ + } + if cnt > 0 { + avgT /= float64(cnt); avgV /= float64(cnt) + } + // Pick the point in the current bucket that forms the largest triangle + rS := int(float64(i)*every) + 1 + rE := int(float64(i+1)*every) + 1 + if rE > n { + rE = n + } + maxArea, next := -1.0, rS + aT, aV := tIn[a], vIn[a] + for j := rS; j < rE; j++ { + area := math.Abs((aT-avgT)*(vIn[j]-aV) - (aT-tIn[j])*(avgV-aV)) + if area > maxArea { + maxArea = area; next = j + } + } + outT[i+1], outV[i+1] = tIn[next], vIn[next] + a = next + } + return outT, outV +} + +// sigData carries one signal's worth of time+value pairs. type sigData struct { - T []float64 `json:"t"` // unix seconds - V []float64 `json:"v"` // physical values + T []float64 `json:"t"` + V []float64 `json:"v"` } -// dataMsg is the JSON envelope for batched data sent to WebSocket clients. -// Each signal carries its own time axis so that temporal arrays (packed sample -// bursts with a known sampling rate) and scalar signals can coexist cleanly. +// dataMsg is the JSON envelope sent to WebSocket clients. type dataMsg struct { - Type string `json:"type"` - Signals map[string]sigData `json:"signals"` + Type string `json:"type"` + SourceID string `json:"sourceId"` + Signals map[string]sigData `json:"signals"` } -// buildDataMessage merges a batch of DataSamples into one JSON message. -// -// Three cases are handled: -// -// 1. Temporal array (NumElements > 1, TimeMode == FirstSample or LastSample): -// The N samples in each packet represent a contiguous time burst at SamplingRate. -// The embedded TimeSignal value (uint32 microseconds) is used as the timing -// anchor for each burst. A wall-clock offset is calibrated once on the first -// packet so that abs-time stays consistent with other signals. -// The full expanded stream is decimated to maxBatchPoints if needed. -// -// 2. Scalar signal (NumElements == 1): -// One {t, v} pair per packet – wall arrival time used as timestamp. -// -// 3. Spatial / PacketTime array (NumElements > 1, TimeMode == 0): -// Each element is tracked as a separate stream keyed "sig[i]", with wall -// arrival time as the shared timestamp. -func (h *Hub) buildDataMessage(batch []DataSample, sigs []SignalInfo) []byte { +// buildDataMessageForSource serialises a batch of samples for one source. +// Signal keys in the output are prefixed with "sourceId:" so the browser can +// store them in a single flat buffer map without collision. +func (h *Hub) buildDataMessageForSource(src *sourceHubState, batch []DataSample) []byte { if len(batch) == 0 { return nil } - // Reset time-signal calibration whenever the config has changed. - h.mu.RLock() - seq := h.configSeq - h.mu.RUnlock() - if seq != h.configSeqAtCalib { - h.configSeqAtCalib = seq - h.timeSigCalib = make(map[string]float64) + // Reset time-signal calibration whenever the signal config changed. + if src.configSeq != src.configSeqAtCalib { + src.configSeqAtCalib = src.configSeq + src.timeSigCalib = make(map[string]float64) } + sigs := src.signals + pfx := src.id + ":" out := make(map[string]sigData, len(sigs)*2) for _, sig := range sigs { @@ -303,19 +462,15 @@ func (h *Hub) buildDataMessage(batch []DataSample, sigs []SignalInfo) []byte { switch { case isTemporal: - // Resolve time signal (scalar that gives the anchor time in microseconds). hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs) var timeSigName string if hasTimeSig { timeSigName = sigs[sig.TimeSignalIdx].Name } - dt := 0.0 if sig.SamplingRate > 0 { dt = 1.0 / sig.SamplingRate } - - // Expand each packet's N samples into individual time-stamped points. allT := make([]float64, 0, len(batch)*n) allV := make([]float64, 0, len(batch)*n) @@ -324,37 +479,25 @@ func (h *Hub) buildDataMessage(batch []DataSample, sigs []SignalInfo) []byte { if !ok || len(vals) < n { continue } - - // Compute the anchor timestamp for this burst. - // Prefer the embedded time-signal value (microseconds) so that - // consecutive bursts are perfectly contiguous regardless of jitter. var anchorTime float64 - anchorIsFirstSample := (sig.TimeMode == TimeModeFirstSample) - + anchorIsFirstSample := sig.TimeMode == TimeModeFirstSample if hasTimeSig { tVals, tOk := s.Values[timeSigName] if tOk && len(tVals) >= 1 { - // Time signal is uint32 microseconds from system start. timerS := tVals[0] * 1e-6 wallT := float64(s.WallTime.UnixNano()) / 1e9 - - // Calibrate the wall-clock offset once per session so that - // anchor times can be expressed as absolute Unix timestamps. - if _, exists := h.timeSigCalib[timeSigName]; !exists { - h.timeSigCalib[timeSigName] = wallT - timerS + if _, exists := src.timeSigCalib[timeSigName]; !exists { + src.timeSigCalib[timeSigName] = wallT - timerS } - anchorTime = h.timeSigCalib[timeSigName] + timerS + anchorTime = src.timeSigCalib[timeSigName] + timerS } else { - // Time signal missing in this packet – fall back to wall clock. anchorTime = float64(s.WallTime.UnixNano()) / 1e9 - anchorIsFirstSample = false // wallT = last sample + anchorIsFirstSample = false } } else { - // No time signal configured – use wall arrival as last-sample anchor. anchorTime = float64(s.WallTime.UnixNano()) / 1e9 anchorIsFirstSample = false } - for k := 0; k < n; k++ { var t float64 if anchorIsFirstSample { @@ -367,24 +510,10 @@ func (h *Hub) buildDataMessage(batch []DataSample, sigs []SignalInfo) []byte { } } - // Decimate to maxBatchPoints if necessary. - step := 1 - if len(allT) > maxBatchPoints { - step = len(allT) / maxBatchPoints - if step < 1 { - step = 1 - } - } - decimT := make([]float64, 0, len(allT)/step+1) - decimV := make([]float64, 0, len(allV)/step+1) - for i := 0; i < len(allT); i += step { - decimT = append(decimT, allT[i]) - decimV = append(decimV, allV[i]) - } - out[sig.Name] = sigData{T: decimT, V: decimV} + decimT, decimV := lttbDecimate(allT, allV, maxTemporalPoints) + out[pfx+sig.Name] = sigData{T: decimT, V: decimV} case n == 1: - // Scalar signal: one sample per packet. ts := make([]float64, 0, len(batch)) vs := make([]float64, 0, len(batch)) for _, s := range batch { @@ -395,12 +524,12 @@ func (h *Hub) buildDataMessage(batch []DataSample, sigs []SignalInfo) []byte { ts = append(ts, float64(s.WallTime.UnixNano())/1e9) vs = append(vs, vals[0]) } - out[sig.Name] = sigData{T: ts, V: vs} + out[pfx+sig.Name] = sigData{T: ts, V: vs} default: - // Spatial / PacketTime array: one stream per element, keyed "sig[i]". + // Spatial / PacketTime array: one stream per element. for i := 0; i < n; i++ { - key := arrayKey(sig.Name, i) + key := pfx + arrayKey(sig.Name, i) ts := make([]float64, 0, len(batch)) vs := make([]float64, 0, len(batch)) for _, s := range batch { @@ -416,10 +545,7 @@ func (h *Hub) buildDataMessage(batch []DataSample, sigs []SignalInfo) []byte { } } - result, err := json.Marshal(dataMsg{ - Type: "data", - Signals: out, - }) + result, err := json.Marshal(dataMsg{Type: "data", SourceID: src.id, Signals: out}) if err != nil { log.Printf("hub: marshal data: %v", err) return nil diff --git a/Client/WebUI/main.go b/Client/WebUI/main.go index 9c6b953..71205b3 100644 --- a/Client/WebUI/main.go +++ b/Client/WebUI/main.go @@ -2,11 +2,13 @@ package main import ( "embed" + "errors" "flag" "fmt" "io/fs" "log" "net/http" + "os" ) // version is set at build time via -ldflags "-X main.version=..." @@ -15,19 +17,36 @@ var version = "dev" //go:embed static var staticFiles embed.FS +// multiFlag allows a flag to be repeated: --source a --source b +type multiFlag []string + +func (f *multiFlag) String() string { return fmt.Sprintf("%v", []string(*f)) } +func (f *multiFlag) Set(v string) error { *f = append(*f, v); return nil } + func main() { - streamerAddr := flag.String("streamer", "127.0.0.1:44500", "MARTe2 UDP streamer address (host:port)") + var sourceArgs multiFlag + flag.Var(&sourceArgs, "source", `Data source in the form [label@]host:port (repeatable)`) + sourcesFile := flag.String("sources-file", "", "JSON file for persistent source list (load on start, save target)") listenAddr := flag.String("listen", ":8080", "HTTP listen address") - clientPort := flag.Int("clientport", 44900, "Local UDP port to bind for streamer data") flag.Parse() hub := NewHub() + sm := NewSourceManager(hub, *sourcesFile) + hub.sm = sm + go hub.Run() - udpClient := NewUDPClient(*streamerAddr, *clientPort, hub) - go udpClient.Run() + // Load sources from file first (if specified), then add any CLI --source flags. + if *sourcesFile != "" { + if err := sm.Load(*sourcesFile); err != nil && !errors.Is(err, os.ErrNotExist) { + log.Printf("sources-file load: %v", err) + } + } + for _, arg := range sourceArgs { + label, addr := ParseSourceArg(arg) + sm.Add(label, addr) + } - // Serve the embedded static/ directory (index.html, style.css, app.js). sub, err := fs.Sub(staticFiles, "static") if err != nil { log.Fatalf("static sub-fs: %v", err) @@ -38,8 +57,7 @@ func main() { fmt.Fprint(w, version) }) - log.Printf("WebUI listening on %s (streamer=%s, local UDP port=%d, version=%s)", - *listenAddr, *streamerAddr, *clientPort, version) + log.Printf("WebUI listening on %s (version=%s)", *listenAddr, version) if err := http.ListenAndServe(*listenAddr, nil); err != nil { log.Fatalf("http: %v", err) } diff --git a/Client/WebUI/static/app.js b/Client/WebUI/static/app.js index 0b1b829..f2bddb4 100644 --- a/Client/WebUI/static/app.js +++ b/Client/WebUI/static/app.js @@ -14,7 +14,8 @@ const TRACE_COLORS = [ /* ════════════════════════════════════════════════════════════════ Globals ════════════════════════════════════════════════════════════════ */ -let signals = []; +// sourcesMap: id → {id, label, addr, state, signals:[]} +const sourcesMap = {}; let buffers = {}; let plots = []; let nextPlotId = 1; @@ -102,7 +103,8 @@ function connectWS() { ws.onerror = () => { }; ws.onmessage = evt => { let msg; try { msg = JSON.parse(evt.data); } catch { return; } - if (msg.type === 'config') onConfig(msg); + if (msg.type === 'sources') onSources(msg); + else if (msg.type === 'config') onConfig(msg); else if (msg.type === 'data') onData(msg); }; } @@ -147,19 +149,32 @@ function numElements(sig) { return (sig.numRows || 1) * (sig.numCols || 1); } function isTemporal(sig) { return numElements(sig) > 1 && (sig.timeMode || 0) !== 0; } function onConfig(msg) { + const sid = msg.sourceId; + if (!sid) return; + // Ensure source exists (may arrive before 'sources' message in some edge cases). + if (!sourcesMap[sid]) { + sourcesMap[sid] = { id: sid, label: sid, addr: '', state: 'connected', signals: [] }; + } + const src = sourcesMap[sid]; const newSigs = msg.signals || []; + const oldSigs = src.signals || []; const fp = s => s.name + ':' + s.typeCode + ':' + (s.numRows || 1) + ':' + (s.numCols || 1) + ':' + (s.timeMode || 0); - const changed = newSigs.length !== signals.length || newSigs.some((s, i) => fp(s) !== fp(signals[i])); - signals = newSigs; + const changed = newSigs.length !== oldSigs.length || newSigs.some((s, i) => fp(s) !== fp(oldSigs[i])); + src.signals = newSigs; if (changed) { - buffers = {}; - signals.forEach(sig => { + // Remove old buffers for this source only (prefix: "sid:"). + const prefix = sid + ':'; + Object.keys(buffers).forEach(k => { if (k.startsWith(prefix)) delete buffers[k]; }); + newSigs.forEach(sig => { const n = numElements(sig); - if (isTemporal(sig)) { buffers[sig.name] = makeBuffer(TEMPORAL_CAP); } - else if (n === 1) { buffers[sig.name] = makeBuffer(); } - else { for (let i = 0; i < n; i++) buffers[sig.name + '[' + i + ']'] = makeBuffer(); } + const base = prefix + sig.name; + if (isTemporal(sig)) { buffers[base] = makeBuffer(TEMPORAL_CAP); } + else if (n === 1) { buffers[base] = makeBuffer(); } + else { for (let i = 0; i < n; i++) buffers[base + '[' + i + ']'] = makeBuffer(); } }); - trigDisarm(); trig.snapshot = null; trig.prevVal = null; + if (trig.signal && trig.signal.startsWith(prefix)) { + trigDisarm(); trig.snapshot = null; trig.prevVal = null; + } zoomHistory.length = 0; document.getElementById('btn-zoom-back').style.display = 'none'; } @@ -345,11 +360,17 @@ function sliceTypedArrayRange(t, v, t0, t1) { // Temporal array signals have a meaningful SamplingRate; scalars return 0. // Used to prefer high-freq signals as the master time grid regardless of trace order. function getKeySamplingRate(key) { - const direct = signals.find(s => s.name === key); - if (direct) return direct.samplingRate || 0; - // Array element key like "Sig[3]" — strip the index - const sig = signals.find(s => key.startsWith(s.name + '[')); - return sig ? (sig.samplingRate || 0) : 0; + // key format: "sourceId:signalName" or "sourceId:signalName[i]" + for (const src of Object.values(sourcesMap)) { + const prefix = src.id + ':'; + if (!key.startsWith(prefix)) continue; + const localKey = key.slice(prefix.length); + const direct = (src.signals || []).find(s => s.name === localKey); + if (direct) return direct.samplingRate || 0; + const sig = (src.signals || []).find(s => localKey.startsWith(s.name + '[')); + if (sig) return sig.samplingRate || 0; + } + return 0; } // Returns a uPlot paths function for dashed/dotted lines, or null for solid (uPlot default). @@ -1230,16 +1251,23 @@ document.getElementById('btn-trig-stop').addEventListener('click', () => { function buildTrigSignalSelect() { const sel = document.getElementById('trig-signal'), cur = sel.value; sel.innerHTML = ''; - signals.forEach(sig => { - const n = numElements(sig); - if (isTemporal(sig) || n === 1) { - const o = document.createElement('option'); o.value = sig.name; o.textContent = sig.name; sel.appendChild(o); - } else { - for (let i = 0; i < n; i++) { - const key = sig.name + '[' + i + ']', o = document.createElement('option'); - o.value = key; o.textContent = key; sel.appendChild(o); + 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); + 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); + } 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); + } } - } + }); }); if (cur && [...sel.options].some(o => o.value === cur)) sel.value = cur; trig.signal = sel.value; @@ -1248,34 +1276,79 @@ function buildTrigSignalSelect() { /* ════════════════════════════════════════════════════════════════ Sidebar ════════════════════════════════════════════════════════════════ */ +const _typeNames = ['u8', 'i8', 'u16', 'i16', 'u32', 'i32', 'u64', 'i64', 'f32', 'f64']; + function buildSidebar() { const list = document.getElementById('signal-list'); list.innerHTML = ''; - if (!signals.length) { - list.innerHTML = '
No signals
'; - return; - } - const typeNames = ['u8', 'i8', 'u16', 'i16', 'u32', 'i32', 'u64', 'i64', 'f32', 'f64']; - signals.forEach(sig => { - const n = numElements(sig), temporal = isTemporal(sig); - const typeName = typeNames[sig.typeCode] || '?'; - if (n === 1 || temporal) { - list.appendChild(makeDraggable(sig.name, sig.name, temporal ? '[' + n + '] ' + typeName : typeName, sig.unit || '')); - } else { - const group = document.createElement('div'); group.className = 'array-group'; - const header = document.createElement('div'); header.className = 'array-header'; - header.innerHTML = '▶' + escHtml(sig.name) + '' - + (sig.unit ? '' + escHtml(sig.unit) + '' : '') - + '[' + n + '] ' + typeName + ''; - header.addEventListener('click', () => header.classList.toggle('open')); - const children = document.createElement('div'); children.className = 'array-children'; - for (let i = 0; i < n; i++) { - const key = sig.name + '[' + i + ']', child = makeDraggable(key, key, typeName, sig.unit || ''); - child.className = 'array-child'; children.appendChild(child); + const sources = Object.values(sourcesMap); + + sources.forEach(src => { + const sigs = src.signals || []; + const prefix = src.id + ':'; + + // Source header + const grp = document.createElement('div'); + grp.className = 'source-group'; + + const hdr = document.createElement('div'); + hdr.className = 'source-group-header'; + + const dot = document.createElement('span'); + dot.className = 'source-state-dot ' + (src.state || 'disconnected'); + + const nameEl = document.createElement('span'); + nameEl.className = 'source-name'; + nameEl.textContent = src.label || src.addr || src.id; + nameEl.title = src.addr || ''; + + const addrEl = document.createElement('span'); + addrEl.className = 'source-addr'; + if (src.label && src.addr) addrEl.textContent = src.addr; + + const rmBtn = document.createElement('button'); + rmBtn.className = 'source-remove-btn'; + rmBtn.title = 'Remove source'; + rmBtn.textContent = '×'; + rmBtn.addEventListener('click', () => removeSource(src.id)); + + hdr.append(dot, nameEl, addrEl, rmBtn); + grp.appendChild(hdr); + + sigs.forEach(sig => { + const n = numElements(sig), temporal = isTemporal(sig); + const typeName = _typeNames[sig.typeCode] || '?'; + const globalKey = prefix + sig.name; + if (n === 1 || temporal) { + grp.appendChild(makeDraggable(globalKey, sig.name, temporal ? '[' + n + '] ' + typeName : typeName, sig.unit || '')); + } else { + const group = document.createElement('div'); group.className = 'array-group'; + const header = document.createElement('div'); header.className = 'array-header'; + header.innerHTML = '▶' + escHtml(sig.name) + '' + + (sig.unit ? '' + escHtml(sig.unit) + '' : '') + + '[' + n + '] ' + typeName + ''; + header.addEventListener('click', () => header.classList.toggle('open')); + const children = document.createElement('div'); children.className = 'array-children'; + for (let i = 0; i < n; i++) { + const key = globalKey + '[' + i + ']'; + const child = makeDraggable(key, sig.name + '[' + i + ']', typeName, sig.unit || ''); + child.className = 'array-child'; children.appendChild(child); + } + group.appendChild(header); group.appendChild(children); grp.appendChild(group); } - group.appendChild(header); group.appendChild(children); list.appendChild(group); - } + }); + + list.appendChild(grp); }); + + if (!sources.length) { + const empty = document.createElement('div'); + empty.style.cssText = 'padding:16px 14px;color:var(--overlay0);font-size:12px;text-align:center;'; + empty.textContent = 'No sources configured'; + list.appendChild(empty); + } + + list.appendChild(makeAddSourceSection()); } function makeDraggable(key, label, typeName, unit) { const item = document.createElement('div'); @@ -1487,7 +1560,9 @@ function addBadge(plotId, key) { e.preventDefault(); showSignalMenu(key, plotId, e.clientX, e.clientY); }); - badge.appendChild(dot); badge.appendChild(document.createTextNode(key)); badge.appendChild(x); + // 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); c.appendChild(badge); } function removeBadge(plotId, key) { @@ -1650,6 +1725,93 @@ function setSidebar(open) { } document.getElementById('btn-sidebar').addEventListener('click', () => setSidebar(!sidebarOpen)); +/* ════════════════════════════════════════════════════════════════ + Multi-source management + ════════════════════════════════════════════════════════════════ */ +function onSources(msg) { + const srcs = msg.sources || []; + const newIds = new Set(srcs.map(s => s.id)); + // Remove sources that disappeared. + Object.keys(sourcesMap).forEach(id => { + if (!newIds.has(id)) { + const prefix = id + ':'; + Object.keys(buffers).forEach(k => { if (k.startsWith(prefix)) delete buffers[k]; }); + delete sourcesMap[id]; + } + }); + // Update or create entries. + srcs.forEach(s => { + if (!sourcesMap[s.id]) { + sourcesMap[s.id] = { id: s.id, label: s.label, addr: s.addr, state: s.state, signals: [] }; + } else { + Object.assign(sourcesMap[s.id], { label: s.label, addr: s.addr, state: s.state }); + } + }); + buildSidebar(); +} + +function addSourceWS(label, addr) { + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify({ type: 'addSource', label, addr })); + } +} + +function removeSource(id) { + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify({ type: 'removeSource', id })); + } +} + +function saveSourcesWS() { + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify({ type: 'saveSources' })); + } +} + +function makeAddSourceSection() { + const section = document.createElement('div'); + section.className = 'add-source-section'; + + const title = document.createElement('div'); + title.className = 'add-source-title'; + title.innerHTML = '▶ Add Source'; + + const body = document.createElement('div'); + body.className = 'add-source-body'; + + const addrInput = document.createElement('input'); + addrInput.className = 'add-src-input'; addrInput.type = 'text'; + addrInput.placeholder = 'host:port'; + + const labelInput = document.createElement('input'); + labelInput.className = 'add-src-input'; labelInput.type = 'text'; + labelInput.placeholder = 'label (optional)'; + + 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 = ''; + }); + addrInput.addEventListener('keydown', e => { if (e.key === 'Enter') addBtn.click(); }); + + const saveBtn = document.createElement('button'); + saveBtn.className = 'add-src-btn save-src-btn'; + saveBtn.textContent = 'Save list'; saveBtn.title = 'Save source list to file'; + saveBtn.addEventListener('click', saveSourcesWS); + + body.append(addrInput, labelInput, addBtn, saveBtn); + section.append(title, body); + + title.addEventListener('click', () => { + const open = section.classList.toggle('open'); + title.querySelector('.add-src-arrow').style.transform = open ? 'rotate(90deg)' : ''; + }); + + return section; +} + /* ════════════════════════════════════════════════════════════════ Utility ════════════════════════════════════════════════════════════════ */ @@ -1736,6 +1898,7 @@ function initSignalMenu() { ════════════════════════════════════════════════════════════════ */ buildLayoutMenu(); applyLayout('l1x1'); +buildSidebar(); // show "Add Source" section even before WS connection initSignalMenu(); document.getElementById('btn-csv-all').addEventListener('click', exportAllCSV); connectWS(); diff --git a/Client/WebUI/static/index.html b/Client/WebUI/static/index.html index 402a231..d006df3 100644 --- a/Client/WebUI/static/index.html +++ b/Client/WebUI/static/index.html @@ -1,163 +1,153 @@ - - - -MARTe2 UDP Streamer - - - - - - - - -
- - MARTe2 UDP Streamer -
- -
-
- A: —│ - B: —│ - ΔT: — -
- Window: - - - - - - - - -
- - -
-
- Signal - -
-
-
- Edge - -
-
-
- Threshold - -
-
-
- Window - -
-
-
- Pre - - 20% -
-
-
- Mode - -
-
-
- IDLE - - -
-
- - -
-