martino/MARTe-Integrated-Components
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

StreamHub: per-signal calibration, config reload, whitespace-tolerant JSON

Browse Source 
- Add CalibrationEntry (heap-allocated array[256], char[] fields to stay within
  the 133 MB struct's canonical address limit) plus calibrationMutex_ and
  numCalibration_.
- Implement SetCalibrationEntry/ClearCalibration, BroadcastCalibration,
  BroadcastConfigAck, HandleSetCalibration, HandleReloadConfig.
- Wire setCalibration and reloadConfig into OnWSCommand dispatch.
- Broadcast calibration to each newly connected client after triggerState.
- Fix JSON round-trip bug: replace JsonGetString/JsonGetBool helpers with a
  shared whitespace-tolerant JsonFindValue (tolerates "key" : "value" as
  written by HandleSaveSources); add JsonIsFinite (no <cmath>).
- Extend LoadSourcesFile(bool skipActive) to also parse calibration blocks;
  SourceIsActive checks live sessions before starting a duplicate.
- Extend HandleSaveSources to persist calibration blocks; emit configSaved ack.
- Reload semantics: calibration replaced wholesale, sources added only.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Martino Ferrari
2026-08-16 19:56:44 +02:00
co-authored by Claude Sonnet 4.6
parent ffe7cb1cc5
commit cdafb877a3
2 changed files with 373 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/Applications/StreamHub/StreamHub.cpp
+313 -42
View File
@@ -19,6 +19,10 @@ namespace StreamHub {
using MARTe::Sleep;
/* Forward declarations for file-scope helpers defined later. */
static const char *JsonFindValue(const char *json, const char *key);
static bool JsonIsFinite(MARTe::float64 v);
/**
* @brief printf-append into a heap buffer, growing it on demand.
* @return false only on encoding error.
@@ -62,6 +66,8 @@ StreamHub::StreamHub()
ringTemporal_(1000000u),
ringScalar_(100000u),
nextSourceId_(1u),
calibration_(static_cast<CalibrationEntry *>(0)),
numCalibration_(0u),
running_(false),
tickCount_(0u),
pendingMaxPointsSet_(false),
@@ -75,6 +81,8 @@ StreamHub::StreamHub()
rearmPending_(false),
rearmAtWallS_(0.0) {
memset(&recorderCfg_, 0, sizeof(recorderCfg_));
calibration_ = new CalibrationEntry[kMaxCalibration];
memset(calibration_, 0, sizeof(CalibrationEntry) * kMaxCalibration);
for (uint32 i = 0u; i < kMaxSessions; i++) {
sessionActive_[i] = false;
configBroadcast_[i] = false;
@@ -88,6 +96,10 @@ StreamHub::StreamHub()
StreamHub::~StreamHub() {
Stop();
if (calibration_ != static_cast<CalibrationEntry *>(0)) {
delete[] calibration_;
calibration_ = static_cast<CalibrationEntry *>(0);
}
if (pushBuf_ != static_cast<uint8 *>(0)) {
delete[] pushBuf_;
pushBuf_ = static_cast<uint8 *>(0);
@@ -249,7 +261,7 @@ bool StreamHub::Initialise(StructuredDataI &cfg) {
}
/* Start any persisted dynamic sources (Go SourceConfig schema). */
LoadSourcesFile();
(void) LoadSourcesFile(false);
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information,
"StreamHub: initialised with %u session(s), WSPort=%u, MaxPoints=%u, PushRate=%u Hz.",
@@ -687,6 +699,120 @@ void StreamHub::BroadcastConfig(uint32 idx) {
delete[] buf;
}
/*---------------------------------------------------------------------------*/
/* Calibration store */
/*---------------------------------------------------------------------------*/
bool StreamHub::SetCalibrationEntry(const char *source, const char *signal,
float64 scale, float64 offset,
const char *unit) {
if ((source == static_cast<const char *>(0)) || (source[0] == '\0')) { return false; }
if ((signal == static_cast<const char *>(0)) || (signal[0] == '\0')) { return false; }
/* A zero or non-finite scale makes the calibration non-invertible, which
* the SPA's trigger-threshold conversion depends on. */
if (!JsonIsFinite(scale) || (scale == 0.0)) { return false; }
if (!JsonIsFinite(offset)) { return false; }
char u[kMaxUnitLen + 1u];
u[0] = '\0';
if (unit != static_cast<const char *>(0)) {
strncpy(u, unit, kMaxUnitLen);
u[kMaxUnitLen] = '\0';
}
/* An identity entry carries no information: drop it rather than store and
* persist it. */
const bool identity = (scale == 1.0) && (offset == 0.0) && (u[0] == '\0');
(void) calibrationMutex_.FastLock();
uint32 found = kMaxCalibration;
for (uint32 i = 0u; i < numCalibration_; i++) {
if ((strcmp(calibration_[i].source, source) == 0) &&
(strcmp(calibration_[i].signal, signal) == 0)) {
found = i;
break;
}
}
if (identity) {
if (found < numCalibration_) {
/* Compact by moving the last entry into the freed slot. */
calibration_[found] = calibration_[numCalibration_ - 1u];
numCalibration_--;
}
calibrationMutex_.FastUnLock();
return true;
}
if (found == kMaxCalibration) {
if (numCalibration_ >= kMaxCalibration) {
calibrationMutex_.FastUnLock();
return false;
}
found = numCalibration_;
numCalibration_++;
}
strncpy(calibration_[found].source, source, sizeof(calibration_[found].source) - 1u);
calibration_[found].source[sizeof(calibration_[found].source) - 1u] = '\0';
strncpy(calibration_[found].signal, signal, sizeof(calibration_[found].signal) - 1u);
calibration_[found].signal[sizeof(calibration_[found].signal) - 1u] = '\0';
strncpy(calibration_[found].unit, u, sizeof(calibration_[found].unit) - 1u);
calibration_[found].unit[sizeof(calibration_[found].unit) - 1u] = '\0';
calibration_[found].scale = scale;
calibration_[found].offset = offset;
calibrationMutex_.FastUnLock();
return true;
}
void StreamHub::ClearCalibration() {
(void) calibrationMutex_.FastLock();
numCalibration_ = 0u;
calibrationMutex_.FastUnLock();
}
void StreamHub::BroadcastCalibration() {
/* Own growable buffer, like BroadcastConfig: the fixed 4096-byte buffer
* BroadcastSources uses would overflow on a full calibration table. */
uint32 cap = 16384u;
char *buf = new char[cap];
uint32 off = 0u;
JsonAppendf(buf, off, cap, "{\"type\":\"calibration\",\"cal\":[");
(void) calibrationMutex_.FastLock();
for (uint32 i = 0u; i < numCalibration_; i++) {
JsonAppendf(buf, off, cap,
"%s{\"source\":\"%s\",\"signal\":\"%s\","
"\"scale\":%.17g,\"offset\":%.17g,\"unit\":\"%s\"}",
(i > 0u) ? "," : "",
calibration_[i].source,
calibration_[i].signal,
calibration_[i].scale,
calibration_[i].offset,
calibration_[i].unit);
}
calibrationMutex_.FastUnLock();
JsonAppendf(buf, off, cap, "]}");
wsServer_.BroadcastText(buf, off);
delete[] buf;
}
void StreamHub::BroadcastConfigAck(const char *msgType, bool ok,
const char *errText) {
char msg[512];
int n;
if (ok) {
n = snprintf(msg, sizeof(msg),
"{\"type\":\"%s\",\"ok\":true,\"path\":\"%s\"}",
msgType, sourcesFile_.Buffer());
}
else {
n = snprintf(msg, sizeof(msg),
"{\"type\":\"%s\",\"ok\":false,\"path\":\"%s\",\"error\":\"%s\"}",
msgType, sourcesFile_.Buffer(),
(errText != static_cast<const char *>(0)) ? errText : "");
}
if (n > 0) { wsServer_.BroadcastText(msg, static_cast<uint32>(n)); }
}
/*---------------------------------------------------------------------------*/
/* WSCommandCallback */
/*---------------------------------------------------------------------------*/
@@ -707,6 +833,9 @@ void StreamHub::OnWSClientConnected() {
/* Let the new client render the current trigger badge/buttons. */
BroadcastTriggerState();
/* Let the new client apply the stored per-signal calibration. */
BroadcastCalibration();
/* Inform the new client about history availability. */
if (history_.IsEnabled()) {
/* Broadcast to all (simple; no unicast-on-connect path for broadcast). */
@@ -758,7 +887,9 @@ void StreamHub::OnWSCommand(const char *json, uint32 /*len*/, uint32 slotIdx) {
else if (strcmp(type, "recStart") == 0) { HandleRecStart(json); }
else if (strcmp(type, "recStop") == 0) { HandleRecStop(json); }
else if (strcmp(type, "recInfo") == 0) { HandleRecInfo(slotIdx); }
else if (strcmp(type, "ping") == 0) { HandlePing(slotIdx); }
else if (strcmp(type, "ping") == 0) { HandlePing(slotIdx); }
else if (strcmp(type, "setCalibration") == 0) { HandleSetCalibration(json); }
else if (strcmp(type, "reloadConfig") == 0) { HandleReloadConfig(); }
}
/*---------------------------------------------------------------------------*/
@@ -851,26 +982,41 @@ bool StreamHub::AddSourceInternal(const char *label, const char *addrPort,
return ok;
}
void StreamHub::LoadSourcesFile() {
if (sourcesFile_.Size() == 0u) { return; }
bool StreamHub::SourceIsActive(const char *addrPort) {
for (uint32 i = 0u; i < kMaxSessions; i++) {
if (!sessionActive_[i]) { continue; }
StreamString adr = sessions_[i].GetAddr();
char cur[96];
(void) snprintf(cur, sizeof(cur), "%s:%u", adr.Buffer(),
static_cast<uint32>(sessions_[i].GetPort()));
if (strcmp(cur, addrPort) == 0) { return true; }
}
return false;
}
bool StreamHub::LoadSourcesFile(bool skipActive) {
if (sourcesFile_.Size() == 0u) { return false; }
FILE *f = fopen(sourcesFile_.Buffer(), "rb");
if (f == static_cast<FILE *>(0)) { return; } /* missing file is fine */
if (f == static_cast<FILE *>(0)) { return false; } /* missing file is fine */
(void) fseek(f, 0, SEEK_END);
const long fsz = ftell(f);
(void) fseek(f, 0, SEEK_SET);
if ((fsz <= 0) || (fsz > (1L << 20))) {
(void) fclose(f);
return;
return false;
}
char *data = new char[static_cast<uint32>(fsz) + 1u];
const MARTe::osulong nRead = fread(data, 1u, static_cast<MARTe::osulong>(fsz), f);
data[nRead] = '\0';
(void) fclose(f);
/* JSON array of flat objects — iterate over each {...} block. */
/* Flat JSON array of flat objects — iterate over each {...} block. A block
* with "addr" is a source, one with "signal" is a calibration. The array
* must stay flat: this scanner takes each "{" up to the next "}". */
uint32 nLoaded = 0u;
uint32 nCal = 0u;
const char *p = data;
while ((p = strchr(p, '{')) != static_cast<const char *>(0)) {
const char *end = strchr(p, '}');
@@ -882,37 +1028,72 @@ void StreamHub::LoadSourcesFile() {
memcpy(obj, p, objLen);
obj[objLen] = '\0';
char label[128] = "";
char addr[80] = "";
char mcGroup[64] = "";
float64 dataPortF = 0.0;
JsonGetString(obj, "label", label, sizeof(label));
JsonGetString(obj, "addr", addr, sizeof(addr));
JsonGetString(obj, "multicastGroup", mcGroup, sizeof(mcGroup));
JsonGetFloat(obj, "dataPort", dataPortF);
char addr[80] = "";
(void) JsonGetString(obj, "addr", addr, sizeof(addr));
if (AddSourceInternal(label, addr, mcGroup,
static_cast<uint16>(dataPortF))) {
nLoaded++;
if (addr[0] != '\0') {
char label[128] = "";
char mcGroup[64] = "";
float64 dataPortF = 0.0;
(void) JsonGetString(obj, "label", label, sizeof(label));
(void) JsonGetString(obj, "multicastGroup", mcGroup, sizeof(mcGroup));
(void) JsonGetFloat(obj, "dataPort", dataPortF);
if (skipActive && SourceIsActive(addr)) {
/* Already streaming — leave the live session untouched. */
}
else if (AddSourceInternal(label, addr, mcGroup,
static_cast<uint16>(dataPortF))) {
nLoaded++;
}
}
else {
char calSignal[128] = "";
(void) JsonGetString(obj, "signal", calSignal, sizeof(calSignal));
if (calSignal[0] != '\0') {
char calSource[128] = "";
char calUnit[64] = "";
float64 calScale = 1.0;
float64 calOffset = 0.0;
(void) JsonGetString(obj, "source", calSource, sizeof(calSource));
(void) JsonGetString(obj, "unit", calUnit, sizeof(calUnit));
(void) JsonGetFloat(obj, "scale", calScale);
(void) JsonGetFloat(obj, "offset", calOffset);
if (SetCalibrationEntry(calSource, calSignal,
calScale, calOffset, calUnit)) {
nCal++;
}
else {
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning,
"StreamHub: skipping invalid calibration '%s'/'%s'.",
calSource, calSignal);
}
}
else {
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning,
"StreamHub: skipping unrecognised config block.");
}
}
p = end + 1;
}
delete[] data;
if (nLoaded > 0u) {
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information,
"StreamHub: loaded %u source(s) from '%s'.",
nLoaded, sourcesFile_.Buffer());
}
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information,
"StreamHub: loaded %u source(s) and %u calibration entr(y/ies) from '%s'.",
nLoaded, nCal, sourcesFile_.Buffer());
return true;
}
void StreamHub::HandleSaveSources() {
if (sourcesFile_.Size() == 0u) { return; }
if (sourcesFile_.Size() == 0u) {
BroadcastConfigAck("configSaved", false, "no sources file configured");
return;
}
FILE *f = fopen(sourcesFile_.Buffer(), "wb");
if (f == static_cast<FILE *>(0)) {
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning,
"StreamHub: cannot write sources file '%s'.", sourcesFile_.Buffer());
BroadcastConfigAck("configSaved", false, "cannot open file for writing");
return;
}
@@ -939,11 +1120,35 @@ void StreamHub::HandleSaveSources() {
(void) fprintf(f, "\n }");
nSaved++;
}
/* Calibration entries are further elements of the SAME flat array. */
uint32 nCal = 0u;
(void) calibrationMutex_.FastLock();
for (uint32 i = 0u; i < numCalibration_; i++) {
(void) fprintf(f,
"%s {\n \"source\": \"%s\",\n \"signal\": \"%s\",\n"
" \"scale\": %.17g,\n \"offset\": %.17g",
((nSaved + nCal) > 0u) ? ",\n" : "",
calibration_[i].source,
calibration_[i].signal,
calibration_[i].scale,
calibration_[i].offset);
if (calibration_[i].unit[0] != '\0') {
(void) fprintf(f, ",\n \"unit\": \"%s\"",
calibration_[i].unit);
}
(void) fprintf(f, "\n }");
nCal++;
}
calibrationMutex_.FastUnLock();
(void) fprintf(f, "\n]\n");
(void) fclose(f);
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information,
"StreamHub: saved %u source(s) to '%s'.", nSaved, sourcesFile_.Buffer());
"StreamHub: saved %u source(s) and %u calibration entr(y/ies) to '%s'.",
nSaved, nCal, sourcesFile_.Buffer());
BroadcastConfigAck("configSaved", true, "");
}
void StreamHub::HandleRemoveSource(const char *json) {
@@ -1001,6 +1206,50 @@ void StreamHub::HandleGetStats() {
PushStats();
}
void StreamHub::HandleSetCalibration(const char *json) {
char source[128] = "";
char signal[128] = "";
char unit[64] = "";
float64 scale = 1.0;
float64 offset = 0.0;
(void) JsonGetString(json, "source", source, sizeof(source));
(void) JsonGetString(json, "signal", signal, sizeof(signal));
(void) JsonGetString(json, "unit", unit, sizeof(unit));
(void) JsonGetFloat(json, "scale", scale);
(void) JsonGetFloat(json, "offset", offset);
/* One entry covers a whole array signal: strip any "[i]" element suffix. */
char *br = strchr(signal, '[');
if (br != static_cast<char *>(0)) { *br = '\0'; }
if (SetCalibrationEntry(source, signal, scale, offset, unit)) {
BroadcastCalibration();
}
else {
/* No broadcast: the offending client reverts to its last known value. */
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning,
"StreamHub: rejected calibration for '%s'/'%s'.", source, signal);
}
}
void StreamHub::HandleReloadConfig() {
if (sourcesFile_.Size() == 0u) {
BroadcastConfigAck("configReloaded", false, "no sources file configured");
return;
}
/* Calibration is replaced wholesale; sources are only added. A reload must
* never interrupt a live UDP session. */
ClearCalibration();
if (!LoadSourcesFile(true)) {
BroadcastConfigAck("configReloaded", false, "cannot read sources file");
return;
}
BroadcastConfigAck("configReloaded", true, "");
BroadcastCalibration();
BroadcastSources();
}
void StreamHub::HandleArm() {
rearmPending_ = false;
trigger_.Arm();
@@ -1573,17 +1822,47 @@ void StreamHub::BroadcastRecStatus() {
/* Tiny JSON helpers */
/*---------------------------------------------------------------------------*/
/**
* Locate the value text for "key" in a flat JSON object, tolerating whitespace
* around the colon. Occurrences of the token that are NOT followed by a colon
* are skipped, so a value that happens to equal a key name (for example
* {"label": "addr", "addr": "..."}) does not shadow the real key.
* @return pointer to the first character of the value, or 0 if not found.
*/
static const char *JsonFindValue(const char *json, const char *key) {
char pattern[128];
(void) snprintf(pattern, sizeof(pattern), "\"%s\"", key);
const size_t plen = strlen(pattern);
const char *p = json;
while ((p = strstr(p, pattern)) != static_cast<const char *>(0)) {
const char *q = p + plen;
while ((*q == ' ') || (*q == '\t') || (*q == '\n') || (*q == '\r')) { q++; }
if (*q == ':') {
q++;
while ((*q == ' ') || (*q == '\t') || (*q == '\n') || (*q == '\r')) { q++; }
return q;
}
p += plen;
}
return static_cast<const char *>(0);
}
/**
* Finite check without <cmath>: NaN fails self-comparison, and both infinities
* fall outside the largest representable finite double.
*/
static bool JsonIsFinite(MARTe::float64 v) {
return (v == v) && (v < 1.0e308) && (v > -1.0e308);
}
bool StreamHub::JsonGetString(const char *json, const char *key,
char *out, uint32 outSize) {
/* Look for "key":"value" */
char pattern[128];
snprintf(pattern, sizeof(pattern), "\"%s\":\"", key);
const char *p = strstr(json, pattern);
const char *p = JsonFindValue(json, key);
if (p == static_cast<const char *>(0)) { return false; }
p += strlen(pattern);
if (*p != '"') { return false; }
p++;
uint32 i = 0u;
while (*p != '\0' && *p != '"' && i < outSize - 1u) {
while ((*p != '\0') && (*p != '"') && (i < (outSize - 1u))) {
out[i++] = *p++;
}
out[i] = '\0';
@@ -1591,12 +1870,8 @@ bool StreamHub::JsonGetString(const char *json, const char *key,
}
bool StreamHub::JsonGetFloat(const char *json, const char *key, float64 &out) {
char pattern[128];
snprintf(pattern, sizeof(pattern), "\"%s\":", key);
const char *p = strstr(json, pattern);
const char *p = JsonFindValue(json, key);
if (p == static_cast<const char *>(0)) { return false; }
p += strlen(pattern);
while (*p == ' ') { p++; }
if (*p == '\0') { return false; }
out = strtod(p, static_cast<char **>(0));
return true;
@@ -1610,12 +1885,8 @@ bool StreamHub::JsonGetUint32(const char *json, const char *key, uint32 &out) {
}
bool StreamHub::JsonGetBool(const char *json, const char *key, bool &out) {
char pattern[128];
snprintf(pattern, sizeof(pattern), "\"%s\":", key);
const char *p = strstr(json, pattern);
const char *p = JsonFindValue(json, key);
if (p == static_cast<const char *>(0)) { return false; }
p += strlen(pattern);
while (*p == ' ') { p++; }
if (strncmp(p, "true", 4u) == 0) {
out = true;
return true;