Martino Ferrari
574 lines
15 KiB
Go
574 lines
15 KiB
Go
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"log"
|
|
"math"
|
|
"net/http"
|
|
"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.
|
|
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
|
|
}
|
|
|
|
// 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),
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
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)
|
|
}
|
|
}
|
|
}
|
|
|
|
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 ───────────────────────────────────────────────────────
|
|
|
|
// 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"`
|
|
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])
|
|
}
|
|
}
|
|
|
|
decimT, decimV := lttbDecimate(allT, allV, maxTemporalPoints)
|
|
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])
|
|
}
|
|
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])
|
|
}
|
|
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:])
|
|
}
|