MARTe-Integrated-Components/Docs/StreamHub-API.md

StreamHub WebSocket API

StreamHub exposes a single WebSocket server (default port 8090, any URL path). Text frames carry JSON commands/events; binary frames carry data pushes and trigger captures. The Go reference hub (Client/udpstreamer) implements the identical protocol; both the browser SPA and the ImGui client work against either.

For the upstream UDPS wire format (source → hub) see Protocol.md.

All numbers in binary frames are little-endian. All timestamps are Unix wall-clock seconds (float64) — see Time base in StreamHub-Developer.md.

---

1. Commands (client → hub, JSON text frames)

Every command is a JSON object with a type field.

ping

{"type":"ping"}

Reply (unicast): {"type":"pong"}.

addSource

{"type":"addSource","label":"PSU","addr":"192.168.0.10:44500",
 "multicastGroup":"239.0.0.1","dataPort":44503}

• addr — host:port of the UDPStreamer control port.
• multicastGroup / dataPort — optional; if present the hub joins the multicast group for DATA and keeps a TCP control connection for CONNECT/CONFIG.
• The hub assigns ids s1, s2, … and broadcasts an updated sources event.

removeSource

{"type":"removeSource","id":"s1"}

saveSources

{"type":"saveSources"}

Writes the hub's SourcesFile: the current dynamically-added source list and the calibration table, as one flat JSON array (see §4). The hub replies with configSaved. Despite the name, this command persists the whole config, not just the sources.

setCalibration

{"type":"setCalibration","source":"wave","signal":"Adc","scale":0.00030518,"offset":-1.25,"unit":"V"}

Records an affine calibration value = raw × scale + offset for one signal, keyed by the source's label (not its runtime id) and the base signal name — one entry covers every element of an array signal.

Field	Type	Default	Validation
source	string	—	non-empty after trimming
signal	string	—	non-empty after trimming; any trailing [i] is stripped
scale	number	1	finite and non-zero
offset	number	0	finite
unit	string	""	trimmed, truncated to 16 UTF-8 bytes (a multi-byte character such as °C consumes more than one byte; truncation never splits a character); empty = use the streamer's own unit

Calibration is metadata only: the hub stores and redistributes it but never applies it. Ring buffers, recorded history, the zoom reply, both binary frames and the trigger comparator all stay in raw units — a client that ignores calibration behaves exactly as before.

An entry that reduces to the identity (scale = 1, offset = 0, unit = "") is deleted rather than stored, so a reset leaves no residue in the config file.

On acceptance the hub broadcasts calibration to every client. A rejected entry produces no broadcast, so the offending client reverts to the last value it was told.

reloadConfig

{"type":"reloadConfig"}

Re-reads SourcesFile and then:

• replaces the calibration table wholesale with the file's contents;
• adds any source in the file that is not already active;
• never removes, restarts or reconnects a live source.

The asymmetry is deliberate: calibration is cheap to reapply, whereas a source is a live UDP session that must not be interrupted. An unsaved source the user added keeps streaming.

The hub replies with configReloaded, followed on success by a calibration broadcast. Whether a sources broadcast follows depends on the hub implementation: the Go hub emits sources only when the file adds at least one new source (each sm.Add() call triggers it individually), while the C++ hub always emits sources unconditionally after a successful reload. Clients must therefore tolerate an unsolicited sources frame after any reload.

getSources / getConfig / getStats

{"type":"getSources"}
{"type":"getConfig","sourceId":"s1"}
{"type":"getStats"}

Force a broadcast of the corresponding event.

setTrigger

{"type":"setTrigger","signal":"s1:Sine","edge":"rising","threshold":0.0,
 "windowSec":0.1,"prePercent":20,"mode":"normal"}

• signal — full key src:sig, or src:sig[i] to trigger on element i of a multi-element PACKET signal.
• edge — "rising", "falling" or "both".
• windowSec — total capture window. preSec = windowSec * prePercent / 100, postSec = windowSec − preSec. Clamped to 1e-4 … 600 s by the Go hub and to 1e-4 … 60 s by the C++ one: the Go rings store min/max pairs once a window outgrows their memory budget, so a long window costs resolution, while the C++ rings are fixed-capacity and would return the window truncated instead.
• mode — "normal" (auto-rearm ~200 ms after capture) or "single" (stays TRIGGERED until rearm).

arm / disarm / rearm / trigStop

{"type":"arm"}
{"type":"disarm"}
{"type":"rearm"}
{"type":"trigStop","stopped":true}

• arm — IDLE → ARMED.
• disarm — any state → IDLE.
• rearm — TRIGGERED → ARMED (single mode).
• trigStop — sets the stopped flag suppressing auto-rearm in normal mode; omit stopped to toggle.

Every transition is broadcast as a triggerState event.

zoom

{"type":"zoom","reqId":17,"t0":1765370000.123,"t1":1765370000.223,
 "n":2400,"signals":"s1:Sine,s2:Wave[0]"}

• reqId — echoed in the reply; lets the client match request/response.
• t0/t1 — Unix seconds window.
• n — target points per signal. Absent or <10 → 2400; n ≤ 0 → no decimation (raw ring contents).
• signals — comma-separated full keys; absent → all signals of all sources.
• The reply is unicast to the requesting client only.

historyZoom

{"type":"historyZoom","reqId":42,"t0":1765360000.0,"t1":1765370000.0,
 "n":2400,"signals":"s1:Sine,s2:Wave"}

• Reads from disk-backed history instead of the in-memory ring buffer. Identical semantics to zoom but queries the .shist files written by HistoryWriter.
• reqId, t0/t1, n, signals — same meaning as zoom.
• If history is not enabled, the reply contains "error":"history not enabled".
• Reply is unicast (same shape as zoom reply, but "type":"historyZoom").

historyInfo

{"type":"historyInfo"}

Request the hub to send a historyInfo event (unicast). Also sent automatically on client connect.

setHistoryBudget (Go hub only)

{"type":"setHistoryBudget","maxMPtsPerSignal":16.0}

Sets the per-signal archive budget in millions of stored points, the runtime equivalent of -history-max-mpts. 0 restores the 16 MPts default; the hub clamps to its own ceiling. Every archive file is re-created at the new size — the archived samples are lost, because a file's capacity and min/max bucket width are fixed at creation. Broadcasts historyInfo rather than answering the requester alone: every client's view of what history exists has been invalidated.

setWindow (Go hub only)

{"type":"setWindow","seconds":60}

Reports how far back this client is plotting. The hub sizes its in-memory buffers for the widest window any connected client has reported (10 s if none has), bucketing each ring as min/max pairs when the window is too long to hold verbatim — see In-memory buffer policy in StreamHub-Developer.md. While a trigger is armed the trigger's own window wins. No reply; send it on connect and whenever the timescale changes. A window the hub is not told about is a window whose start may already have rolled out of the ring, leaving a zoom over it nothing to answer with.

setMaxPoints

{"type":"setMaxPoints","maxPoints":50000}

Resizes all ring buffers (applied safely inside the push loop; push cursors are reset). Broadcasts maxPointsUpdated.

---

2. Events (hub → client, JSON text frames)

sources

{"type":"sources","sources":[
  {"id":"s1","label":"PSU","addr":"192.168.0.10:44500","state":"streaming"}]}

Sent on client connect, after add/remove/getSources, and when a source first delivers its CONFIG.

config

{"type":"config","sourceId":"s1","publishMode":0,"signals":[
  {"name":"Sine","typeCode":10,"quantType":0,"numDimensions":0,
   "numRows":1,"numCols":1,"rangeMin":-1.0,"rangeMax":1.0,
   "timeMode":0,"samplingRate":5000000.0,"timeSignalIdx":-1,"unit":"V"}]}

Field semantics follow the UDPS signal descriptor (Protocol.md). numElements = numRows × numCols.

stats

Sent at StatsRate Hz (default 1 Hz):

{"type":"stats","sources":{"s1":{
  "state":"streaming","totalReceived":1234,"totalLost":0,
  "rateHz":100.1,"rateStdHz":0.3,
  "fragsPerCycle":3.0,"bytesPerCycle":4200.0,
  "cycleAvgMs":10.0,"cycleStdMs":0.1,"cycleMinMs":9.8,"cycleMaxMs":10.4,
  "cycleHistMin":9.8,"cycleHistMax":10.4,"cycleHist":[0,1,5,"…(20 bins)"]}}}

triggerState

{"type":"triggerState","state":"triggered","mode":"normal",
 "stopped":false,"trigTime":1765370000.1234567}

state ∈ idle | armed | collecting | triggered; trigTime present once a trigger has fired.

The Go hub adds bufferFill (0…1) and bufferNeedSec while state is armed and its buffers do not yet reach back far enough to deliver a whole window. Edges are ignored until they do, so that no capture arrives with a front that was never recorded; the fields are absent once the requirement is met. bufferNeedSec is how far back the hub must reach now, which is less than the window by however much its buffers will fill in on their own while the post-trigger window is collected: the pre-trigger span while they keep up with the stream, and up to the whole windowSec when they do not (a full ring re-bucketing for a longer window fills slower than real time, so the front of a capture recedes while it is being collected). The event is re-broadcast as the fraction grows, so a client can show the progress instead of an armed trigger that appears to be ignoring the signal. forceTrigger fires regardless.

zoom (reply)

{"type":"zoom","reqId":17,"signals":{
  "s1:Sine":{"t":[1765370000.1230000,"…"],"v":[0.123456789,"…"]}}}

t is serialised with %.17g (full float64 precision — required for µs windows at Unix-epoch magnitudes), v with %.9g.

historyInfo

Sent on client connect (if history is enabled) and on historyInfo command:

{"type":"historyInfo","enabled":true,"windowSec":600.0,"decimation":10,
 "maxMPtsPerSignal":16.777216,
 "signals":{
   "scalar:Sine1":{"t0":1765360000.0,"t1":1765370000.0,"count":360000,"capacity":360000,"bucket":1},
   "scalar:Sine2":{"t0":1765360000.0,"t1":1765370000.0,"count":360000,"capacity":360000,"bucket":1}}}

• enabled — true if the +History config block is present and valid.
• windowSec — the timespan the files are sized to hold, i.e. the live or trigger window the clients are displaying (Go hub). The C++ StreamHub instead keeps a fixed retention period and reports it as durationHours.
• decimation — samples-to-disk decimation factor (1 = every sample).
• maxMPtsPerSignal — current per-signal budget, in millions of stored points (Go hub only; see setHistoryBudget).
• signals — per-signal metadata keyed by "sourceId:signalName":
  • t0/t1 — oldest/newest timestamp stored on disk (Unix seconds).
  • count — number of valid entries currently in the circular file.
  • capacity — total capacity of the circular file.
  • bucket — source samples per stored min/max pair (Go hub only); 1 means the signal is archived verbatim, higher means it is stored as an envelope because it is too fast to fit the budget at full resolution.

historyZoom (reply)

Same shape as zoom reply, but "type":"historyZoom":

{"type":"historyZoom","reqId":42,"signals":{
  "s1:Sine":{"t":[1765360000.1230000,"…"],"v":[0.123456789,"…"]}}}

If history is not enabled: {"type":"historyZoom","error":"history not enabled"}.

maxPointsUpdated

{"type":"maxPointsUpdated","maxPoints":50000}

calibration

{"type":"calibration","cal":[
  {"source":"wave","signal":"Adc","scale":0.00030518,"offset":-1.25,"unit":"V"}
]}

The complete calibration table. Broadcast when a client connects (as an empty array when nothing is calibrated), after every accepted setCalibration, and after a successful reloadConfig. It is a separate frame rather than a field on sources because sources is serialised into a fixed 4 KiB buffer.

configSaved

{"type":"configSaved","ok":true,"path":"/etc/streamhub/sources.json"}
{"type":"configSaved","ok":false,"path":"","error":"no sources file configured"}

Broadcast in reply to saveSources. path is always present (empty when the hub has no config file configured); error only when ok is false. The exact error text is not part of the protocol contract and differs between hubs (the Go hub uses "no sources-file configured", the C++ hub "no sources file configured").

configReloaded

{"type":"configReloaded","ok":true,"path":"/etc/streamhub/sources.json"}
{"type":"configReloaded","ok":false,"path":"/etc/streamhub/sources.json","error":"cannot read sources file"}

Broadcast in reply to reloadConfig; same shape as configSaved. On success it is followed by a calibration broadcast. Whether a sources broadcast also follows is hub-specific: the Go hub sends it only if the reload added at least one new source; the C++ hub sends it unconditionally. Clients must tolerate an unsolicited sources frame after any reload.

---

3. Binary frames (hub → client)

The first byte of every binary WS frame is a version discriminator.

Version 1 — data push

Sent at PushRate Hz per source. Contains only the samples that are new since the previous push (per-signal cursors hub-side), LTTB-decimated to at most MaxPushPoints (default 50) per signal.

[1]   version = 1
[1]   sourceIdLen (L)
[L]   sourceId (UTF-8, no NUL)
[4]   numSignals (uint32)

per signal:
  [2]   keyLen (uint16) (K)
  [K]   key = signal name; "name[i]" per element for multi-element PACKET signals
  [4]   pairCount (uint32) (N)
  [N×8] t (float64, Unix seconds)
  [N×8] v (float64, physical units)

Clients must append samples verbatim — there is no overlap between pushes.

Version 2 — trigger capture

Broadcast once per capture (FSM COLLECTING → TRIGGERED). Contains all signals over [trigTime − preSec, trigTime + postSec], each LTTB-decimated to ≤ 20 000 points.

[1]   version = 2
[8]   trigTime (float64, Unix seconds)
[8]   preSec   (float64)
[8]   postSec  (float64)
[4]   numSignals (uint32)

per signal:
  [2]   keyLen (uint16) (K)
  [K]   fullKey = "src:sig" (UTF-8)
  [4]   pairCount (uint32) (N)
  [N×8] t (float64, Unix seconds)
  [N×8] v (float64)

---

4. Config file format

SourcesFile (C++ SourcesFile config key, Go -sources-file flag) is a flat JSON array of flat objects. A block containing addr is a source; a block containing signal is a calibration entry; anything else is skipped with a warning.

[
  {"label": "wave", "addr": "127.0.0.1:44500"},
  {"label": "mc",   "addr": "127.0.0.1:44501", "multicastGroup": "239.0.0.1", "dataPort": 44502},
  {"source": "wave", "signal": "Adc", "scale": 0.00030518, "offset": -1.25, "unit": "V"}
]

Every object must stay flat. The C++ StreamHub::LoadSourcesFile parser takes each { up to the next } as one object, so a nested object anywhere in the file would truncate the parse at the inner brace. A nested "calibration": {…} inside a source entry is therefore not an option, and this is why calibration entries are siblings of sources rather than children.

Files written by hub versions predating calibration load unchanged, and a file written by either hub loads in the other.

---

5. Limits

Limit	Value
Concurrent WS clients	16
UDPS source sessions	32
Max received WS payload	64 KiB
Max sent WS payload	4 MiB
Calibration entries	256 (C++ hub, kMaxCalibration); unbounded (Go hub)
Calibration unit override	16 UTF-8 bytes