added trigger
This commit is contained in:
Martino Ferrari
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user