// SPDX-License-Identifier: MIT // SPDX-FileCopyrightText: 2026 Martino Ferrari #pragma once #include "channel.hpp" #include "config.hpp" #include "event_series.hpp" #include "trigger.hpp" #include #include #include #include #include #include #include #include "math_channel.hpp" #include namespace udpscope { /// One streamed signal, as the CONFIG packet described it. struct SignalInfo { std::string name; /// True for the entries the model *synthesised* for a math channel: the stream never sends them, /// the producer fills their samples by evaluating the graph. The flag is what lets a CONFIG /// update tell them from the stream's own signals and put them back afterwards. bool math = false; std::string unit; std::uint8_t type_code = 0; std::uint8_t quant_type = 0; std::uint32_t rows = 1; std::uint32_t cols = 1; double range_min = 0.0; double range_max = 1.0; std::uint8_t time_mode = 0; double sampling_rate = 0.0; std::uint32_t time_signal_idx = 0xFFFFFFFFu; std::uint32_t elements() const noexcept { const std::uint32_t n = rows * cols; return n == 0 ? 1u : n; } }; /// One plottable signal element: a signal with a single element, or one element /// of a vector/matrix signal. The unit of enable/disable is the *signal*; the /// unit of acquisition and plotting is the channel. struct ChannelInfo { std::size_t signal = 0; bool enabled = false; ///< whether its signal is enabled double gain = 1.0; ///< calibration, applied as sample * gain + offset double offset = 0.0; std::string name; ///< the signal's name /// True for a channel the producer *derives* rather than reads. The ingest must skip it: it has /// no packet signal, so reading one gives zero -- and pushing that zero into a ring whose time /// ring advances once per slot is what made a math channel alternate between its value and 0, and /// then drift off the time ring entirely once the buffer was full. bool math = false; std::string unit; ///< the signal's unit, or a math channel's own /// Samples of this channel per DATA packet (`elements()` of the signal). A /// UDPStreamer signal is one channel: its columns are consecutive samples of the /// same signal, not separate channels. std::uint32_t elements = 1; /// Sample interval of this channel's own cadence, in seconds. 0 when unknown. double dt = 0.0; }; /// How one buffer's storage is divided between the shared time base and the /// channels (spec item 2: "split in N+1", the +1 being the time base). /// /// The split is by *time span*, not by bytes: every channel's ring and the time /// ring cover the same amount of stream time, which is what makes a shared time /// base meaningful. When no channel is decimated (the common case) the two rules /// coincide and every slice is the same size. struct Partition { u64 time_capacity = 0; ///< time-base slots retained /// Base ticks per stored slot: the decimation factor the requested window forced. 1 /// until the window needs more samples than the budget can hold. The ingest path /// *averages* each group of this many samples rather than keeping one of them (see /// `Acquisition::on_data()`), so this is a sample rate reduction with an anti-alias /// filter in front of it, not a stride. u64 time_stride = 1; std::vector stride; ///< time slots per channel sample (>= 1) std::vector value_capacity; ///< samples retained per channel double base_dt = 1.0; ///< seconds per base tick double span_s = 0.0; ///< stream seconds one buffer covers /// Seconds between two time-base slots (the sample interval of every channel). double slot_dt() const noexcept { return base_dt * static_cast(time_stride); } /// The rate samples are actually stored at, in samples per second. double stored_rate() const noexcept { const double dt = slot_dt(); return dt > 0.0 ? 1.0 / dt : 0.0; } /// The anti-alias filter's corner, in Hz: half the stored rate, which is the new /// Nyquist. The boxcar the ingest path averages with has its first null at the stored /// rate itself, so everything that would fold back is attenuated. double antialias_hz() const noexcept { return 0.5 * stored_rate(); } }; /// Plans a partition for `channel_dt` cadences within `buffer_bytes`, covering at least /// `target_span_s` seconds of stream -- decimating (with the ingest filter) by whatever /// power of two that takes. Throws std::runtime_error when the budget cannot hold even /// one sample per channel. Partition plan_partition(const std::vector& channel_dt, double buffer_bytes, double target_span_s); /// One of the two identical sample buffers: a shared time-base ring plus one /// value ring and one `ChannelBuffer` per channel. class BufferStore { public: BufferStore(const Partition& part, f64 span_s); /// Re-bases every channel's x (see `ChannelBuffer::set_epoch`). Used to make the /// axis read seconds from the trigger event. void set_epoch(f64 epoch) noexcept; /// The shared clock, in **f64**. The values are f32 and the clock is not: a stamp is /// `stream_time - epoch` and the epoch is fixed at the first sample ever seen, so the /// magnitude grows without bound over a run's life while the interval that has to be /// resolved stays the sample period. f32 has 24 bits of mantissa, so its ULP at 32 s is /// 3.8 us -- four consecutive samples of a 1 MS/s stream then share one stamp, and at /// 600 s uptime 61 of them do. A value's precision is a property of the signal and 24 /// bits is plenty; a *timestamp's* is a property of how long the scope has been running. plot::data::RingBuffer& time() noexcept { return *time_; } const plot::data::RingBuffer& time() const noexcept { return *time_; } plot::data::RingBuffer& values(std::size_t channel) { return *values_[channel]; } ChannelBuffer& channel(std::size_t channel) { return *channels_[channel]; } /// Shared handle to a channel view: the caller keeps the storage alive even if /// the model rebuilds. std::shared_ptr shared_channel(std::size_t channel) { return channels_[channel]; } const Partition& partition() const noexcept { return part_; } /// Consumer: publishes every channel's new samples. void acquire_all(); private: Partition part_; std::shared_ptr> time_; std::vector>> values_; std::vector> channels_; }; /// Live connection and counter state, readable from the UI thread. struct StreamStats { std::atomic packets{0}; std::atomic frames{0}; std::atomic gaps{0}; std::atomic reconnects{0}; std::atomic connected{false}; std::atomic have_config{false}; }; /// What the plot is showing, which is one decision made once per frame: the bind, the /// export and the status wording all have to agree, and the producer can latch between /// any two of them. /// /// `Capturing` is the live-updating capture: an edge has been accepted (so the pre /// window was full) and the post window is still filling, and the plot follows the live /// ring on an edge-relative axis until the snapshot lands. It is a deliberate departure /// from SPECS:37 ("not refreshed till next event") -- see README. enum class DisplayMode { Live, ///< no trigger: the live ring, stream clock Capturing, ///< an edge was accepted: the live ring, edge-relative clock Held, ///< the frozen snapshot of the last event Waiting, ///< armed, nothing captured yet: blank /// The user paused: the acquisition thread keeps writing the live ring, and the plot draws /// the view that was frozen when the pause began -- a snapshot, exactly as `Held` is, so /// resuming costs no data. Paused, }; /// The application's data model: the stream description, the two sample buffers, /// the trigger configuration and the counters. Structural changes (a CONFIG /// update, an enable/disable, a calibration) rebuild the buffers, so all mutating /// access is serialised on one mutex; the acquisition thread takes it once per /// DATA frame. class ScopeModel { public: explicit ScopeModel(const Config& cfg); const Config& config() const noexcept { return cfg_; } // --- stream description --------------------------------------------------- /// Takes a new CONFIG: rebuilds the signal and channel lists and the two /// buffers. Clears the trigger's channel selection if the stream changed. void set_signals(std::vector signals); /// The user's math channels, as configured. Read it under the lock, like the other accessors /// here -- the UI copies what it needs while it holds it. std::vector math_channels() const { return math_; } /// Replaces the math channels. A *structural* change, exactly like a CONFIG or an /// enable/disable: the enabled ones are re-appended to the signal list, their channels are /// rebuilt and their runtimes re-prepared for the new stored rate. **Takes the lock itself** -- /// do not call it holding `mutex()`, which is not recursive. void set_math_channels(std::vector channels); /// One stored sample's worth of derived values, called by the producer right after it has pushed /// the streamed channels. `sample[c]` is channel `c`'s value for this slot; the math channels are /// evaluated from those, in order, so a math channel may use another one. void push_math_sample(const std::vector& sample); /// What the derived channels cost the *producer*: every stored slot that went through /// `push_math_sample()`, and the nanoseconds taken by the `math_timed()` of them that were /// measured. The producer owns the budget between two DATA packets, so this cost is invisible in /// the frame mean and surfaces only as dropped packets -- which is why it is counted rather than /// inferred. It is *sampled*, so that the instrumentation cannot itself cause what it is looking /// for: read `math_cost_ns() / math_timed()` as the per-slot figure, scaled by `math_slots()`. std::uint64_t math_slots() const noexcept { return math_slots_.load(std::memory_order_relaxed); } std::uint64_t math_timed() const noexcept { return math_timed_.load(std::memory_order_relaxed); } std::uint64_t math_cost_ns() const noexcept { return math_cost_ns_.load(std::memory_order_relaxed); } const std::vector& signals() const noexcept { return signals_; } const std::vector& channels() const noexcept { return channels_; } bool signal_enabled(std::size_t signal) const; /// True when the signal is the stream's time base, i.e. when some other signal /// declares it via `time_signal_idx`. A signal used as a time base builds the x /// axis and is not a plottable channel. static bool is_time_signal(const std::vector& signals, std::size_t signal); void set_signal_enabled(std::size_t signal, bool on); /// Stores the calibration for a signal's *name* and applies it to that signal's /// live channels. Keyed by name, not index: a CONFIG renumbers and an /// enable/disable rebuilds the channel list, and `rebuild_channels()` re-applies /// from here -- so a calibration set before a signal was enabled, or surviving a /// CONFIG, is not silently dropped. void set_calibration(std::size_t signal, double gain, double offset); std::size_t channel_count() const noexcept { return channels_.size(); } const Partition& partition() const noexcept { return part_; } /// The channel's sample interval (0 when the stream did not say). double channel_dt(std::size_t channel) const; // --- buffers -------------------------------------------------------------- /// The one sample store: the producer always writes it, and the plot draws from /// it whenever it is not showing a frozen event. Always-filling is what /// guarantees a full pre-trigger history on the *first* latch after the trigger /// is armed. BufferStore& live(); /// Forgets the live buffer's history: every ring is emptied and every channel's /// published window is dropped, so the two stay index-aligned. Used when the /// acquisition has to change time source (see `Acquisition::build_stamps`): the /// stamps already recorded are on the old timeline, and a ring holding two /// timelines is not a time base at all -- the x axis, the crosshair and an export /// all read it by index. /// /// Takes `mutex()` itself: call it from the GTK thread. The acquisition thread /// calls `reset_live_history_locked()` instead, already holding the lock (the /// same split `publish_event()` follows). void reset_live_history(); /// The body of the above for a caller that already holds `mutex()`. void reset_live_history_locked(); /// Builds the captured event from the slice of the live buffer covering /// `[edge - pre_s, edge + post_s]` in stream seconds, and publishes it atomically. /// The stamps are re-based so the edge sits at x = 0. The plot binds to this /// immutable snapshot -- it is the "frozen buffer" -- while the live ring keeps /// filling behind it. /// /// The cut is by time, not by a sample count: the window the trigger panel asks /// for is in seconds and the time ring stores seconds, so any count in between /// needs a cadence estimate -- which is exactly what used to size the capture /// wrong. void publish_event(double pre_s, double post_s, double edge); /// The live buffer's absolute index range covering `[lo_s, hi_s]` in stream seconds, /// clamped to what the time ring and *every* value ring still hold. One range for the /// time ring and all the channels, which is what keeps a cut index-aligned by /// construction; the bounds are compared in the ring's own type, which is now the /// `f64` they arrive in, so a bound that lands on a sample includes it exactly. /// /// Caller holds `mutex()` (the acquisition thread already does when it latches). void live_window_locked(double lo_s, double hi_s, u64& first, u64& last); /// The frozen view taken when the user paused, or null while running. A pause is a /// snapshot of the *visible window* in exactly the sense a trigger event is -- an immutable /// series the plot binds to -- and both are anchored the same way: the stamps are the /// coordinates the axis was already reading, with `x_offset` recording which instant that /// x = 0 was. So the axis does not move when the pause begins. std::shared_ptr pause_view() const noexcept { return pause_view_.load(std::memory_order_acquire); } /// Snapshots the live ring over `[lo_s, hi_s]` (stream seconds, i.e. the display window put /// back on the stream's clock) as the pause view, rebased onto `x_offset` so its stamps are /// the x the axis is already showing. The acquisition thread keeps writing, so this is a /// copy: what is frozen is the picture, not the data. void publish_pause_view(double lo_s, double hi_s, double x_offset); void clear_pause_view() noexcept { pause_view_.store(nullptr, std::memory_order_release); } /// Whether the user paused acquisition. Lives here rather than on `Acquisition` because the /// bind, the measurement cut and the export all have to agree about what is on screen. bool paused() const noexcept { return paused_.load(std::memory_order_relaxed); } void set_paused(bool on) noexcept { paused_.store(on, std::memory_order_relaxed); } /// The snapshot the plot should draw from, if any: the pause's frozen view while acquisition /// is paused, else the trigger's held event. One accessor, so the bind, the measurements and /// the export cannot disagree about which picture is on screen. std::shared_ptr display_snapshot() const noexcept { if (paused()) return pause_view(); return display_mode() == DisplayMode::Held ? event() : nullptr; } /// The event on screen, or null before the first latch. std::shared_ptr event() const noexcept { return event_.load(std::memory_order_acquire); } /// How many snapshots `publish_event()` has produced. Not interchangeable with /// `trigger_latches()`: a Disarm during a capture publishes what the capture got to /// *without* an event having fired, so this counts "there is a new picture to frame" /// while the latch count keeps answering "how many events fired". std::uint64_t event_seq() const noexcept { return event_seq_.load(std::memory_order_acquire); } /// Publishes the live buffer's new samples to its `ChannelBuffer`s. void acquire_all(); // --- trigger -------------------------------------------------------------- /// The trigger configuration. Mutating it means holding `mutex()`: the producer /// reads it once per frame (and writes `armed` itself when a single-shot latch /// disarms), so an unguarded field write can be lost or read half-updated. The /// panel's callbacks take the lock for the fields they touch. TriggerConfig& trigger() noexcept { return trigger_; } const TriggerConfig& trigger() const noexcept { return trigger_; } /// Trigger mode on/off, as both the frame tick and the producer read it: the one /// trigger field the tick needs and the only one written from the UI, so it has /// an atomic mirror and a setter that keeps the copy under the lock in step. bool trigger_enabled() const noexcept { return trigger_enabled_.load(std::memory_order_relaxed); } void set_trigger_enabled(bool on); /// Whether the engine is mid-capture (an edge was seen -- so the pre window was /// full -- and the post-trigger window is still filling). This is what puts the /// display in `DisplayMode::Capturing`: the plot follows the live ring, on an axis /// pinned to the coming event's window, until the snapshot replaces it. /// /// Release/acquire, not relaxed: `capture_edge()` is written before this is set and /// has to be visible to a consumer that sees the flag. bool trigger_pending() const noexcept { return trigger_pending_.load(std::memory_order_acquire); } void set_trigger_pending(bool on) noexcept { trigger_pending_.store(on, std::memory_order_release); } /// The stream time of the edge of the capture `trigger_pending()` refers to. Written /// by the producer before it raises the flag, so a consumer that sees the flag sees /// this too; stale once the flag clears. f64 capture_edge() const noexcept { return capture_edge_.load(std::memory_order_relaxed); } void set_capture_edge(f64 e) noexcept { capture_edge_.store(e, std::memory_order_relaxed); } /// What the plot should bind to this frame. Read it *once* per frame and pass the /// answer around: calling it twice can straddle a latch, and then the bind and the /// axis disagree about which clock they are on. DisplayMode display_mode() const noexcept { // A pause is the most recent instruction, so it wins over the trigger's own state: what the // user last asked for is a still picture. if (paused()) return DisplayMode::Paused; if (!trigger_enabled()) return DisplayMode::Live; if (trigger_pending()) return DisplayMode::Capturing; return event() != nullptr ? DisplayMode::Held : DisplayMode::Waiting; } /// The offset the **live** channels report their x with: `display_x = stream_t + epoch`. /// Minus the newest stored sample's stamp while streaming, so the axis reads /// `[-window, 0]` and the newest data is at 0; minus the trigger instant while a capture /// is filling, so the axis reads seconds from the event. It is 0 while a *snapshot* is on /// screen (a held event, a paused view), because then the live channels are bound to /// nothing and the snapshot's own `x_offset` is what says which clock x = 0 is -- /// `stream_at_x0()` is the accessor that does not have to know which of the two it is. f64 display_epoch() const noexcept { return display_epoch_.load(std::memory_order_relaxed); } void set_display_epoch(f64 e) noexcept; /// The stream time of the display's x = 0, so `stream_t = display_x + stream_at_x0()` and /// `display_x = stream_t - stream_at_x0()` in **every** display mode. /// /// One accessor for the two places the offset can live: a snapshot carries its own /// (`EventSnapshot::x_offset`, frozen with the picture), and a live view carries it as /// minus the display epoch. Everything that has to cross between the two clocks goes /// through here -- the export's absolute column, and the pause handler putting a display /// window back on the ring to cut it -- rather than picking the branch itself and getting /// one of the four modes wrong. f64 stream_at_x0() const noexcept { if (std::shared_ptr snap = display_snapshot()) return snap->x_offset; return -display_epoch(); } // --- stream state --------------------------------------------------------- StreamStats& stats() noexcept { return stats_; } const StreamStats& stats() const noexcept { return stats_; } /// Seconds since the epoch of the streamed time, or 0 before the first frame. f64 epoch() const noexcept { return epoch_; } void set_epoch_if_unset(f64 e) noexcept; /// Bumped whenever the signal or channel lists change, so the UI knows to /// rebuild the panels it derives from them. unsigned generation() const noexcept { return generation_.load(std::memory_order_acquire); } /// The length of the event the trigger latches, in seconds: the time window the /// toolbar selects. The pre-trigger percentage is a share of *this*, not of the /// whole ring, which is what makes the control mean something on a long buffer. void set_event_span(double seconds) noexcept { event_span_.store(seconds > 0.0 ? seconds : 1.0, std::memory_order_relaxed); } double event_span() const noexcept { return event_span_.load(std::memory_order_relaxed); } /// The window the buffer has to cover, which is the display window. Re-plans the /// partition -- and so drops the history -- only when the decimation factor it implies /// actually changes, so walking the window inside one decade costs nothing. Takes /// `mutex()` itself; do not call it holding the lock. /// Returns true when the buffer was rebuilt. bool set_target_span(double seconds); double target_span() const noexcept { return target_span_.load(std::memory_order_relaxed); } /// Trigger state for the UI, published by the acquisition thread. Atomics /// rather than fields of `TriggerConfig` because the producer writes the arm /// state when a single-shot trigger disarms itself, and the frame tick reads it /// without holding the model mutex. std::atomic& trigger_armed() noexcept { return trigger_armed_; } /// Incremented every time an event is latched, so the UI can tell "waiting for /// an event" from "an event is on screen". std::atomic& trigger_latches() noexcept { return trigger_latches_; } /// The stream time of the edge behind the event on screen, and whether there is /// one. Written by the acquisition thread on a latch and cleared by the UI when /// the event stops meaning anything (arming again, or leaving trigger mode). std::atomic& trigger_time() noexcept { return trigger_time_; } std::atomic& trigger_time_valid() noexcept { return trigger_time_valid_; } /// Serialises a structural change against the acquisition thread. std::mutex& mutex() noexcept { return mutex_; } /// Rebuilds buffers from the current channel list. Caller holds `mutex()`. void rebuild_locked(); private: void rebuild_channels(); /// Re-appends one synthetic signal per math channel, after the stream's own. Called under the /// lock by both `set_signals()` (a CONFIG) and `set_math_channels()` (an edit). void append_math_signals_locked(); /// Rebuilds the math runtimes for the *current* plan: the biquad coefficients follow the stored /// sample interval, and their state is reset with them. Called at the end of `rebuild_locked()`, /// so a window change that moves the decimation re-prepares the filters exactly as an edit does. void prepare_math_locked(); Config cfg_; std::vector signals_; std::vector channels_; // --- math channels -------------------------------------------------------- std::vector math_; std::vector math_rt_; /// The channel index each math channel was given by `rebuild_channels()`, -1 while it has none /// (disabled, or before the first rebuild). std::vector math_channel_; /// The names of the synthetic signal entries, so they can be lifted out of `signals_`/`enabled_` /// before the stream's own are re-appended -- a CONFIG must not leave a stale derived channel /// behind, and must not drop a live one either. std::vector math_signal_names_; /// The producer's scratch for one slot's channel values, derived ones appended as they are /// computed. A member rather than a local because `push_math_sample()` runs once per *stored /// slot* -- up to the wire rate -- and a local `std::vector` is a heap allocation and free per /// slot. `assign()` reuses the capacity, so after the first slot it is a memcpy. std::vector math_values_; /// The counters' own copies, written and read by the producer and nobody else, so each publish /// below is a relaxed **store** of a value already in hand rather than a `fetch_add`. The /// read-modify-write is a locked instruction and this is the per-slot path: one `lock xadd` costs /// about what a two-node graph's whole arithmetic does, so counting the slots cost more than the /// work it was counting. std::uint64_t math_slot_count_ = 0; std::uint64_t math_timed_count_ = 0; std::uint64_t math_cost_count_ = 0; /// When the slot being timed began. std::uint64_t math_cost_t0_ = 0; std::atomic math_slots_{0}; std::atomic math_timed_{0}; std::atomic math_cost_ns_{0}; std::vector enabled_; Partition part_; std::unique_ptr store_; TriggerConfig trigger_{}; StreamStats stats_{}; f64 epoch_ = 0.0; std::atomic event_span_{1.0}; /// The window the buffer is planned to cover. Read by `rebuild_locked()`, which a /// CONFIG can reach from the producer thread, so it is an atomic rather than a plain /// field. std::atomic target_span_{0.0}; std::atomic trigger_pending_{false}; std::atomic capture_edge_{0.0}; std::atomic> event_; ///< the event on screen std::atomic> pause_view_; ///< the view frozen at pause std::atomic paused_{false}; /// Per-signal-name calibration (see `set_calibration`). std::unordered_map> calibration_; std::atomic event_seq_{0}; std::atomic display_epoch_{0.0}; std::atomic trigger_armed_{false}; std::atomic trigger_enabled_{false}; std::atomic trigger_latches_{0}; std::atomic trigger_time_{0.0}; std::atomic trigger_time_valid_{false}; std::atomic generation_{0}; std::mutex mutex_; }; } // namespace udpscope