Martino Ferrari
702 lines
19 KiB
Go
702 lines
19 KiB
Go
package main
|
||
|
||
import (
|
||
"encoding/json"
|
||
"log"
|
||
"math"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/gorilla/websocket"
|
||
)
|
||
|
||
// ─── WebSocket client ─────────────────────────────────────────────────────────
|
||
|
||
type wsClient struct {
|
||
hub *Hub
|
||
conn *websocket.Conn
|
||
send chan []byte
|
||
}
|
||
|
||
func (c *wsClient) writePump() {
|
||
pingTicker := time.NewTicker(30 * time.Second)
|
||
defer func() {
|
||
pingTicker.Stop()
|
||
c.conn.Close()
|
||
}()
|
||
for {
|
||
select {
|
||
case msg, ok := <-c.send:
|
||
if !ok {
|
||
c.conn.WriteMessage(websocket.CloseMessage, []byte{})
|
||
return
|
||
}
|
||
if err := c.conn.WriteMessage(websocket.TextMessage, msg); err != nil {
|
||
return
|
||
}
|
||
case <-pingTicker.C:
|
||
if err := c.conn.WriteControl(websocket.PingMessage, []byte{},
|
||
time.Now().Add(10*time.Second)); err != nil {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (c *wsClient) readPump() {
|
||
defer func() {
|
||
c.hub.unregister <- c
|
||
c.conn.Close()
|
||
}()
|
||
c.conn.SetReadLimit(64 * 1024)
|
||
c.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
|
||
c.conn.SetPongHandler(func(string) error {
|
||
c.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
|
||
return nil
|
||
})
|
||
for {
|
||
_, msg, err := c.conn.ReadMessage()
|
||
if err != nil {
|
||
break
|
||
}
|
||
var env map[string]interface{}
|
||
if json.Unmarshal(msg, &env) == nil {
|
||
if t, ok := env["type"].(string); ok {
|
||
switch t {
|
||
case "ping":
|
||
resp, _ := json.Marshal(map[string]string{"type": "pong"})
|
||
select {
|
||
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:
|
||
}
|
||
}
|
||
}
|
||
}
|
||
c.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
|
||
}
|
||
}
|
||
|
||
// ─── Hub ─────────────────────────────────────────────────────────────────────
|
||
|
||
var upgrader = websocket.Upgrader{
|
||
ReadBufferSize: 4096,
|
||
WriteBufferSize: 64 * 1024,
|
||
CheckOrigin: func(r *http.Request) bool { return true },
|
||
}
|
||
|
||
// 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, except
|
||
// ringsMu/rings which are also read by HTTP handler goroutines.
|
||
type Hub struct {
|
||
clients map[*wsClient]bool
|
||
register chan *wsClient
|
||
unregister chan *wsClient
|
||
broadcastCh chan []byte
|
||
dataCh chan taggedSample
|
||
commandCh chan hubCmd
|
||
|
||
sm *SourceManager // set after construction; used for WS-initiated source changes
|
||
|
||
// Ring buffers for hi-res zoom data.
|
||
// ringsMu protects the map structure; each sigRing has its own RWMutex for data.
|
||
ringsMu sync.RWMutex
|
||
rings map[string]*sigRing // "sourceId:signalKey" → ring
|
||
}
|
||
|
||
// 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 taggedSample, 256),
|
||
commandCh: make(chan hubCmd, 64),
|
||
rings: make(map[string]*sigRing),
|
||
}
|
||
}
|
||
|
||
// getRing returns the ring buffer for a fully-prefixed signal key, or nil.
|
||
func (h *Hub) getRing(key string) *sigRing {
|
||
h.ringsMu.RLock()
|
||
rb := h.rings[key]
|
||
h.ringsMu.RUnlock()
|
||
return rb
|
||
}
|
||
|
||
// HandleZoom serves GET /api/zoom?t0=...&t1=...&n=...&signals=key1,key2
|
||
// It reads from the ring buffers (safe for concurrent access) and returns
|
||
// LTTB-decimated signal data for the requested time range.
|
||
func (h *Hub) HandleZoom(w http.ResponseWriter, r *http.Request) {
|
||
q := r.URL.Query()
|
||
t0, err0 := strconv.ParseFloat(q.Get("t0"), 64)
|
||
t1, err1 := strconv.ParseFloat(q.Get("t1"), 64)
|
||
if err0 != nil || err1 != nil || t1 <= t0 {
|
||
http.Error(w, "invalid t0/t1", http.StatusBadRequest)
|
||
return
|
||
}
|
||
// n=0 (or explicit "0") means no LTTB decimation — return all ring data in range.
|
||
// n omitted / invalid → default 2400 (display quality).
|
||
var n int
|
||
if nStr := q.Get("n"); nStr == "" {
|
||
n = 2400
|
||
} else {
|
||
n, _ = strconv.Atoi(nStr)
|
||
if n <= 0 {
|
||
n = 1 << 30 // no decimation
|
||
} else if n < 10 {
|
||
n = 2400
|
||
}
|
||
}
|
||
|
||
keys := strings.Split(q.Get("signals"), ",")
|
||
|
||
// Collect ring references under a brief RLock.
|
||
h.ringsMu.RLock()
|
||
refs := make(map[string]*sigRing, len(keys))
|
||
for _, k := range keys {
|
||
k = strings.TrimSpace(k)
|
||
if k == "" {
|
||
continue
|
||
}
|
||
if rb, ok := h.rings[k]; ok {
|
||
refs[k] = rb
|
||
}
|
||
}
|
||
h.ringsMu.RUnlock()
|
||
|
||
result := make(map[string]sigData, len(refs))
|
||
for k, rb := range refs {
|
||
rt, rv := rb.slice(t0, t1)
|
||
if len(rt) == 0 {
|
||
continue
|
||
}
|
||
dt, dv := lttbDecimate(rt, rv, n)
|
||
result[k] = sigData{T: dt, V: dv}
|
||
}
|
||
|
||
w.Header().Set("Content-Type", "application/json")
|
||
if err := json.NewEncoder(w).Encode(map[string]any{
|
||
"type": "zoom",
|
||
"signals": result,
|
||
}); err != nil {
|
||
log.Printf("hub: zoom encode: %v", err)
|
||
}
|
||
}
|
||
|
||
// AddSource notifies the Hub that a new source has been registered.
|
||
func (h *Hub) AddSource(id, label, addr string) {
|
||
select {
|
||
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:
|
||
}
|
||
}
|
||
|
||
// broadcast enqueues a message for delivery to all WebSocket clients.
|
||
func (h *Hub) broadcast(msg []byte) {
|
||
select {
|
||
case h.broadcastCh <- msg:
|
||
default:
|
||
}
|
||
}
|
||
|
||
// HandleWebSocket upgrades an HTTP request to a WebSocket connection.
|
||
func (h *Hub) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
|
||
conn, err := upgrader.Upgrade(w, r, nil)
|
||
if err != nil {
|
||
log.Printf("ws upgrade: %v", err)
|
||
return
|
||
}
|
||
c := &wsClient{hub: h, conn: conn, send: make(chan []byte, 64)}
|
||
h.register <- c
|
||
go c.writePump()
|
||
go c.readPump()
|
||
}
|
||
|
||
// 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() {
|
||
ticker := time.NewTicker(time.Second / 30)
|
||
defer ticker.Stop()
|
||
|
||
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.clients[c] = true
|
||
// 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:
|
||
if _, ok := h.clients[c]; ok {
|
||
delete(h.clients, c)
|
||
close(c.send)
|
||
}
|
||
|
||
case msg := <-h.broadcastCh:
|
||
for c := range h.clients {
|
||
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)
|
||
pfxDel := cmd.sourceID + ":"
|
||
h.ringsMu.Lock()
|
||
for k := range h.rings {
|
||
if strings.HasPrefix(k, pfxDel) {
|
||
delete(h.rings, k)
|
||
}
|
||
}
|
||
h.ringsMu.Unlock()
|
||
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]any{
|
||
"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)
|
||
// Rebuild ring buffers for this source.
|
||
pfxUpd := cmd.sourceID + ":"
|
||
h.ringsMu.Lock()
|
||
for k := range h.rings {
|
||
if strings.HasPrefix(k, pfxUpd) {
|
||
delete(h.rings, k)
|
||
}
|
||
}
|
||
for _, sig := range cmd.sigs {
|
||
ne := sig.NumElements()
|
||
isTemporal := ne > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample)
|
||
if isTemporal {
|
||
h.rings[pfxUpd+sig.Name] = newSigRing(ringCapTemporal)
|
||
} else if ne == 1 {
|
||
h.rings[pfxUpd+sig.Name] = newSigRing(ringCapScalar)
|
||
} else {
|
||
for i := 0; i < ne; i++ {
|
||
h.rings[pfxUpd+arrayKey(sig.Name, i)] = newSigRing(ringCapScalar)
|
||
}
|
||
}
|
||
}
|
||
h.ringsMu.Unlock()
|
||
|
||
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)
|
||
}
|
||
}
|
||
}
|
||
|
||
case ts := <-h.dataCh:
|
||
pending[ts.sourceID] = append(pending[ts.sourceID], ts.sample)
|
||
|
||
case <-ticker.C:
|
||
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)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// ─── Data serialisation ───────────────────────────────────────────────────────
|
||
|
||
// maxPushPoints is the LTTB target for data pushed over WebSocket per 30 Hz tick.
|
||
// At 2 k pts/tick × 30 Hz = 60 k pts/s; the 600 k frontend buffer covers ~10 s.
|
||
// Kept deliberately low so the rolling window shows plenty of history even for
|
||
// multi-MHz signals — zoom resolution comes from the ring buffer instead.
|
||
const maxPushPoints = 2_000
|
||
|
||
// maxRingPoints is the LTTB target written into the ring buffer per tick.
|
||
// At 5 Msps / 5 kHz packet rate ≈ 167 k raw samples/tick → LTTB to 20 k →
|
||
// min Δt ≈ 33 ms / 20 k ≈ 1.65 µs, sufficient for sub-10 µs zoom resolution.
|
||
const maxRingPoints = 20_000
|
||
|
||
// ringCapTemporal is the ring buffer capacity for temporal-array signals.
|
||
// At 20 k pts/tick × 30 Hz = 600 k pts/s → 6 M cap gives ~10 s of hi-res
|
||
// history — the same temporal coverage as the frontend push buffer.
|
||
const ringCapTemporal = 6_000_000
|
||
|
||
// ringCapScalar is the ring buffer capacity for scalar / spatial-array signals.
|
||
// At ≤10 kHz → ~333 pts/tick × 30 Hz ≈ 10 k pts/s → ~10 s of history.
|
||
const ringCapScalar = 100_000
|
||
|
||
// 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"`
|
||
V []float64 `json:"v"`
|
||
}
|
||
|
||
// dataMsg is the JSON envelope sent to WebSocket clients.
|
||
type dataMsg struct {
|
||
Type string `json:"type"`
|
||
SourceID string `json:"sourceId"`
|
||
Signals map[string]sigData `json:"signals"`
|
||
}
|
||
|
||
// 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 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 {
|
||
n := sig.NumElements()
|
||
isTemporal := n > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample)
|
||
|
||
switch {
|
||
case isTemporal:
|
||
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
|
||
}
|
||
allT := make([]float64, 0, len(batch)*n)
|
||
allV := make([]float64, 0, len(batch)*n)
|
||
|
||
for _, s := range batch {
|
||
vals, ok := s.Values[sig.Name]
|
||
if !ok || len(vals) < n {
|
||
continue
|
||
}
|
||
var anchorTime float64
|
||
anchorIsFirstSample := sig.TimeMode == TimeModeFirstSample
|
||
if hasTimeSig {
|
||
tVals, tOk := s.Values[timeSigName]
|
||
if tOk && len(tVals) >= 1 {
|
||
timerS := tVals[0] * 1e-6
|
||
wallT := float64(s.WallTime.UnixNano()) / 1e9
|
||
if _, exists := src.timeSigCalib[timeSigName]; !exists {
|
||
src.timeSigCalib[timeSigName] = wallT - timerS
|
||
}
|
||
anchorTime = src.timeSigCalib[timeSigName] + timerS
|
||
} else {
|
||
anchorTime = float64(s.WallTime.UnixNano()) / 1e9
|
||
anchorIsFirstSample = false
|
||
}
|
||
} else {
|
||
anchorTime = float64(s.WallTime.UnixNano()) / 1e9
|
||
anchorIsFirstSample = false
|
||
}
|
||
for k := 0; k < n; k++ {
|
||
var t float64
|
||
if anchorIsFirstSample {
|
||
t = anchorTime + float64(k)*dt
|
||
} else {
|
||
t = anchorTime - float64(n-1-k)*dt
|
||
}
|
||
allT = append(allT, t)
|
||
allV = append(allV, vals[k])
|
||
}
|
||
}
|
||
|
||
// Write hi-res LTTB data to ring for on-demand zoom queries.
|
||
ringT, ringV := lttbDecimate(allT, allV, maxRingPoints)
|
||
if rb := h.getRing(pfx + sig.Name); rb != nil {
|
||
rb.write(ringT, ringV)
|
||
}
|
||
// Decimate further for WebSocket push (rolling window).
|
||
decimT, decimV := lttbDecimate(allT, allV, maxPushPoints)
|
||
out[pfx+sig.Name] = sigData{T: decimT, V: decimV}
|
||
|
||
case n == 1:
|
||
ts := make([]float64, 0, len(batch))
|
||
vs := make([]float64, 0, len(batch))
|
||
for _, s := range batch {
|
||
vals, ok := s.Values[sig.Name]
|
||
if !ok || len(vals) < 1 {
|
||
continue
|
||
}
|
||
ts = append(ts, float64(s.WallTime.UnixNano())/1e9)
|
||
vs = append(vs, vals[0])
|
||
}
|
||
if rb := h.getRing(pfx + sig.Name); rb != nil {
|
||
rb.write(ts, vs)
|
||
}
|
||
out[pfx+sig.Name] = sigData{T: ts, V: vs}
|
||
|
||
default:
|
||
// Spatial / PacketTime array: one stream per element.
|
||
for i := 0; i < n; i++ {
|
||
key := pfx + arrayKey(sig.Name, i)
|
||
ts := make([]float64, 0, len(batch))
|
||
vs := make([]float64, 0, len(batch))
|
||
for _, s := range batch {
|
||
vals, ok := s.Values[sig.Name]
|
||
if !ok || len(vals) <= i {
|
||
continue
|
||
}
|
||
ts = append(ts, float64(s.WallTime.UnixNano())/1e9)
|
||
vs = append(vs, vals[i])
|
||
}
|
||
if rb := h.getRing(key); rb != nil {
|
||
rb.write(ts, vs)
|
||
}
|
||
out[key] = sigData{T: ts, V: vs}
|
||
}
|
||
}
|
||
}
|
||
|
||
result, err := json.Marshal(dataMsg{Type: "data", SourceID: src.id, Signals: out})
|
||
if err != nil {
|
||
log.Printf("hub: marshal data: %v", err)
|
||
return nil
|
||
}
|
||
return result
|
||
}
|
||
|
||
// arrayKey returns the buffer key for element i of an array signal.
|
||
func arrayKey(name string, i int) string {
|
||
return name + "[" + itoa(i) + "]"
|
||
}
|
||
|
||
func itoa(n int) string {
|
||
if n == 0 {
|
||
return "0"
|
||
}
|
||
buf := [20]byte{}
|
||
pos := len(buf)
|
||
for n > 0 {
|
||
pos--
|
||
buf[pos] = byte('0' + n%10)
|
||
n /= 10
|
||
}
|
||
return string(buf[pos:])
|
||
}
|