multi source
This commit is contained in:
Martino Ferrari
+302
-176
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user