martino/udpscope
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
Martino Ferrari committed 2026-10-01 15:42:48 +02:00
commit 8059b90de7
29 files changed
+7893
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/acquisition.cpp
+476
View File
@@ -0,0 +1,476 @@
// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari
#include "acquisition.hpp"
#include <udps_client.h>
#include <algorithm>
#include <cmath>
#include <cstring>
#include <stdexcept>
#include <string>
namespace udpscope {
namespace {
SignalInfo to_info(const udps_signal_t& s) {
SignalInfo info;
info.name = s.name;
info.unit = s.unit;
info.type_code = s.type_code;
info.quant_type = s.quant_type;
info.rows = s.num_rows;
info.cols = s.num_cols;
info.range_min = s.range_min;
info.range_max = s.range_max;
info.time_mode = s.time_mode;
info.sampling_rate = s.sampling_rate;
info.time_signal_idx = s.time_signal_idx;
return info;
}
} // namespace
Acquisition::Acquisition(ScopeModel& model, const Config& cfg) : model_(model), cfg_(cfg) {}
Acquisition::~Acquisition() { stop(); }
void Acquisition::start() {
udps_client_config_t c;
udps_client_config_init(&c);
c.server_addr = cfg_.server.c_str();
c.server_port = cfg_.port;
c.multicast_group = cfg_.multicast.empty() ? nullptr : cfg_.multicast.c_str();
c.interface_addr = cfg_.interface_addr.empty() ? nullptr : cfg_.interface_addr.c_str();
c.recv_buffer_bytes = static_cast<std::uint32_t>(cfg_.recv_buffer_bytes);
c.silence_timeout_s = 2.0;
c.reconnect_delay_s = 0.5;
c.keepalive_interval_s = 1.0;
client_ = udps_client_create(&c);
if (client_ == nullptr) {
throw std::runtime_error("could not create the UDPS client for " + cfg_.server + ":" +
std::to_string(cfg_.port));
}
udps_client_set_callbacks(client_, &Acquisition::config_trampoline, &Acquisition::data_trampoline,
&Acquisition::event_trampoline, this);
started_ = std::chrono::steady_clock::now();
running_.store(true, std::memory_order_release);
thread_ = std::thread([this] { run(); });
}
void Acquisition::stop() {
if (!running_.exchange(false, std::memory_order_acq_rel)) {
if (client_ != nullptr) {
udps_client_destroy(client_);
client_ = nullptr;
}
return;
}
if (thread_.joinable()) thread_.join();
if (client_ != nullptr) {
udps_client_destroy(client_);
client_ = nullptr;
}
}
void Acquisition::run() {
while (running_.load(std::memory_order_acquire)) {
udps_client_poll(client_, 50);
udps_stats_t s;
udps_client_stats(client_, &s);
StreamStats& out = model_.stats();
out.packets.store(s.packets_received, std::memory_order_relaxed);
out.frames.store(s.frames_delivered, std::memory_order_relaxed);
out.gaps.store(s.counter_gaps, std::memory_order_relaxed);
out.reconnects.store(s.reconnects, std::memory_order_relaxed);
out.connected.store(udps_client_is_connected(client_) != 0, std::memory_order_relaxed);
}
}
void Acquisition::on_config(const udps_signal_t* signals, std::uint32_t count,
std::uint8_t publish_mode) {
std::vector<SignalInfo> info;
info.reserve(count);
for (std::uint32_t i = 0; i < count; ++i) info.push_back(to_info(signals[i]));
// One line per CONFIG on stderr: what the stream says it is. This is the first
// thing to look at when a signal refuses to plot.
static const char* const kModes[] = {"strict", "accumulate", "decimate"};
std::fprintf(stderr, "[udpscope] CONFIG: %u signals, publish=%s\n", count,
publish_mode < 3 ? kModes[publish_mode] : "?");
for (const SignalInfo& sig : info) {
std::fprintf(stderr,
"[udpscope] '%s' %ux%u (%u elem) rate=%g Hz time=%s idx=%u unit='%s'\n",
sig.name.c_str(), sig.rows, sig.cols, sig.elements(), sig.sampling_rate,
sig.time_mode == UDPS_TIME_PACKET ? "packet"
: sig.time_mode == UDPS_TIME_FULL_ARRAY ? "full-array"
: sig.time_mode == UDPS_TIME_FIRST_SAMPLE ? "first-sample"
: "last-sample",
sig.time_signal_idx, sig.unit.c_str());
}
std::fflush(stderr);
model_.stats().have_config.store(true, std::memory_order_relaxed);
model_.set_signals(std::move(info));
}
void Acquisition::on_event(int event, const char* detail) {
(void)detail;
if (event == UDPS_EVENT_CONNECTED) {
model_.stats().connected.store(true, std::memory_order_relaxed);
} else if (event == UDPS_EVENT_DISCONNECTED) {
model_.stats().connected.store(false, std::memory_order_relaxed);
}
}
std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_signal) {
// The frame's span in base ticks: a scalar signal carries `num_samples` values
// (one per batched cycle), a block signal (rows*cols > 1) carries its whole
// element block. The largest of those is how many base ticks this frame spans --
// in `strict` mode `num_samples` is 1 while a 1x1000 block carries 1000 samples.
std::uint32_t n = frame->num_samples;
for (std::uint32_t i = 0; i < frame->num_signals; ++i) {
n = std::max(n, udps_signal_num_elements(&frame->signals[i]));
}
stamps_.assign(n, 0.0);
double period = model_.partition().base_dt;
if (!(period > 0.0)) period = 1.0e-3;
bool used_stream_time = false;
if (time_signal >= 0 && time_signal < static_cast<int>(frame->num_signals)) {
const udps_signal_t& ts = frame->signals[time_signal];
// How the client indexes a signal's values depends on its shape
// (`udps_frame_value`): a scalar carries one value per batched cycle, so the
// stamp of sample k is (sample=k, elem=0); a multi-element signal carries one
// *sample* of `elements` values per packet, so the stamp of sample k is
// (sample=0, elem=k). Passing the scalar form for both is why FULL_ARRAY time
// signals -- the only mode that carries a stamp per element -- read as
// element 0 for every sample and were then judged degenerate.
const bool scalar_time = udps_signal_num_elements(&ts) == 1;
auto value_of = [&](std::uint32_t k) {
return scalar_time
? udps_frame_value(frame, static_cast<std::uint32_t>(time_signal), k, 0)
: udps_frame_value(frame, static_cast<std::uint32_t>(time_signal), 0, k);
};
switch (ts.time_mode) {
case UDPS_TIME_FULL_ARRAY:
for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = value_of(k);
used_stream_time = true;
break;
case UDPS_TIME_FIRST_SAMPLE: {
const double anchor = value_of(0);
const double step = n > 1 ? (value_of(n - 1) - anchor) / double(n - 1) : period;
for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = anchor + double(k) * step;
used_stream_time = true;
break;
}
case UDPS_TIME_LAST_SAMPLE: {
const double anchor = value_of(n > 0 ? n - 1 : 0);
for (std::uint32_t k = 0; k < n; ++k)
stamps_[k] = anchor - double(n - 1 - k) * period;
used_stream_time = true;
break;
}
default:
break;
}
}
if (used_stream_time && n > 0) {
// A time signal that does not actually vary -- a vector time signal, whose
// values the client reads one element at a time, or a producer that leaves it
// zero -- would put every sample at one x and collapse the axis to a point.
// The arrival estimate is a worse clock but a usable one; a degenerate one is
// not a time base at all.
bool usable = std::isfinite(stamps_[0]);
double lo = stamps_[0];
double hi = stamps_[0];
for (std::uint32_t k = 1; usable && k < n; ++k) {
if (!std::isfinite(stamps_[k])) {
usable = false;
break;
}
lo = std::min(lo, stamps_[k]);
hi = std::max(hi, stamps_[k]);
}
if (usable && hi > lo) {
// The stamps must also ADVANCE across packets. A producer that restarts its
// stamps every packet (or a signal mislabelled as time) keeps handing back
// the same values; the max-clamp would then pin every sample at one x, and
// the trigger's hold-off clock with it -- the engine waited forever and no
// second event ever fired.
if (have_stream_anchor_ && !(stamps_[n - 1] > last_stream_anchor_)) {
usable = false;
} else {
last_stream_anchor_ = stamps_[n - 1];
have_stream_anchor_ = true;
}
}
if (!usable || !(hi > lo)) {
used_stream_time = false; // no span to time with; fall back to arrival
static bool warned = false;
if (!warned) {
warned = true;
std::fprintf(stderr,
"[udpscope] warning: the stream's time signal carries no usable span "
"(%g..%g) or does not advance across packets; timing samples by "
"arrival instead. Check the time signal's mode and index in the "
"CONFIG dump above.\n",
lo, hi);
}
}
}
if (!used_stream_time) {
// Arrival-anchored: the packet is assumed to have arrived as its last element
// was produced, and the earlier ones are dated backwards. A gap (`lost`)
// shows up as extra elapsed time between arrivals, which is exactly the
// spacing the samples had.
const double end = frame->recv_time;
if (!have_arrival_epoch_) {
arrival_epoch_ = end;
have_arrival_epoch_ = true;
}
const double base = end - arrival_epoch_;
double step = period;
if (have_recv_ && n > 0) {
const double elapsed = end - last_recv_;
const double cycles = double(n) * double(frame->lost + 1);
if (elapsed > 0.0 && cycles > 0.0) step = elapsed / cycles;
}
for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = base - double(n - 1 - k) * step;
}
// A stream-time signal may be declared in a time unit rather than seconds
// ('ns' is the common one): scale it to seconds so the x axis is in seconds
// either way. The scale applies ONLY to stream time: the arrival fallback
// already produces seconds off the socket clock -- scaling those by the time
// signal's unit shrank the whole time base by 1e9 on 'ns' streams, which
// froze the trigger's hold-off clock and pinned the axis at ~0.
if (used_stream_time && time_signal >= 0 &&
time_signal < static_cast<int>(model_.signals().size())) {
double unit_scale = 1.0;
const std::string& unit = model_.signals()[static_cast<std::size_t>(time_signal)].unit;
if (unit == "ns") unit_scale = 1e-9;
else if (unit == "us" || unit == "\u00b5s" || unit == "\xce\xbc" "s") unit_scale = 1e-6;
else if (unit == "ms") unit_scale = 1e-3;
if (unit_scale != 1.0) {
for (std::uint32_t k = 0; k < n; ++k) stamps_[k] *= unit_scale;
}
}
last_recv_ = frame->recv_time;
have_recv_ = true;
return n;
}
void Acquisition::on_data(const udps_frame_t* frame) {
if (paused_.load(std::memory_order_relaxed)) return;
std::lock_guard<std::mutex> lock(model_.mutex());
if (!running_.load(std::memory_order_relaxed)) return;
const Partition& part = model_.partition();
const std::vector<ChannelInfo>& channels = model_.channels();
if (part.time_capacity == 0 || channels.empty()) return;
// The time reference, in order of preference: the signal the first channel's
// CONFIG entry names via `time_signal_idx`, or -- when the stream instead
// carries its time base as an explicit full-array signal -- that signal.
int time_signal = -1;
{
const SignalInfo& sig = model_.signals()[channels[0].signal];
if (sig.time_signal_idx != 0xFFFFFFFFu && sig.time_signal_idx < model_.signals().size()) {
time_signal = static_cast<int>(sig.time_signal_idx);
}
}
if (time_signal < 0) {
// No channel names a time signal: look for the standalone timestamp signal
// (full-array, more than one element, no declared rate).
for (std::size_t i = 0; i < model_.signals().size(); ++i) {
const SignalInfo& sig = model_.signals()[i];
if (sig.time_mode == UDPS_TIME_FULL_ARRAY && sig.elements() > 1 &&
sig.sampling_rate == 0.0) {
time_signal = static_cast<int>(i);
break;
}
}
}
const std::uint32_t n = build_stamps(frame, time_signal);
if (n == 0) return;
if (model_.epoch() == 0.0) {
model_.set_epoch_if_unset(stamps_[0]);
std::fprintf(stderr, "[udpscope] time base: epoch=%.3f, first frame recv=%.3f, n=%u\n",
model_.epoch(), frame->recv_time, n);
std::fflush(stderr);
}
const double epoch = model_.epoch();
TriggerConfig& tc = model_.trigger();
// One ring, always filling, whatever the trigger is doing: the pre-trigger
// history has to already be there when an edge arrives, and the first latch after
// arming would otherwise open on however little had accumulated since. What
// freezes in trigger mode is the *view*, not acquisition -- the latch copies the
// event window out into an immutable `EventSnapshot`.
BufferStore& store = model_.live();
const u64 g = part.time_stride;
// Calibration is read per frame; it only changes under the same lock.
TriggerEngine& engine = engine_;
const int trig_channel = tc.channel;
// The post-trigger window straight through in seconds. There used to be a measured,
// smoothed samples-per-second estimate here, converting the window into a sample
// count for the engine; the estimate is what sized captures wrong. Its floor
// (`inst >= 1.0`) turned any stream whose time signal advances in large steps into
// 1 S/s, the count came out as 1, and the engine latched on the edge sample itself
// -- an event with no post-trigger data, which is exactly the "post window is
// empty" defect. The engine is fed stream time per sample anyway, so it can count
// the window itself and no conversion has to exist.
//
// The oldest buffered stamp goes with it, so the engine can refuse an edge the
// history does not reach back far enough behind. Read from the ring rather than
// timed from the arm: the ring fills with the trigger off too, so the pre window is
// usually already there and a timer would delay the first event for nothing.
const double event_span = model_.event_span();
const double post_want = tc.post_span_s(event_span);
const double oldest_s =
store.time().size() > 0 ? double(store.time()[0]) : last_stamp_;
engine.configure(tc, post_want, event_span - post_want, oldest_s);
// The engine's armed state follows the panel's, both ways: without the disarm
// half, Disarm left the engine firing (its own `armed_` was only ever set), and
// re-entering trigger mode kept the stale prev/edge/hold-off state.
if (tc.enabled && tc.armed) {
if (!engine.armed()) {
engine.arm(last_stamp_);
model_.trigger_armed().store(true, std::memory_order_relaxed);
}
} else if (engine.armed()) {
engine.disarm();
model_.trigger_armed().store(false, std::memory_order_relaxed);
}
double last_stamp_used = last_stamp_;
// The latch is handled on the sample it happens on, and with the edge the engine
// latched *on* -- after that sample's channels have all been pushed (see the loop).
//
// It used to set a flag and publish once after the loop, reading the edge back from
// the engine. But the loop keeps feeding the engine after a latch, so any further
// edge in the rest of the packet -- the hold-off is 10 ms and a packet can be a long
// block of samples -- overwrote that edge before it was read. The capture was then
// cut around an edge whose post window had not been collected yet, so the event came
// out all pre-trigger with the marker and x = 0 at the right-hand end of the data.
// Publishing here also means the ring provably holds the post window: the latch
// condition is that stream time has passed `edge + post`, and every sample up to
// this one has already been pushed.
const double span = std::max(1.0e-9, model_.event_span());
const double post_s = tc.post_span_s(span);
const auto latch = [&](double edge) {
model_.publish_event(span - post_s, post_s, edge);
model_.trigger_latches().fetch_add(1, std::memory_order_relaxed);
model_.trigger_time().store(edge, std::memory_order_relaxed);
model_.trigger_time_valid().store(true, std::memory_order_relaxed);
// The captured window is reported from the snapshot itself, not from the request:
// `want` versus `got` is what distinguishes "the engine did not wait" from "the
// cut lost the data", which is otherwise guesswork from the picture alone.
auto snap = model_.event();
std::fprintf(stderr,
"[udpscope] trigger: edge at %.3f s, latched at %.3f s of stream time "
"(want [-%.4f..%.4f] got [-%.4f..%.4f] n=%llu, frame recv=%.3f, "
"ticks_so_far=%llu, wall=%.3f s)\n",
edge, last_stamp_, span - post_s, post_s, snap ? snap->pre_s : 0.0,
snap ? snap->post_s : 0.0,
static_cast<unsigned long long>(snap ? snap->stamps.size() : 0),
frame->recv_time, static_cast<unsigned long long>(tick_),
std::chrono::duration<double>(std::chrono::steady_clock::now() - started_)
.count());
std::fflush(stderr);
if (tc.mode == TriggerMode::Single) tc.armed = false;
};
for (std::uint32_t k = 0; k < n; ++k) {
const u64 tick = tick_++;
const double rel = std::max(stamps_[k] - epoch, last_stamp_);
last_stamp_used = rel;
last_stamp_ = rel;
if (tick % g == g - 1) store.time().push(static_cast<f32>(rel));
// The latch belongs to this sample, but it cannot be published from inside the
// channel loop: `publish_event()` clamps the slice to every value ring's
// `produced()`, and the channels *after* the trigger channel have not pushed this
// sample yet, so the newest sample -- the one the post window was waiting for --
// would be clamped away. One sample is invisible in a 20 000-sample capture (it is
// the `n=19999` and the `0.9999` instead of `1.0000` in the probe's output), and at
// one sample per second it is the entire post side: `got [-0.9997..0.0000]`, which
// reads exactly like the "post window is empty" defect.
bool latch_here = false;
double latch_edge = 0.0;
for (std::size_t i = 0; i < channels.size(); ++i) {
const u64 period = g * part.stride[i];
if (tick % period != period - 1) continue;
const ChannelInfo& ch = channels[i];
// A signal's columns are consecutive samples of the same channel: a scalar
// (one element) carries one value per batched cycle, a block signal carries
// the cycle's sample at element `k`. Either way element 0 is the channel.
const double raw =
ch.elements > 1
? udps_frame_value(frame, static_cast<std::uint32_t>(ch.signal), 0,
std::min<std::uint32_t>(k, ch.elements - 1))
: udps_frame_value(frame, static_cast<std::uint32_t>(ch.signal), k, 0);
const f32 value = static_cast<f32>(raw * ch.gain + ch.offset);
store.values(i).push(value);
if (static_cast<int>(i) == trig_channel) {
if (engine.process(value, rel)) {
latch_here = true;
latch_edge = engine.last_trigger_time();
}
}
}
if (latch_here) latch(latch_edge);
}
last_stamp_ = last_stamp_used;
model_.trigger_armed().store(engine.armed(), std::memory_order_relaxed);
model_.set_trigger_pending(engine.pending());
static std::int64_t next_engine_log = 0;
const std::int64_t now_e = static_cast<std::int64_t>(frame->recv_time * 1.0e6);
if (now_e >= next_engine_log) {
next_engine_log = now_e + 1000000;
// `elapsed/post_want` is the capture's progress in the only unit that matters now
// (stream seconds); `accum` is how many samples that window has collected, which
// is what says whether a capture is thin because the window is short or because
// the stream is slow.
std::fprintf(stderr,
"[udpscope] trig engine: ch=%d armed=%d pending=%d elapsed=%.4f/%.4f s "
"accum=%llu history=%.4f/%.4f s pre_pct=%.0f v=%.4g now_t=%.3f "
"holdoff_until=%.3f\n",
trig_channel, int(engine.armed()), int(engine.pending()),
engine.post_elapsed(), engine.post_seconds(),
static_cast<unsigned long long>(engine.post_accum()), engine.history(),
engine.pre_seconds(), tc.pretrigger_pct, engine.last_value(),
engine.last_t(), engine.holdoff_until());
std::fflush(stderr);
}
}
// Static trampolines: the client's C ABI cannot call a member function directly.
void Acquisition::config_trampoline(const udps_signal_t* signals, std::uint32_t count,
std::uint8_t publish_mode, void* user) {
static_cast<Acquisition*>(user)->on_config(signals, count, publish_mode);
}
void Acquisition::data_trampoline(const udps_frame_t* frame, void* user) {
static_cast<Acquisition*>(user)->on_data(frame);
}
void Acquisition::event_trampoline(udps_event_t event, const char* detail, void* user) {
static_cast<Acquisition*>(user)->on_event(static_cast<int>(event), detail);
}
} // namespace udpscope