From cdafb877a3e08d4dc655cf57f8895a02a3deb75b Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Sun, 16 Aug 2026 19:56:44 +0200 Subject: [PATCH] StreamHub: per-signal calibration, config reload, whitespace-tolerant JSON - 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 ). - 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 --- Source/Applications/StreamHub/StreamHub.cpp | 355 +++++++++++++++++--- Source/Applications/StreamHub/StreamHub.h | 63 +++- 2 files changed, 373 insertions(+), 45 deletions(-) diff --git a/Source/Applications/StreamHub/StreamHub.cpp b/Source/Applications/StreamHub/StreamHub.cpp index 087eff3..ce45342 100644 --- a/Source/Applications/StreamHub/StreamHub.cpp +++ b/Source/Applications/StreamHub/StreamHub.cpp @@ -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(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(0)) { + delete[] calibration_; + calibration_ = static_cast(0); + } if (pushBuf_ != static_cast(0)) { delete[] pushBuf_; pushBuf_ = static_cast(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(0)) || (source[0] == '\0')) { return false; } + if ((signal == static_cast(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(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(0)) ? errText : ""); + } + if (n > 0) { wsServer_.BroadcastText(msg, static_cast(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(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(0)) { return; } /* missing file is fine */ + if (f == static_cast(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(fsz) + 1u]; const MARTe::osulong nRead = fread(data, 1u, static_cast(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(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(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(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(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(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(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(0); +} + +/** + * Finite check without : 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(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(0)) { return false; } - p += strlen(pattern); - while (*p == ' ') { p++; } if (*p == '\0') { return false; } out = strtod(p, static_cast(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(0)) { return false; } - p += strlen(pattern); - while (*p == ' ') { p++; } if (strncmp(p, "true", 4u) == 0) { out = true; return true; diff --git a/Source/Applications/StreamHub/StreamHub.h b/Source/Applications/StreamHub/StreamHub.h index 5524854..e2c49e8 100644 --- a/Source/Applications/StreamHub/StreamHub.h +++ b/Source/Applications/StreamHub/StreamHub.h @@ -44,6 +44,32 @@ using MARTe::StructuredDataI; /** Maximum number of simultaneously connected UDPStreamer sources. */ static const uint32 kMaxSessions = 32u; +/** Maximum number of stored per-signal calibration entries. */ +static const uint32 kMaxCalibration = 256u; + +/** Maximum length of a calibration unit override (mirrors the Go maxUnitLen). */ +static const uint32 kMaxUnitLen = 16u; + +/** + * @brief One per-signal affine calibration: y = raw*scale + offset. + * + * Keyed by the source LABEL (not the runtime "sN" id, which is assigned in + * add-order and would rebind if the source list were reordered) and by the + * BASE signal name (no "[i]" suffix: one entry covers a whole array signal). + * + * Fixed-size char arrays are used deliberately: embedding 256 StreamString + * (each of which allocates its own heap buffer) into a 133 MB struct that is + * itself heap-allocated pushes offsets beyond the canonical x86-64 address + * limit and causes a SIGSEGV in the constructor. + */ +struct CalibrationEntry { + char source[128]; ///< Source label + char signal[128]; ///< Base signal name (no "[i]" suffix) + char unit[17]; ///< Unit override (max kMaxUnitLen chars + NUL) + MARTe::float64 scale; + MARTe::float64 offset; +}; + /** * @brief Top-level StreamHub orchestrator. * @@ -108,6 +134,12 @@ private: /** Broadcast {"type":"config","sourceId":...} for one session. */ void BroadcastConfig(uint32 sessionIdx); + /** Broadcast {"type":"calibration","cal":[...]} to all clients. */ + void BroadcastCalibration(); + + /** Broadcast {"type":"configSaved"|"configReloaded","ok":...} to all clients. */ + void BroadcastConfigAck(const char *msgType, bool ok, const char *errText); + /* ---- Trigger (push loop side) ----------------------------------------- */ /** @@ -146,6 +178,8 @@ private: void HandleHistoryInfo(uint32 slotIdx); void HandleSetMaxPoints(const char *json); void HandlePing(uint32 slotIdx); + void HandleSetCalibration(const char *json); + void HandleReloadConfig(); /* ---- Binary recorder commands --------------------------------------- */ @@ -172,10 +206,29 @@ private: const char *mcGroup, uint16 dataPort); /** - * @brief Load sources from sourcesFile_ (JSON array of - * {"label","addr","multicastGroup","dataPort"}) and start them. + * @brief Load sources and calibration from sourcesFile_ (a flat JSON array + * of {"label","addr","multicastGroup","dataPort"} source blocks and + * {"source","signal","scale","offset","unit"} calibration blocks). + * @param skipActive when true, a source whose "host:port" is already + * streaming is left alone instead of being started a second time. + * @return true if the file was read. */ - void LoadSourcesFile(); + bool LoadSourcesFile(bool skipActive); + + /** @return true if a session for this "host:port" is already active. */ + bool SourceIsActive(const char *addrPort); + + /** + * @brief Store or replace one calibration entry. An identity entry + * (scale 1, offset 0, empty unit) removes any stored one instead. + * @return true if the entry was valid (and therefore stored or removed). + */ + bool SetCalibrationEntry(const char *source, const char *signal, + MARTe::float64 scale, MARTe::float64 offset, + const char *unit); + + /** Drop every calibration entry (used by reload, which replaces wholesale). */ + void ClearCalibration(); /* ---- Tiny JSON helpers ----------------------------------------------- */ @@ -220,6 +273,10 @@ private: StreamString sourcesFile_; ///< Persistent dynamic source list (JSON) uint32 nextSourceId_; ///< Counter for generated session ids ("sN") + CalibrationEntry *calibration_; ///< Heap-allocated array[kMaxCalibration] + uint32 numCalibration_; + FastPollingMutexSem calibrationMutex_; ///< Serializes calibration reads/writes + /* Push loop state */ volatile bool running_; uint32 tickCount_; ///< incremented each push tick