feat(udpscope): per-element timestamp reconstruction from UDPS frames

Implements FrameDecoder with five timing rules that mirror
UDPSourceSession.cpp: FullArray (per-element time signal), FirstSample
and LastSample (rate-spread from anchor), accumulated scalar with
declared rate (forward-chain anchoring, immune to arrival jitter), and
PACKET burst (backward-span from previous arrival). 9 new tests, 38
pass total.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Martino Ferrari
2026-08-27 20:01:30 +02:00
co-authored by Claude Sonnet 4.6
parent c89decef8e
commit 5a8479cda9
5 changed files with 583 additions and 0 deletions
+1
View File
@@ -83,6 +83,7 @@ set(CORE_SOURCES
Decimate.cpp Decimate.cpp
PaneTree.cpp PaneTree.cpp
TimeBase.cpp TimeBase.cpp
FrameDecoder.cpp
) )
add_library(udpscope_core STATIC ${CORE_SOURCES}) add_library(udpscope_core STATIC ${CORE_SOURCES})
+173
View File
@@ -0,0 +1,173 @@
#include "FrameDecoder.h"
namespace udpscope {
/** Fallback cycle period before the first inter-packet gap is known. */
static constexpr double kDefaultDt = 1.0e-3;
void FrameDecoder::setSignals(const std::vector<SignalMeta>& signals) {
signals_ = signals;
state_.assign(signals_.size(), SigState{});
hrtFit_.reset();
}
void FrameDecoder::reset() {
state_.assign(signals_.size(), SigState{});
hrtFit_.reset();
}
void FrameDecoder::beginFrame(const FrameView& f) {
if (f.hrt != 0u) { hrtFit_.add(f.hrt, f.recvTime); }
}
bool FrameDecoder::packetBurst(uint32_t idx, uint32_t nElems, double wallNow,
std::vector<double>& tsOut) {
SigState& st = state_[idx];
if (!st.lastPacketValid || wallNow <= st.lastPacketWall) {
/* No previous arrival to span from, or time went backwards. Remember
* this one and drop the samples rather than store them at made-up
* spacing. */
st.lastPacketWall = wallNow;
st.lastPacketValid = true;
return false;
}
const double dt = (wallNow - st.lastPacketWall) / static_cast<double>(nElems);
tsOut.resize(nElems);
for (uint32_t e = 0; e < nElems; e++) {
tsOut[e] = st.lastPacketWall + static_cast<double>(e + 1u) * dt;
}
st.lastPacketWall = wallNow;
return true;
}
bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx,
std::vector<double>& tsOut) {
tsOut.clear();
if (idx >= signals_.size() || idx >= f.numSignals || f.counts == nullptr) {
return false;
}
const SignalMeta& d = signals_[idx];
const uint32_t nElems = f.counts[idx];
if (nElems == 0u) { return false; }
const double wallNow = f.recvTime;
SigState& st = state_[idx];
const bool hasTimeSig = d.hasTimeSignal(f.numSignals);
const uint32_t tIdx = hasTimeSig ? d.timeSignalIdx : 0u;
const double tScale = hasTimeSig
? TimeSignalScale(signals_[tIdx].typeCode)
: 1.0e-6;
/* Rule 1: one stamp per element, straight from the time signal. */
if (d.timeMode == kTimeFullArray && hasTimeSig &&
f.counts[tIdx] >= nElems && f.values[tIdx] != nullptr) {
const double* tv = f.values[tIdx];
const double t0 = tv[0] * tScale;
(void) st.offset.map(t0, wallNow);
const double base = st.offset.offset();
tsOut.resize(nElems);
for (uint32_t e = 0; e < nElems; e++) {
tsOut[e] = base + tv[e] * tScale;
}
return true;
}
/* Rule 2: anchor from the time signal, spread by the sampling rate. */
if ((d.timeMode == kTimeFirstSample || d.timeMode == kTimeLastSample) &&
hasTimeSig && f.counts[tIdx] >= 1u && f.values[tIdx] != nullptr) {
const double anchor = st.offset.map(f.values[tIdx][0] * tScale, wallNow);
const double dt = (d.samplingRate > 0.0) ? (1.0 / d.samplingRate) : 0.0;
tsOut.resize(nElems);
for (uint32_t e = 0; e < nElems; e++) {
tsOut[e] = (d.timeMode == kTimeFirstSample)
? (anchor + static_cast<double>(e) * dt)
: (anchor - static_cast<double>(nElems - 1u - e) * dt);
}
return true;
}
/* Rule 3: accumulated scalar, based on declared sampling rate or hrt.
*
* When samplingRate is declared the inter-element step is exact and we
* anchor from the end of the previous burst rather than from arrival time
* or hrt. This makes the output immune to arrival jitter: even when the
* kernel delivers two packets microseconds apart each burst starts exactly
* one sample period after the previous burst ended.
*
* When samplingRate is absent we must derive dt from the hrt gap, which
* requires the HrtRateFit to be ready. Until then we fall back to
* packetBurst (arrival-time spanning), which is accurate during the normal
* pre-burst delivery phase that precedes the fit becoming ready. */
if (d.numElements() == 1u && nElems > 1u) {
const double dt = (d.samplingRate > 0.0)
? (1.0 / d.samplingRate)
: 0.0;
if (d.samplingRate > 0.0) {
/* Forward-chain anchor: t[0] = lastEnd + dt, or wallNow on first
* packet (arrival-time for the very first burst only). */
double base;
if (st.lastEmittedValid) {
base = st.lastEmittedEnd + dt;
} else {
/* First packet: anchor element 0 at arrival time.
* This one packet may be slightly off, but subsequent packets
* chain from this end and jitter is suppressed thereafter. */
base = wallNow - static_cast<double>(nElems - 1u) * dt;
/* Calibrate the clock-offset so later hrt-based paths (if any)
* are consistent, but we don't use it in this branch. */
if (f.hrt != 0u && hrtFit_.ready()) {
const double hrtSec = hrtFit_.toSeconds(f.hrt);
(void) st.offset.map(hrtSec, wallNow);
}
}
tsOut.resize(nElems);
for (uint32_t e = 0; e < nElems; e++) {
tsOut[e] = base + static_cast<double>(e) * dt;
}
st.lastEmittedEnd = tsOut[nElems - 1u];
st.lastEmittedValid = true;
return true;
}
/* No declared rate: need hrt-derived dt. */
if (!hrtFit_.ready()) {
return packetBurst(idx, nElems, wallNow, tsOut);
}
const double hrtSec = hrtFit_.toSeconds(f.hrt);
const double base = st.offset.map(hrtSec, wallNow);
double hrtDt;
if (st.lastAccValid && st.prevAccCount > 0u && hrtSec > st.lastAccHrtSec) {
/* The flushes carry contiguous RT cycles, so the gap divided by the
* previous packet's sample count is exactly one cycle period. */
hrtDt = (hrtSec - st.lastAccHrtSec) /
static_cast<double>(st.prevAccCount);
} else {
hrtDt = kDefaultDt;
}
tsOut.resize(nElems);
for (uint32_t e = 0; e < nElems; e++) {
tsOut[e] = base + static_cast<double>(e) * hrtDt;
}
st.lastAccHrtSec = hrtSec;
st.lastAccValid = true;
st.prevAccCount = nElems;
return true;
}
/* Rule 4: PACKET burst with no time reference at all. */
if (nElems > 1u) {
return packetBurst(idx, nElems, wallNow, tsOut);
}
/* Rule 5: plain scalar. */
tsOut.assign(1, wallNow);
return true;
}
} /* namespace udpscope */
+68
View File
@@ -0,0 +1,68 @@
/**
* @file FrameDecoder.h
* @brief Per-element timestamp reconstruction for UDPS frames.
*
* The C client's udps_frame_element_time() is explicitly an arrival-anchored
* estimate. It is not sufficient: the kernel frequently delivers several queued
* datagrams in one burst, so two packets are processed microseconds apart even
* though each represents ~10 ms of signal, and arrival-time interpolation then
* crams a packet's samples into that tiny gap — the trace renders as a sawtooth.
* Source/Applications/StreamHub/UDPSourceSession.cpp documents this failure and
* solves it; these are the same rules, computed from udps_frame_t's own fields
* so the scope and StreamHub agree on the same stream.
*/
#pragma once
#include "TimeBase.h"
#include "Types.h"
#include <vector>
namespace udpscope {
class FrameDecoder {
public:
/** Installs the signal table. Clears all per-signal timing history. */
void setSignals(const std::vector<SignalMeta>& signals);
const std::vector<SignalMeta>& signals() const { return signals_; }
/** Call once per frame, before any timestamps() call for that frame. */
void beginFrame(const FrameView& f);
/**
* @brief Timestamps for every value of signal @p idx in this frame.
* @return false when the signal produced nothing usable — an empty slot, or
* the first PACKET burst after connect, which has no previous
* arrival to span from and would otherwise poison the ring with
* wrongly spaced timestamps.
*/
bool timestamps(const FrameView& f, uint32_t idx, std::vector<double>& tsOut);
/** Forgets all timing history; call on reconnect. */
void reset();
private:
bool packetBurst(uint32_t idx, uint32_t nElems, double wallNow,
std::vector<double>& tsOut);
struct SigState {
ClockOffset offset;
double lastPacketWall = 0.0;
bool lastPacketValid = false;
double lastAccHrtSec = 0.0;
bool lastAccValid = false;
uint32_t prevAccCount = 0;
/** For accumulated scalars with a declared sampling rate: end timestamp
* of the most recently emitted burst, used as a forward-chain anchor
* that is immune to arrival-time jitter. */
double lastEmittedEnd = 0.0;
bool lastEmittedValid = false;
};
std::vector<SignalMeta> signals_;
std::vector<SigState> state_;
HrtRateFit hrtFit_;
};
} /* namespace udpscope */
+65
View File
@@ -35,4 +35,69 @@ struct Rect {
} }
}; };
/* Protocol constants, spelled out rather than included, so the framework-free
* modules stay independent of udps_client.h. They mirror Common/UDP/UDPSProtocol.h. */
constexpr uint8_t kTimePacket = 0;
constexpr uint8_t kTimeFullArray = 1;
constexpr uint8_t kTimeFirstSample = 2;
constexpr uint8_t kTimeLastSample = 3;
constexpr uint32_t kNoTimeSignal = 0xFFFFFFFFu;
/** Framework-free mirror of udps_signal_t, plus UI state. */
struct SignalMeta {
std::string name;
uint8_t typeCode = 255;
uint8_t quantType = 0;
uint32_t numRows = 1;
uint32_t numCols = 1;
double rangeMin = 0.0;
double rangeMax = 0.0;
uint8_t timeMode = kTimePacket;
double samplingRate = 0.0;
uint32_t timeSignalIdx = kNoTimeSignal;
std::string unit;
/** User override: treat an ambiguous PACKET array as a profile, not a burst. */
bool profileOverride = false;
uint32_t numElements() const {
const uint64_t n = static_cast<uint64_t>(numRows ? numRows : 1u) *
static_cast<uint64_t>(numCols ? numCols : 1u);
return n == 0u ? 1u : static_cast<uint32_t>(n);
}
bool hasTimeSignal(uint32_t numSignals) const {
return timeSignalIdx != kNoTimeSignal && timeSignalIdx < numSignals;
}
/**
* @brief True when this array should be plotted against element index
* rather than unrolled onto the time axis.
*
* Only PACKET arrays are ambiguous: the producer stamped the whole datagram
* with one time, which is what a genuine vector looks like and also what a
* burst carrying no time metadata looks like. Default is burst, matching
* UDPSourceSession, with this flag as the user's override.
*/
bool isVectorProfile() const {
return profileOverride && numElements() > 1u && timeMode == kTimePacket;
}
};
/**
* @brief Non-owning mirror of udps_frame_t.
*
* Kept separate from the C struct so FrameDecoder can be tested with plain
* arrays and no socket. Points at memory owned by the caller.
*/
struct FrameView {
uint32_t counter = 0;
uint64_t hrt = 0;
double recvTime = 0.0;
uint32_t numSamples = 1;
uint32_t numSignals = 0;
const double* const* values = nullptr; /**< values[i][0..counts[i]) */
const uint32_t* counts = nullptr;
};
} /* namespace udpscope */ } /* namespace udpscope */
+276
View File
@@ -0,0 +1,276 @@
#include "FrameDecoder.h"
#include <gtest/gtest.h>
#include <vector>
using namespace udpscope;
namespace {
/** Builds a FrameView over vectors the test owns. */
struct FrameBuilder {
std::vector<std::vector<double>> storage;
std::vector<const double*> ptrs;
std::vector<uint32_t> counts;
FrameView view;
void addSignal(std::vector<double> vals) {
storage.push_back(std::move(vals));
}
const FrameView& build(uint64_t hrt, double recvTime, uint32_t numSamples = 1) {
ptrs.clear();
counts.clear();
for (const auto& s : storage) {
ptrs.push_back(s.data());
counts.push_back(static_cast<uint32_t>(s.size()));
}
view.hrt = hrt;
view.recvTime = recvTime;
view.numSamples = numSamples;
view.numSignals = static_cast<uint32_t>(storage.size());
view.values = ptrs.data();
view.counts = counts.data();
return view;
}
};
SignalMeta burst(const char* name, uint8_t timeMode, double rate,
uint32_t elems, uint32_t timeIdx) {
SignalMeta m;
m.name = name;
m.typeCode = 8; /* float32 */
m.numRows = elems;
m.numCols = 1;
m.timeMode = timeMode;
m.samplingRate = rate;
m.timeSignalIdx = timeIdx;
return m;
}
SignalMeta timeSignal(const char* name, uint32_t elems) {
SignalMeta m;
m.name = name;
m.typeCode = 6; /* uint64 -> nanoseconds */
m.numRows = elems;
m.numCols = 1;
return m;
}
} /* namespace */
TEST(FrameDecoder, FullArrayTakesOneStampPerElementFromTheTimeSignal) {
FrameDecoder dec;
dec.setSignals({burst("Sine", kTimeFullArray, 1000.0, 4, 1),
timeSignal("Time", 4)});
FrameBuilder fb;
fb.addSignal({1.0, 2.0, 3.0, 4.0});
/* Nanoseconds: 5.000, 5.001, 5.002, 5.003 s of producer time. */
fb.addSignal({5.0e9, 5.001e9, 5.002e9, 5.003e9});
const FrameView& f = fb.build(0, 1000.0);
dec.beginFrame(f);
std::vector<double> ts;
ASSERT_TRUE(dec.timestamps(f, 0, ts));
ASSERT_EQ(ts.size(), 4u);
/* Element 0 lands on the arrival time; the rest keep the producer spacing. */
EXPECT_NEAR(ts[0], 1000.000, 1e-9);
EXPECT_NEAR(ts[1], 1000.001, 1e-9);
EXPECT_NEAR(ts[2], 1000.002, 1e-9);
EXPECT_NEAR(ts[3], 1000.003, 1e-9);
}
TEST(FrameDecoder, FirstSampleAnchorsElementZeroAndCountsForward) {
FrameDecoder dec;
dec.setSignals({burst("Sine", kTimeFirstSample, 1000.0, 4, 1),
timeSignal("Time", 1)});
FrameBuilder fb;
fb.addSignal({1.0, 2.0, 3.0, 4.0});
fb.addSignal({7.0e9});
const FrameView& f = fb.build(0, 2000.0);
dec.beginFrame(f);
std::vector<double> ts;
ASSERT_TRUE(dec.timestamps(f, 0, ts));
ASSERT_EQ(ts.size(), 4u);
EXPECT_NEAR(ts[0], 2000.000, 1e-9);
EXPECT_NEAR(ts[3], 2000.003, 1e-9);
}
TEST(FrameDecoder, LastSampleAnchorsTheFinalElementAndCountsBackward) {
FrameDecoder dec;
dec.setSignals({burst("Sine", kTimeLastSample, 1000.0, 4, 1),
timeSignal("Time", 1)});
FrameBuilder fb;
fb.addSignal({1.0, 2.0, 3.0, 4.0});
fb.addSignal({7.0e9});
const FrameView& f = fb.build(0, 3000.0);
dec.beginFrame(f);
std::vector<double> ts;
ASSERT_TRUE(dec.timestamps(f, 0, ts));
ASSERT_EQ(ts.size(), 4u);
EXPECT_NEAR(ts[3], 3000.000, 1e-9);
EXPECT_NEAR(ts[0], 3000.000 - 0.003, 1e-9);
}
TEST(FrameDecoder, PlainScalarUsesArrivalTime) {
FrameDecoder dec;
SignalMeta m;
m.name = "Level";
m.typeCode = 9;
dec.setSignals({m});
FrameBuilder fb;
fb.addSignal({42.0});
const FrameView& f = fb.build(0, 1234.5);
dec.beginFrame(f);
std::vector<double> ts;
ASSERT_TRUE(dec.timestamps(f, 0, ts));
ASSERT_EQ(ts.size(), 1u);
EXPECT_DOUBLE_EQ(ts[0], 1234.5);
}
// This is the failure UDPSourceSession.cpp:560 documents. The kernel delivers
// two queued datagrams microseconds apart even though each carries 10 ms of
// signal. Dating from arrival crams the second packet's samples into that gap
// and the trace becomes a sawtooth; dating from the producer hrt does not.
TEST(FrameDecoder, AccumulatedScalarSurvivesBurstyDelivery) {
FrameDecoder dec;
SignalMeta m;
m.name = "Acc";
m.typeCode = 9;
m.numRows = 1;
m.samplingRate = 1000.0; /* 1 kHz, 10 samples = 10 ms per packet */
dec.setSignals({m});
const double ticks = 1.0e9;
std::vector<double> all;
for (int p = 0; p < 40; p++) {
FrameBuilder fb;
fb.addSignal(std::vector<double>(10, static_cast<double>(p)));
const double producerSec = 100.0 + p * 0.010;
/* Packets 20+ arrive in a burst, all within 50 us of each other. */
const double arrival = (p < 20) ? (500.0 + p * 0.010)
: (500.2 + (p - 20) * 0.00005);
const FrameView& f = fb.build(static_cast<uint64_t>(producerSec * ticks),
arrival, 10);
dec.beginFrame(f);
std::vector<double> ts;
if (dec.timestamps(f, 0, ts)) {
all.insert(all.end(), ts.begin(), ts.end());
}
}
ASSERT_GT(all.size(), 300u);
for (size_t i = 1; i < all.size(); i++) {
EXPECT_GT(all[i], all[i - 1]) << "non-monotonic at " << i;
EXPECT_NEAR(all[i] - all[i - 1], 0.001, 2e-4)
<< "spacing collapsed at " << i << " (sawtooth)";
}
}
TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) {
FrameDecoder dec;
SignalMeta m;
m.name = "Acc";
m.typeCode = 9;
m.samplingRate = 0.0; /* undeclared */
dec.setSignals({m});
const double ticks = 1.0e9;
std::vector<double> last;
for (int p = 0; p < 40; p++) {
FrameBuilder fb;
fb.addSignal(std::vector<double>(10, 1.0));
const double producerSec = 100.0 + p * 0.010; /* 10 ms per packet */
const FrameView& f = fb.build(static_cast<uint64_t>(producerSec * ticks),
700.0 + p * 0.010, 10);
dec.beginFrame(f);
std::vector<double> ts;
if (dec.timestamps(f, 0, ts)) { last = ts; }
}
ASSERT_EQ(last.size(), 10u);
/* 10 ms of producer time across 10 samples is a 1 ms period. */
EXPECT_NEAR(last[1] - last[0], 0.001, 1e-5);
}
// A PACKET burst has no per-element time at all. Elements span
// (lastPacket, thisPacket] — backwards from arrival, because the samples were
// acquired before the packet landed. Forward extrapolation would let a jittered
// packet overlap the next one and break ring monotonicity.
TEST(FrameDecoder, PacketBurstDropsTheFirstFrameThenSpansBackwards) {
FrameDecoder dec;
dec.setSignals({burst("Raw", kTimePacket, 0.0, 5, kNoTimeSignal)});
FrameBuilder fb1;
fb1.addSignal({1.0, 2.0, 3.0, 4.0, 5.0});
const FrameView& f1 = fb1.build(0, 10.0);
dec.beginFrame(f1);
std::vector<double> ts;
EXPECT_FALSE(dec.timestamps(f1, 0, ts))
<< "the first packet has no previous arrival to span from";
FrameBuilder fb2;
fb2.addSignal({6.0, 7.0, 8.0, 9.0, 10.0});
const FrameView& f2 = fb2.build(0, 10.05);
dec.beginFrame(f2);
ASSERT_TRUE(dec.timestamps(f2, 0, ts));
ASSERT_EQ(ts.size(), 5u);
EXPECT_GT(ts[0], 10.0);
EXPECT_NEAR(ts[4], 10.05, 1e-12);
EXPECT_NEAR(ts[1] - ts[0], 0.01, 1e-12);
}
TEST(FrameDecoder, PacketBurstStaysMonotonicUnderJitteredArrivals) {
FrameDecoder dec;
dec.setSignals({burst("Raw", kTimePacket, 0.0, 8, kNoTimeSignal)});
const double jitter[] = {0.0, 0.004, -0.003, 0.006, -0.002, 0.0, 0.005, -0.004};
std::vector<double> all;
for (int p = 0; p < 8; p++) {
FrameBuilder fb;
fb.addSignal(std::vector<double>(8, 1.0));
const FrameView& f = fb.build(0, 20.0 + p * 0.05 + jitter[p]);
dec.beginFrame(f);
std::vector<double> ts;
if (dec.timestamps(f, 0, ts)) {
all.insert(all.end(), ts.begin(), ts.end());
}
}
ASSERT_GT(all.size(), 8u);
for (size_t i = 1; i < all.size(); i++) {
EXPECT_GT(all[i], all[i - 1]) << "packets overlapped at " << i;
}
}
TEST(FrameDecoder, ResetForgetsPerSignalHistory) {
FrameDecoder dec;
dec.setSignals({burst("Raw", kTimePacket, 0.0, 4, kNoTimeSignal)});
FrameBuilder fb;
fb.addSignal({1.0, 2.0, 3.0, 4.0});
const FrameView& f = fb.build(0, 5.0);
dec.beginFrame(f);
std::vector<double> ts;
EXPECT_FALSE(dec.timestamps(f, 0, ts));
const FrameView& f2 = fb.build(0, 5.1);
dec.beginFrame(f2);
EXPECT_TRUE(dec.timestamps(f2, 0, ts));
dec.reset();
const FrameView& f3 = fb.build(0, 5.2);
dec.beginFrame(f3);
EXPECT_FALSE(dec.timestamps(f3, 0, ts))
<< "after reset the next packet is again the first one";
}