martino/MARTe_IO_Components
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improved uo and added timerarraygam for testing

Browse Source
This commit is contained in:
Martino Ferrari
2026-05-20 00:15:38 +02:00
parent 620542a722
commit 62545503c4
13 changed files with 960 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Client/WebUI/hub.go
+97 -5
View File
@@ -156,6 +156,9 @@ type Hub struct {
// ringsMu protects the map structure; each sigRing has its own RWMutex for data.
ringsMu sync.RWMutex
rings map[string]*sigRing // "sourceId:signalKey" → ring
statsMu sync.RWMutex
statsMap map[string]*SourceStat
}
// NewHub creates an initialised Hub.
@@ -168,6 +171,7 @@ func NewHub() *Hub {
dataCh: make(chan taggedSample, 256),
commandCh: make(chan hubCmd, 64),
rings: make(map[string]*sigRing),
statsMap: make(map[string]*SourceStat),
}
}
@@ -321,6 +325,9 @@ func (h *Hub) Run() {
ticker := time.NewTicker(time.Second / 30)
defer ticker.Stop()
statsTicker := time.NewTicker(time.Second)
defer statsTicker.Stop()
sourcesMap := make(map[string]*sourceHubState)
var sourcesMsg []byte
@@ -367,6 +374,9 @@ func (h *Hub) Run() {
connState: "connecting",
timeSigCalib: make(map[string]float64),
}
h.statsMu.Lock()
h.statsMap[cmd.sourceID] = &SourceStat{}
h.statsMu.Unlock()
rebuildSources()
case "removeSource":
@@ -380,6 +390,9 @@ func (h *Hub) Run() {
}
}
h.ringsMu.Unlock()
h.statsMu.Lock()
delete(h.statsMap, cmd.sourceID)
h.statsMu.Unlock()
rebuildSources()
case "setSourceState":
@@ -416,7 +429,7 @@ func (h *Hub) Run() {
}
for _, sig := range cmd.sigs {
ne := sig.NumElements()
isTemporal := ne > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample)
isTemporal := ne > 1 && sig.TimeMode != TimeModePacket
if isTemporal {
h.rings[pfxUpd+sig.Name] = newSigRing(ringCapTemporal)
} else if ne == 1 {
@@ -466,6 +479,18 @@ func (h *Hub) Run() {
h.broadcast(msg)
}
}
case <-statsTicker.C:
h.statsMu.RLock()
snap := make(map[string]StatInfo, len(h.statsMap))
for id, st := range h.statsMap {
snap[id] = st.Snapshot()
}
h.statsMu.RUnlock()
if len(snap) > 0 {
msg, _ := json.Marshal(map[string]any{"type": "stats", "sources": snap})
h.broadcast(msg)
}
}
}
}
@@ -574,14 +599,19 @@ func (h *Hub) buildDataMessageForSource(src *sourceHubState, batch []DataSample)
for _, sig := range sigs {
n := sig.NumElements()
isTemporal := n > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample)
switch {
case isTemporal:
case n > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample):
hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs)
var timeSigName string
// uint64 signals carry nanoseconds (HRT); everything else is assumed microseconds.
timerToSec := 1e-6
if hasTimeSig {
timeSigName = sigs[sig.TimeSignalIdx].Name
ts := sigs[sig.TimeSignalIdx]
timeSigName = ts.Name
if ts.TypeCode == 6 { // uint64 → nanoseconds
timerToSec = 1e-9
}
}
dt := 0.0
if sig.SamplingRate > 0 {
@@ -600,7 +630,7 @@ func (h *Hub) buildDataMessageForSource(src *sourceHubState, batch []DataSample)
if hasTimeSig {
tVals, tOk := s.Values[timeSigName]
if tOk && len(tVals) >= 1 {
timerS := tVals[0] * 1e-6
timerS := tVals[0] * timerToSec
wallT := float64(s.WallTime.UnixNano()) / 1e9
if _, exists := src.timeSigCalib[timeSigName]; !exists {
src.timeSigCalib[timeSigName] = wallT - timerS
@@ -635,6 +665,58 @@ func (h *Hub) buildDataMessageForSource(src *sourceHubState, batch []DataSample)
decimT, decimV := lttbDecimate(allT, allV, maxPushPoints)
out[pfx+sig.Name] = sigData{T: decimT, V: decimV}
case n > 1 && sig.TimeMode == TimeModeFullArray:
// The time signal has the same N elements as the data signal.
// Each element pair (timeSig[k], dataSig[k]) is one (t, v) sample.
hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs)
var timeSigName string
timerToSec := 1e-6
if hasTimeSig {
ts := sigs[sig.TimeSignalIdx]
timeSigName = ts.Name
if ts.TypeCode == 6 { // uint64 → nanoseconds
timerToSec = 1e-9
}
}
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
}
if hasTimeSig {
tVals, tOk := s.Values[timeSigName]
if tOk && len(tVals) >= n {
// Calibrate once: map timer ticks to wall-clock seconds.
if _, exists := src.timeSigCalib[timeSigName]; !exists {
wallT := float64(s.WallTime.UnixNano()) / 1e9
src.timeSigCalib[timeSigName] = wallT - tVals[0]*timerToSec
}
calib := src.timeSigCalib[timeSigName]
for k := 0; k < n; k++ {
allT = append(allT, calib+tVals[k]*timerToSec)
allV = append(allV, vals[k])
}
continue
}
}
// Fallback: stamp all elements with packet arrival time.
wallT := float64(s.WallTime.UnixNano()) / 1e9
for k := 0; k < n; k++ {
allT = append(allT, wallT)
allV = append(allV, vals[k])
}
}
ringT, ringV := lttbDecimate(allT, allV, maxRingPoints)
if rb := h.getRing(pfx + sig.Name); rb != nil {
rb.write(ringT, ringV)
}
decimT, decimV := lttbDecimate(allT, allV, maxPushPoints)
out[pfx+sig.Name] = sigData{T: decimT, V: decimV}
case n == 1:
ts := make([]float64, 0, len(batch))
vs := make([]float64, 0, len(batch))
@@ -681,6 +763,16 @@ func (h *Hub) buildDataMessageForSource(src *sourceHubState, batch []DataSample)
return result
}
// RecordDataFragment is called by UDPClient for every incoming DATA datagram.
func (h *Hub) RecordDataFragment(sourceID string, counter uint32, nBytes int, arrivalNs int64, complete bool) {
h.statsMu.RLock()
st := h.statsMap[sourceID]
h.statsMu.RUnlock()
if st != nil {
st.RecordFragment(counter, nBytes, arrivalNs, complete)
}
}
// arrayKey returns the buffer key for element i of an array signal.
func arrayKey(name string, i int) string {
return name + "[" + itoa(i) + "]"