From 237084899469e53ac65e706692cf9373efb0ca2b Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 13 Aug 2026 10:28:56 +0200 Subject: [PATCH] added trigger --- Common/Client/go/wshub/trigger.go | 435 +++++++++++++++++++++++++ Common/Client/go/wshub/trigger_test.go | 194 +++++++++++ Common/Client/go/wshub/zoom_test.go | 61 ++++ 3 files changed, 690 insertions(+) create mode 100644 Common/Client/go/wshub/trigger.go create mode 100644 Common/Client/go/wshub/trigger_test.go create mode 100644 Common/Client/go/wshub/zoom_test.go diff --git a/Common/Client/go/wshub/trigger.go b/Common/Client/go/wshub/trigger.go new file mode 100644 index 0000000..ab74970 --- /dev/null +++ b/Common/Client/go/wshub/trigger.go @@ -0,0 +1,435 @@ +package wshub + +import ( + "encoding/binary" + "encoding/json" + "math" + "strconv" + "strings" + "sync" + "time" + + "github.com/gorilla/websocket" +) + +// Trigger FSM states, matching the C++ StreamHub TriggerEngine and the strings +// expected by the web SPA's "triggerState" handler. +const ( + trigIdle = "idle" + trigArmed = "armed" + trigCollecting = "collecting" + trigTriggered = "triggered" +) + +// captureMarginSec is the extra delay past the post-trigger window before the +// capture is extracted, so the rings have received the last samples. +const captureMarginSec = 0.15 + +// autoRearmDelaySec is the pause between a completed capture and the automatic +// rearm in "normal" mode. +const autoRearmDelaySec = 0.2 + +// trigConfig is the client-settable part of the trigger. +type trigConfig struct { + signalKey string // "src:sig" or "src:sig[i]" + edge string // "rising" | "falling" | "both" + threshold float64 + windowSec float64 + prePercent float64 + mode string // "normal" | "single" +} + +// triggerEngine implements the hub-side trigger FSM. Its methods are safe to +// call from the WebSocket read goroutines and from Hub.Run() concurrently. +type triggerEngine struct { + mu sync.Mutex + cfg trigConfig + + // Parsed form of cfg.signalKey, refreshed by SetConfig. + baseKey string // "src:sig" + elemIdx int // -1 when the key has no "[i]" suffix + + state string + stopped bool + + prevValue float64 + prevValid bool + lastT float64 + lastTOK bool + + trigTime float64 + firedPre float64 + firedPost float64 + firedValid bool + + rearmAt float64 // wall-clock seconds; 0 when no rearm is pending +} + +func newTriggerEngine() *triggerEngine { + return &triggerEngine{ + cfg: trigConfig{edge: "rising", windowSec: 1, prePercent: 20, mode: "normal"}, + elemIdx: -1, + state: trigIdle, + } +} + +// parseSignalKey splits "src:sig[3]" into ("src:sig", 3). A key without an +// element suffix yields an index of -1. +func parseSignalKey(key string) (string, int) { + if !strings.HasSuffix(key, "]") { + return key, -1 + } + open := strings.LastIndexByte(key, '[') + if open < 0 { + return key, -1 + } + idx, err := strconv.Atoi(key[open+1 : len(key)-1]) + if err != nil || idx < 0 { + return key, -1 + } + return key[:open], idx +} + +func (te *triggerEngine) SetConfig(cfg trigConfig) { + te.mu.Lock() + defer te.mu.Unlock() + // Clamp to the bounds the web UI offers. + if cfg.windowSec < 1e-4 { + cfg.windowSec = 1e-4 + } + if cfg.windowSec > 10 { + cfg.windowSec = 10 + } + if cfg.prePercent < 0 { + cfg.prePercent = 0 + } + if cfg.prePercent > 100 { + cfg.prePercent = 100 + } + te.cfg = cfg + te.baseKey, te.elemIdx = parseSignalKey(cfg.signalKey) + te.prevValid = false + te.prevValue = 0 +} + +func (te *triggerEngine) Config() trigConfig { + te.mu.Lock() + defer te.mu.Unlock() + return te.cfg +} + +func (te *triggerEngine) Arm() { + te.mu.Lock() + te.state = trigArmed + te.prevValid = false + te.prevValue = 0 + te.rearmAt = 0 + te.mu.Unlock() +} + +func (te *triggerEngine) Disarm() { + te.mu.Lock() + te.state = trigIdle + te.stopped = false + te.prevValid = false + te.prevValue = 0 + te.firedValid = false + te.rearmAt = 0 + te.mu.Unlock() +} + +func (te *triggerEngine) SetStopped(v bool) { + te.mu.Lock() + te.stopped = v + if v { + te.rearmAt = 0 + } + te.mu.Unlock() +} + +func (te *triggerEngine) Stopped() bool { + te.mu.Lock() + defer te.mu.Unlock() + return te.stopped +} + +func (te *triggerEngine) State() string { + te.mu.Lock() + defer te.mu.Unlock() + return te.state +} + +// Active reports whether a trigger signal is configured. The rings must stay +// populated from that moment on: a capture reaches back over the pre-trigger +// window, so waiting until the trigger arms would leave that window empty. +func (te *triggerEngine) Active() bool { + te.mu.Lock() + defer te.mu.Unlock() + return te.baseKey != "" +} + +// latchWindowLocked freezes the pre/post split at fire time so later config +// edits do not change how the capture is rendered. +func (te *triggerEngine) latchWindowLocked(t float64) { + te.state = trigCollecting + te.trigTime = t + te.firedPre = te.cfg.windowSec * te.cfg.prePercent / 100 + te.firedPost = te.cfg.windowSec - te.firedPre + te.firedValid = true + te.rearmAt = 0 +} + +// Force fires the trigger immediately at the most recent sample time (falling +// back to the current wall clock when no sample has been seen yet). +func (te *triggerEngine) Force() { + te.mu.Lock() + defer te.mu.Unlock() + if te.state == trigCollecting { + return + } + t := float64(time.Now().UnixNano()) / 1e9 + if te.lastTOK { + t = te.lastT + } + te.latchWindowLocked(t) +} + +// feed passes a batch of full-resolution samples for one signal to the FSM. +// key is the fully-prefixed "src:sig" name; nElem is the signal's element count +// so that an "[i]"-suffixed configuration can select a single column out of the +// flattened element-major batch. +func (te *triggerEngine) feed(key string, nElem int, t, v []float64) { + if len(t) == 0 || len(t) != len(v) { + return + } + te.mu.Lock() + defer te.mu.Unlock() + if key != te.baseKey { + return + } + te.lastT = t[len(t)-1] + te.lastTOK = true + if te.state != trigArmed { + return + } + step, start := 1, 0 + if te.elemIdx >= 0 && nElem > 1 { + if te.elemIdx >= nElem { + return + } + step, start = nElem, te.elemIdx + } + thr := te.cfg.threshold + for i := start; i < len(t); i += step { + if !te.prevValid { + te.prevValue = v[i] + te.prevValid = true + continue + } + up := te.prevValue < thr && v[i] >= thr + down := te.prevValue > thr && v[i] <= thr + te.prevValue = v[i] + fired := false + switch te.cfg.edge { + case "falling": + fired = down + case "both": + fired = up || down + default: + fired = up + } + if fired { + te.latchWindowLocked(t[i]) + return + } + } +} + +// dueCapture reports whether a collecting trigger's post-window has elapsed and +// returns the latched window. +func (te *triggerEngine) dueCapture(nowSec float64) (trigTime, pre, post float64, ok bool) { + te.mu.Lock() + defer te.mu.Unlock() + if te.state != trigCollecting || !te.firedValid { + return 0, 0, 0, false + } + if nowSec < te.trigTime+te.firedPost+captureMarginSec { + return 0, 0, 0, false + } + return te.trigTime, te.firedPre, te.firedPost, true +} + +// markTriggered completes a capture and schedules the automatic rearm when the +// engine runs in "normal" mode. +func (te *triggerEngine) markTriggered(nowSec float64) { + te.mu.Lock() + if te.state == trigCollecting { + te.state = trigTriggered + if te.cfg.mode != "single" && !te.stopped { + te.rearmAt = nowSec + autoRearmDelaySec + } + } + te.mu.Unlock() +} + +// dueRearm reports whether a pending automatic rearm has come due, consuming it. +func (te *triggerEngine) dueRearm(nowSec float64) bool { + te.mu.Lock() + defer te.mu.Unlock() + if te.state != trigTriggered || te.rearmAt == 0 || nowSec < te.rearmAt { + return false + } + te.rearmAt = 0 + return !te.stopped +} + +// stateMsg builds the JSON "triggerState" broadcast for the current FSM state. +func (te *triggerEngine) stateMsg() []byte { + te.mu.Lock() + m := map[string]any{ + "type": "triggerState", + "state": te.state, + "mode": te.cfg.mode, + "stopped": te.stopped, + } + if te.firedValid { + m["trigTime"] = te.trigTime + } + te.mu.Unlock() + msg, _ := json.Marshal(m) + return msg +} + +/* ─── Hub integration ─────────────────────────────────────────────────────── */ + +// broadcastTriggerState pushes the current FSM state to every client. +func (h *Hub) broadcastTriggerState() { + h.broadcast(h.trigger.stateMsg()) +} + +// handleTriggerCommand processes a trigger-related browser message. It returns +// false when the message type is not a trigger command. +func (h *Hub) handleTriggerCommand(t string, env map[string]interface{}) bool { + switch t { + case "setTrigger": + cfg := h.trigger.Config() + if s, ok := env["signal"].(string); ok { + cfg.signalKey = s + } + if s, ok := env["edge"].(string); ok { + cfg.edge = s + } + if s, ok := env["mode"].(string); ok { + cfg.mode = s + } + if f, ok := env["threshold"].(float64); ok { + cfg.threshold = f + } + if f, ok := env["windowSec"].(float64); ok { + cfg.windowSec = f + } + if f, ok := env["prePercent"].(float64); ok { + cfg.prePercent = f + } + h.trigger.SetConfig(cfg) + case "arm", "rearm": + h.trigger.Arm() + case "disarm": + h.trigger.Disarm() + case "trigStop": + stopped := !h.trigger.Stopped() + if b, ok := env["stopped"].(bool); ok { + stopped = b + } + h.trigger.SetStopped(stopped) + case "forceTrigger": + h.trigger.Force() + default: + return false + } + h.broadcastTriggerState() + return true +} + +// triggerTick services the trigger FSM; called from Hub.Run() on every push tick. +func (h *Hub) triggerTick() { + nowSec := float64(time.Now().UnixNano()) / 1e9 + prev := h.trigger.State() + + if trigTime, pre, post, ok := h.trigger.dueCapture(nowSec); ok { + if msg := h.buildTriggerCapture(trigTime, pre, post); msg != nil { + for c := range h.clients { + select { + case c.send <- wsMessage{websocket.BinaryMessage, msg}: + default: + } + } + } + h.trigger.markTriggered(nowSec) + } else if h.trigger.dueRearm(nowSec) { + h.trigger.Arm() + } + + if h.trigger.State() != prev { + h.broadcastTriggerState() + } +} + +// buildTriggerCapture extracts [trigTime-pre, trigTime+post] from every ring +// buffer and encodes the version-2 binary capture frame: +// +// [u8 2][f64 trigTime][f64 preSec][f64 postSec][u32 nSig] +// {[u16 keyLen][fullKey][u32 N][t f64×N][v f64×N]} +func (h *Hub) buildTriggerCapture(trigTime, pre, post float64) []byte { + t0, t1 := trigTime-pre, trigTime+post + + type sigSlice struct { + key string + t, v []float64 + } + h.ringsMu.RLock() + keys := make([]string, 0, len(h.rings)) + rings := make([]*sigRing, 0, len(h.rings)) + for k, rb := range h.rings { + keys = append(keys, k) + rings = append(rings, rb) + } + h.ringsMu.RUnlock() + + slices := make([]sigSlice, 0, len(keys)) + total := 1 + 8 + 8 + 8 + 4 + for i, k := range keys { + st, sv := rings[i].slice(t0, t1) + if len(st) == 0 { + continue + } + slices = append(slices, sigSlice{key: k, t: st, v: sv}) + total += 2 + len(k) + 4 + len(st)*16 + } + if len(slices) == 0 { + return nil + } + + buf := make([]byte, total) + buf[0] = 2 + off := 1 + binary.LittleEndian.PutUint64(buf[off:], math.Float64bits(trigTime)) + off += 8 + binary.LittleEndian.PutUint64(buf[off:], math.Float64bits(pre)) + off += 8 + binary.LittleEndian.PutUint64(buf[off:], math.Float64bits(post)) + off += 8 + binary.LittleEndian.PutUint32(buf[off:], uint32(len(slices))) + off += 4 + for _, s := range slices { + binary.LittleEndian.PutUint16(buf[off:], uint16(len(s.key))) + off += 2 + copy(buf[off:], s.key) + off += len(s.key) + binary.LittleEndian.PutUint32(buf[off:], uint32(len(s.t))) + off += 4 + off = writeFloat64s(buf, off, s.t) + off = writeFloat64s(buf, off, s.v) + } + return buf +} diff --git a/Common/Client/go/wshub/trigger_test.go b/Common/Client/go/wshub/trigger_test.go new file mode 100644 index 0000000..b6aa8c3 --- /dev/null +++ b/Common/Client/go/wshub/trigger_test.go @@ -0,0 +1,194 @@ +package wshub + +import "testing" + +func TestParseSignalKey(t *testing.T) { + cases := []struct { + in string + base string + idx int + }{ + {"src:sig", "src:sig", -1}, + {"src:sig[0]", "src:sig", 0}, + {"src:sig[3]", "src:sig", 3}, + {"src:sig[x]", "src:sig[x]", -1}, + {"src:sig]", "src:sig]", -1}, + } + for _, c := range cases { + base, idx := parseSignalKey(c.in) + if base != c.base || idx != c.idx { + t.Errorf("parseSignalKey(%q) = (%q,%d), want (%q,%d)", + c.in, base, idx, c.base, c.idx) + } + } +} + +func armed(key, edge string, thr float64) *triggerEngine { + te := newTriggerEngine() + te.SetConfig(trigConfig{signalKey: key, edge: edge, threshold: thr, + windowSec: 1, prePercent: 20, mode: "normal"}) + te.Arm() + return te +} + +func TestFeedRisingEdge(t *testing.T) { + te := armed("src:sig", "rising", 0.5) + te.feed("src:sig", 1, []float64{1, 2, 3, 4}, []float64{0, 0.2, 0.9, 1.0}) + if te.State() != trigCollecting { + t.Fatalf("state = %q, want collecting", te.State()) + } + // Fires at the sample that crossed, i.e. t=3. + trigTime, pre, post, ok := te.dueCapture(1e9) + if !ok || trigTime != 3 { + t.Fatalf("dueCapture = (%v,%v), want trigTime 3", trigTime, ok) + } + if pre != 0.2 || post != 0.8 { + t.Errorf("pre/post = %v/%v, want 0.2/0.8", pre, post) + } +} + +func TestFeedFallingEdgeIgnoresRising(t *testing.T) { + te := armed("src:sig", "falling", 0.5) + te.feed("src:sig", 1, []float64{1, 2, 3}, []float64{0, 0.9, 1.0}) + if te.State() != trigArmed { + t.Fatalf("state = %q, want armed (no falling edge)", te.State()) + } + te.feed("src:sig", 1, []float64{4, 5}, []float64{0.6, 0.1}) + if te.State() != trigCollecting { + t.Fatalf("state = %q, want collecting", te.State()) + } +} + +func TestFeedIgnoresOtherSignals(t *testing.T) { + te := armed("src:sig", "rising", 0.5) + te.feed("src:other", 1, []float64{1, 2}, []float64{0, 1}) + if te.State() != trigArmed { + t.Fatalf("state = %q, want armed", te.State()) + } +} + +func TestFeedArrayElementSelection(t *testing.T) { + // 2-element signal, element-major: [e0,e1, e0,e1, ...]. Only element 1 + // crosses the threshold. + te := armed("src:sig[1]", "rising", 0.5) + tt := []float64{1, 1, 2, 2} + vv := []float64{0, 0, 0, 1} + te.feed("src:sig", 2, tt, vv) + if te.State() != trigCollecting { + t.Fatalf("state = %q, want collecting", te.State()) + } + + // Element 0 never crosses, so a config on [0] must not fire. + te2 := armed("src:sig[0]", "rising", 0.5) + te2.feed("src:sig", 2, tt, vv) + if te2.State() != trigArmed { + t.Fatalf("state = %q, want armed", te2.State()) + } +} + +func TestForceUsesLastSampleTime(t *testing.T) { + te := newTriggerEngine() + te.SetConfig(trigConfig{signalKey: "src:sig", edge: "rising", threshold: 1e9, + windowSec: 2, prePercent: 50, mode: "single"}) + te.Arm() + te.feed("src:sig", 1, []float64{10, 11, 12}, []float64{0, 0, 0}) + if te.State() != trigArmed { + t.Fatalf("state = %q, want armed (threshold unreachable)", te.State()) + } + te.Force() + trigTime, pre, post, ok := te.dueCapture(1e9) + if !ok || trigTime != 12 || pre != 1 || post != 1 { + t.Fatalf("dueCapture = (%v,%v,%v,%v), want (12,1,1,true)", + trigTime, pre, post, ok) + } +} + +func TestForceFromIdle(t *testing.T) { + te := newTriggerEngine() + te.SetConfig(trigConfig{signalKey: "src:sig", edge: "rising", + windowSec: 1, prePercent: 20, mode: "normal"}) + te.Force() + if te.State() != trigCollecting { + t.Fatalf("state = %q, want collecting", te.State()) + } +} + +func TestCaptureMarginDelaysExtraction(t *testing.T) { + te := armed("src:sig", "rising", 0.5) + te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1}) // fires at t=1 + // post = 0.8 s; capture is due at 1 + 0.8 + 0.15. + if _, _, _, ok := te.dueCapture(1.9); ok { + t.Error("capture extracted before the margin elapsed") + } + if _, _, _, ok := te.dueCapture(1.96); !ok { + t.Error("capture not extracted after the margin elapsed") + } +} + +func TestAutoRearmNormalMode(t *testing.T) { + te := armed("src:sig", "rising", 0.5) + te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1}) + te.markTriggered(100) + if te.State() != trigTriggered { + t.Fatalf("state = %q, want triggered", te.State()) + } + if te.dueRearm(100.1) { + t.Error("rearmed before the delay elapsed") + } + if !te.dueRearm(100.3) { + t.Error("did not rearm after the delay elapsed") + } + if te.dueRearm(200) { + t.Error("rearm was not consumed") + } +} + +func TestNoAutoRearmInSingleMode(t *testing.T) { + te := newTriggerEngine() + te.SetConfig(trigConfig{signalKey: "src:sig", edge: "rising", threshold: 0.5, + windowSec: 1, prePercent: 20, mode: "single"}) + te.Arm() + te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1}) + te.markTriggered(100) + if te.dueRearm(200) { + t.Error("single mode must not auto-rearm") + } +} + +func TestStoppedSuppressesRearm(t *testing.T) { + te := armed("src:sig", "rising", 0.5) + te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1}) + te.SetStopped(true) + te.markTriggered(100) + if te.dueRearm(200) { + t.Error("stopped engine must not rearm") + } +} + +func TestSetConfigClamps(t *testing.T) { + te := newTriggerEngine() + te.SetConfig(trigConfig{signalKey: "s:x", windowSec: 100, prePercent: 500}) + if cfg := te.Config(); cfg.windowSec != 10 || cfg.prePercent != 100 { + t.Errorf("upper clamp = %v/%v, want 10/100", cfg.windowSec, cfg.prePercent) + } + te.SetConfig(trigConfig{signalKey: "s:x", windowSec: 0, prePercent: -5}) + if cfg := te.Config(); cfg.windowSec != 1e-4 || cfg.prePercent != 0 { + t.Errorf("lower clamp = %v/%v, want 1e-4/0", cfg.windowSec, cfg.prePercent) + } +} + +func TestActiveTracksConfiguredSignal(t *testing.T) { + te := newTriggerEngine() + if te.Active() { + t.Error("a fresh engine must not be active") + } + te.SetConfig(trigConfig{signalKey: "src:sig", windowSec: 1}) + if !te.Active() { + t.Error("engine must be active once a signal is configured") + } + // Rings must keep filling after a capture completes, not just while armed. + te.Disarm() + if !te.Active() { + t.Error("engine must stay active after disarm while a signal is set") + } +} diff --git a/Common/Client/go/wshub/zoom_test.go b/Common/Client/go/wshub/zoom_test.go new file mode 100644 index 0000000..87b54bc --- /dev/null +++ b/Common/Client/go/wshub/zoom_test.go @@ -0,0 +1,61 @@ +package wshub + +import "testing" + +func TestZoomPoints(t *testing.T) { + cases := []struct { + n int + present bool + want int + }{ + {0, false, 2400}, // absent → default budget + {2400, true, 2400}, // explicit budget honoured + {0, true, 1 << 30}, // 0 → every sample in range + {-1, true, 1 << 30}, // negative → every sample in range + {5, true, 2400}, // implausibly small → default budget + } + for _, c := range cases { + if got := zoomPoints(c.n, c.present); got != c.want { + t.Errorf("zoomPoints(%d,%v) = %d, want %d", c.n, c.present, got, c.want) + } + } +} + +func TestZoomSliceReturnsFullResolution(t *testing.T) { + h := NewHub() + rb := newSigRing(1000) + ts := make([]float64, 500) + vs := make([]float64, 500) + for i := range ts { + ts[i] = float64(i) * 0.001 // 1 kHz + vs[i] = float64(i) + } + rb.write(ts, vs) + h.rings["s1:sig"] = rb + + // A budget larger than the range must return every sample untouched. + res := h.zoomSlice(0.100, 0.199, []string{"s1:sig"}, 1<<30) + sd, ok := res["s1:sig"] + if !ok { + t.Fatal("signal missing from zoom result") + } + if len(sd.T) != 100 { + t.Fatalf("got %d points, want 100", len(sd.T)) + } + if sd.V[0] != 100 || sd.V[99] != 199 { + t.Errorf("value range = %v..%v, want 100..199", sd.V[0], sd.V[99]) + } + + // A small budget decimates but keeps the endpoints. + dec := h.zoomSlice(0.100, 0.199, []string{"s1:sig"}, 20) + if len(dec["s1:sig"].T) != 20 { + t.Errorf("decimated to %d points, want 20", len(dec["s1:sig"].T)) + } +} + +func TestZoomSliceUnknownSignal(t *testing.T) { + h := NewHub() + if res := h.zoomSlice(0, 1, []string{"nope", ""}, 100); len(res) != 0 { + t.Errorf("got %d entries, want 0", len(res)) + } +}