udpscope/src/model.hpp

// 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 <plot/data/ring_buffer.hpp>

#include <atomic>
#include <unordered_map>
#include <cstdint>
#include <memory>
#include <mutex>
#include <string>

#include "math_channel.hpp"
#include <vector>

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<u64> stride;             ///< time slots per channel sample (>= 1)
  std::vector<u64> 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

  /// True when a second f64 ring parallel to the clock was planned and paid for in the
  /// byte budget. Relative mode only -- in stream mode `time()` already *is* the absolute
  /// clock and a copy would cost ~22% of the user's `--buffer-mb` for nothing.
  bool abs_time = false;

  /// Seconds between two time-base slots (the sample interval of every channel).
  double slot_dt() const noexcept { return base_dt * static_cast<double>(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. `with_abs_time` adds a second f64 per time slot to the budget
/// and sets `Partition::abs_time`.
Partition plan_partition(const std::vector<double>& channel_dt, double buffer_bytes,
                         double target_span_s, bool with_abs_time = false);

/// 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<f64>& time() noexcept { return *time_; }
  const plot::data::RingBuffer<f64>& time() const noexcept { return *time_; }
  /// The stream's **absolute** clock, parallel to `time()` and on the same slots, or
  /// `nullptr` when the store was planned without it.
  ///
  /// There is exactly one time ring per store, shared by every channel, and relative mode
  /// fills it with counter-derived stamps -- so without this ring, selecting relative mode
  /// would silently drop absolute time from an export, the one output `todo.md` item 2 says
  /// must keep it. Allocated in relative mode only; read only by the export path.
  plot::data::RingBuffer<f64>* abs_time() noexcept { return abs_time_.get(); }
  const plot::data::RingBuffer<f64>* abs_time() const noexcept { return abs_time_.get(); }
  plot::data::RingBuffer<f32>& 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<ChannelBuffer> 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<plot::data::RingBuffer<f64>> time_;
  std::shared_ptr<plot::data::RingBuffer<f64>> abs_time_;
  std::vector<std::shared_ptr<plot::data::RingBuffer<f32>>> values_;
  std::vector<std::shared_ptr<ChannelBuffer>> channels_;
};

/// One reserved, invalidated span: samples the counter says exist and the stream has not
/// delivered. Its stored slots hold NaN, so a reader needs no special case -- the table is
/// what lets a *late* packet find the slots it belongs to, and what the band draws.
struct Hole {
  f64 x_begin = 0.0;          ///< stream seconds of the hole's first slot
  f64 x_end = 0.0;            ///< stream seconds of the hole's last slot
  /// Absolute **time-ring** slot of the first reserved slot, and the count of them. The
  /// time ring is the one clock every channel shares, so this is the only slot address
  /// that is unambiguous: per-channel value rings run at `time_stride * stride[i]`, which
  /// differs between channels. A consumer that needs a *channel's* slots derives them from
  /// `tick_begin` and that channel's stride rather than reusing these.
  u64 slot_begin = 0;
  u64 slot_count = 0;
  u64 tick_begin = 0;         ///< absolute base tick of the hole's first raw sample
  std::uint32_t counter_first = 0;  ///< counter of the first missing packet
  std::uint32_t packets = 0;  ///< packets the hole is worth
  /// How many of those packets have since arrived late and been written in. The band's two tints
  /// are this field against `packets`: a hole still missing data is shaded solidly and a fully
  /// repaired one faintly, so a reader can tell "there is no data here" from "there was none when
  /// this window was first drawn".
  ///
  /// Counted in *packets* and not in slots, because a packet is the unit the reservation was made
  /// in: the hole is worth exactly `packets` of them, each late arrival delivers exactly one, and
  /// the client refuses a duplicate -- so this count is exact. A slot count would not be: the
  /// channels have different strides, so "how many slots did this hole want" has one answer per
  /// channel and the repaired test would have to pick one.
  std::uint32_t filled_packets = 0;

  /// Every packet this hole was reserved for has arrived. A hole worth no packets is never
  /// repaired -- there is nothing for a late packet to deliver -- which keeps a degenerate
  /// reservation shaded rather than silently downgraded.
  bool repaired() const noexcept { return packets > 0 && filled_packets >= packets; }
};

/// Live connection and counter state, readable from the UI thread.
struct StreamStats {
  std::atomic<std::uint64_t> packets{0};
  std::atomic<std::uint64_t> frames{0};
  std::atomic<std::uint64_t> gaps{0};
  std::atomic<std::uint64_t> reconnects{0};
  std::atomic<bool> connected{false};
  std::atomic<bool> have_config{false};
  std::atomic<std::uint64_t> holes{0};       ///< reserved invalidated spans
  std::atomic<std::uint64_t> hole_slots{0};  ///< stored slots those spans cover
  /// Out-of-order packets put back in order inside the reorder window: no hole was ever
  /// seen for these.
  std::atomic<std::uint64_t> resequenced{0};
  /// Out-of-order packets delivered behind the sequence, past the window.
  std::atomic<std::uint64_t> late{0};
  /// Times the client's reorder ring was full when a packet needed a slot, so the gap was
  /// committed early. Nothing is dropped; the window was just shorter than asked for, and a
  /// non-zero reading is what says the ring is undersized for the stream.
  std::atomic<std::uint64_t> overflows{0};
  /// Packets the client still refused outright (a duplicate, or any late packet while
  /// resequencing is disabled). Counted since the beginning and displayed nowhere until
  /// now, which is why an inversion had to be inferred from `packets` and `frames`
  /// diverging.
  std::atomic<std::uint64_t> stale{0};
  /// Stored slots a late packet repaired. Declared here rather than with the backfill that
  /// writes it, because the display segment below is built once and reads it from the start:
  /// a reading of 0 beside a non-zero `holes` is exactly the honest answer until the
  /// backfill lands.
  std::atomic<std::uint64_t> backfilled{0};
};

/// 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<SignalInfo> 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<MathChannel> 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<MathChannel> 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<f32>& sample);

  /// Pushes one stored slot of a **file** load -- the values, the shared clock and the absolute
  /// clock -- and then the derived channels, in the producer's own order and under the model's own
  /// lock. The loader cannot simply hold the lock and push into `live()` itself: every setter it
  /// needs (`set_signals`, `set_signal_enabled`, ...) takes that same non-recursive lock, so the
  /// outer one deadlocks the first call.
  void push_loaded_slot(const std::vector<f32>& values, f64 t, f64 abs_t);
  /// 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);
  }
  /// What freezing a view costs, and *which thread it costs it to*. The two publishes are stalls on
  /// opposite sides of `mutex()` and so are counted apart: `publish_event()` runs on the **producer**
  /// with the lock already held, so its cost lands on the UI as a long tick (the documented `tick`
  /// max); `publish_pause_view()` runs on the **UI** thread and takes the lock, so its cost lands on
  /// the producer and can only surface as `gaps`. Neither is attributable in a frame mean, which is
  /// why they are counted rather than inferred. Each counter is written by one thread and read by the
  /// dump, so a relaxed store of a value already in hand is enough.
  ///
  /// This is the number to judge a second live ring against: freezing a whole `BufferStore` would
  /// make both figures zero, since a `ChannelBuffer` maintains its pyramid incrementally and there
  /// would be nothing to copy -- but it costs a second budget's worth of memory and a second
  /// per-sample write on the producer, so it is worth paying only if the worst figures here are
  /// large enough to matter.
  std::uint64_t publish_event_count() const noexcept {
    return publish_event_count_.load(std::memory_order_relaxed);
  }
  std::uint64_t publish_event_us_total() const noexcept {
    return publish_event_us_total_.load(std::memory_order_relaxed);
  }
  std::uint64_t publish_event_us_worst() const noexcept {
    return publish_event_us_worst_.load(std::memory_order_relaxed);
  }
  std::uint64_t publish_pause_count() const noexcept {
    return publish_pause_count_.load(std::memory_order_relaxed);
  }
  std::uint64_t publish_pause_us_total() const noexcept {
    return publish_pause_us_total_.load(std::memory_order_relaxed);
  }
  std::uint64_t publish_pause_us_worst() const noexcept {
    return publish_pause_us_worst_.load(std::memory_order_relaxed);
  }
  const std::vector<SignalInfo>& signals() const noexcept { return signals_; }
  const std::vector<ChannelInfo>& 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<SignalInfo>& 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);

  /// The calibration that *will* be applied to that signal, as `{gain, offset}` -- the identity
  /// `{1, 0}` for a signal nobody calibrated. The read counterpart of `set_calibration()`, and read
  /// from the model's own record rather than from the channels: a signal that is not acquired has
  /// no channel to read it from, and the record is what a dialog must show.
  std::pair<double, double> calibration(std::size_t signal);

  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();

  /// Records a reserved hole. Caller holds `mutex()` (the producer already does). The table
  /// is capped at `kMaxHoles`; past that the oldest is dropped, because an unbounded list of
  /// gaps is a leak and the oldest gap is the one furthest from being backfilled.
  static constexpr std::size_t kMaxHoles = 64;
  void note_hole(const Hole& h);
  const std::vector<Hole>& holes() const noexcept { return holes_; }
  /// The producer's view of the same table. `Acquisition::backfill_from()` marks a hole's packet
  /// filled as it repairs it, and that mark is what the band's fainter tint reads. Caller holds
  /// `mutex()` -- the same requirement `note_hole()` above states, and the producer already does.
  std::vector<Hole>& holes() noexcept { return holes_; }
  /// Drops holes whose time-ring slots the ring has already discarded.
  void forget_holes_before(u64 slot);

  /// 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<const EventSnapshot> 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<const EventSnapshot> 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<const EventSnapshot> 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<const EventSnapshot> 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<bool>& 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<std::uint64_t>& 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<double>& trigger_time() noexcept { return trigger_time_; }
  std::atomic<bool>& 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();

  /// Counts buffer re-plans: every `rebuild_locked()` bumps it. The producer caches the value
  /// it last saw and resets the state tied to the old store when it moves (see
  /// `Acquisition::on_data`). Not an atomic on purpose: `rebuild_locked()` only runs under
  /// `mutex()` and the producer holds that lock for all of `on_data`, so a plain read under the
  /// lock is correct, and a reconnect with an identical CONFIG is a re-plan this sees and the
  /// stride/channel-count comparison does not.
  unsigned plan_generation() const noexcept { return plan_generation_; }

 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<SignalInfo> signals_;
  std::vector<ChannelInfo> channels_;
  // --- math channels --------------------------------------------------------
  std::vector<MathChannel> math_;
  std::vector<MathRuntime> 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<int> 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<std::string> 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<f32> 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<std::uint64_t> math_slots_{0};
  std::atomic<std::uint64_t> math_timed_{0};
  std::atomic<std::uint64_t> math_cost_ns_{0};
  /// See `publish_event_count()`. Single-writer each: the first three are the producer's, the last
  /// three the UI thread's.
  std::atomic<std::uint64_t> publish_event_count_{0};
  std::atomic<std::uint64_t> publish_event_us_total_{0};
  std::atomic<std::uint64_t> publish_event_us_worst_{0};
  std::atomic<std::uint64_t> publish_pause_count_{0};
  std::atomic<std::uint64_t> publish_pause_us_total_{0};
  std::atomic<std::uint64_t> publish_pause_us_worst_{0};
  /// Fold one publish's elapsed microseconds into a counter triple. Called by the one thread that
  /// owns the triple, so the load/store pairs need no read-modify-write.
  static void note_publish(std::atomic<std::uint64_t>& count, std::atomic<std::uint64_t>& total,
                           std::atomic<std::uint64_t>& worst, std::uint64_t us) noexcept {
    count.store(count.load(std::memory_order_relaxed) + 1, std::memory_order_relaxed);
    total.store(total.load(std::memory_order_relaxed) + us, std::memory_order_relaxed);
    if (us > worst.load(std::memory_order_relaxed)) worst.store(us, std::memory_order_relaxed);
  }
  std::vector<bool> enabled_;

  Partition part_;
  std::unique_ptr<BufferStore> store_;
  std::vector<Hole> holes_;
  /// See `plan_generation()`. Written and read under `mutex()` only.
  unsigned plan_generation_ = 0;
  TriggerConfig trigger_{};
  StreamStats stats_{};
  f64 epoch_ = 0.0;
  std::atomic<double> 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<double> target_span_{0.0};
  std::atomic<bool> trigger_pending_{false};
  std::atomic<double> capture_edge_{0.0};
  std::atomic<std::shared_ptr<const EventSnapshot>> event_;  ///< the event on screen
  std::atomic<std::shared_ptr<const EventSnapshot>> pause_view_;  ///< the view frozen at pause
  std::atomic<bool> paused_{false};
  /// Per-signal-name calibration (see `set_calibration`).
  std::unordered_map<std::string, std::pair<double, double>> calibration_;
  std::atomic<std::uint64_t> event_seq_{0};
  std::atomic<double> display_epoch_{0.0};
  std::atomic<bool> trigger_armed_{false};
  std::atomic<bool> trigger_enabled_{false};
  std::atomic<std::uint64_t> trigger_latches_{0};
  std::atomic<double> trigger_time_{0.0};
  std::atomic<bool> trigger_time_valid_{false};
  std::atomic<unsigned> generation_{0};
  std::mutex mutex_;
};

}  // namespace udpscope