improved timing acq

src/acquisition.cpp

@@ -138,9 +138,31 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign
 
   double period = model_.partition().base_dt;
   if (!(period > 0.0)) period = 1.0e-3;
+  // The cadence of the samples actually stored: the partition's slot interval (base
+  // tick times the clock's decimation), as *declared* by the CONFIG.
+  const double slot = model_.partition().slot_dt();
+  double cadence = slot > 0.0 ? slot : period;
+  // A declared rate is a claim the stream may not keep (a CONFIG declaring 1 MS/s
+  // while delivering a hundred samples every fifth of a second is the case that
+  // exposed this): stamps spaced by the declaration then span microseconds inside a
+  // frame and jump by the frame interval between frames, which reads as a staircase
+  // in the x column and, in an export, as data that is "downsampled" while the
+  // values are fine. When the arrivals disagree with the declaration, the arrivals
+  // are the truth -- the mean spacing of the samples that actually arrived.
+  if (have_arrival_interval_ && n > 0) {
+    const double measured = arrival_interval_ / double(n);
+    const bool agrees = cadence > 0.0 && measured >= 0.75 * cadence && measured <= 1.25 * cadence;
+    if (measured > 0.0 && std::isfinite(measured) && !agrees) {
+      cadence = measured;
+    }
+  }
 
   bool used_stream_time = false;
-  if (time_signal >= 0 && time_signal < static_cast<int>(frame->num_signals)) {
+  // Once the stream has been judged to have no usable time base, it keeps the
+  // arrival clock for the rest of its life (until a CONFIG says otherwise): letting
+  // the decision flip per frame is what interleaved two timelines in one ring.
+  if (time_source_ != 2 && 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
@@ -170,7 +192,7 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign
       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;
+          stamps_[k] = anchor - double(n - 1 - k) * cadence;
         used_stream_time = true;
         break;
       }
@@ -233,15 +255,59 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign
     if (!have_arrival_epoch_) {
       arrival_epoch_ = end;
       have_arrival_epoch_ = true;
+    } else {
+      const double elapsed = end - arrival_prev_;
+      if (elapsed > 0.0 && std::isfinite(elapsed)) {
+        // Smoothed: the socket clock jitters by a millisecond or so per frame, and
+        // an unsmoothed step would make every frame's span disagree slightly.
+        arrival_interval_ = have_arrival_interval_ ? 0.8 * arrival_interval_ + 0.2 * elapsed
+                                                   : elapsed;
+        have_arrival_interval_ = true;
+      }
     }
+    arrival_prev_ = end;
     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;
+    // The spacing between two *stored* samples is the partition's own cadence --
+    // the exact interval the ingest filter decimates to, which is `slot_dt()` and
+    // not the base tick. Deriving it from the arrival gap divided by
+    // `n * (lost + 1)` collapsed the step to ~zero whenever the client reported
+    // losses, and a whole frame then shared one f32 instant: the x column became a
+    // staircase (many rows at one time, then a jump) while the values were fine --
+    // which is what an export of it looks like, and what the plot's crosshair and
+    // trigger times were reading. The arrival clock still anchors the frame's LAST
+    // sample, so a lost packet or a stalled producer still shows up as a jump
+    // *between* frames.
+    // The frame's stamps start at the later of two anchors: where the arrival clock
+    // puts the frame's first sample, and one cadence past the previous frame's last.
+    // The second is what keeps the timeline strictly increasing -- anchoring the
+    // frame's *end* at the arrival let a cadence estimate even slightly long reach
+    // back past the previous end, and the monotonic clamp then flattened the overlap
+    // into equal timestamps (a consumer plotting against them draws a vertical line
+    // there, and an export reads as if sampled twice at one instant). A gap still
+    // shows: when the arrivals run ahead, the first anchor wins and the frame jumps.
+    const double base_rel = base - model_.epoch();
+    const double start = std::max(base_rel - double(n - 1) * cadence, last_stamp_ + cadence);
+    for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = start + double(k) * cadence + model_.epoch();
+  }
+
+  // A change of time source invalidates the running mark: the stream-time branch
+  // stamps in the time signal's own values and the arrival branch in seconds since
+  // the first packet, and the two timelines share no zero. Leaving the mark at the
+  // old source's last value made every arriving frame clamp flat onto it -- which is
+  // the 2.5 s plateau of one timestamp that a run reports when its declared time
+  // signal turns out unusable and the app falls back mid-stream.
+  if (time_source_ == 0) {
+    time_source_ = used_stream_time ? 1 : 2;
+  } else if (time_source_ == 1 && !used_stream_time) {
+    // The declared time signal failed its checks: keep the arrival clock from here
+    // on, discard the stamps already recorded on the old timeline, and make the
+    // arrival clock continue from where the mark is, so the axis has no step.
+    time_source_ = 2;
+    const double end = frame->recv_time;
+    arrival_epoch_ = end - (last_stamp_ + model_.epoch());
+    arrival_prev_ = end;
+    last_stamp_ = 0.0;
+    history_reset_pending_ = true;
   }
 
   // A stream-time signal may be declared in a time unit rather than seconds
@@ -262,8 +328,6 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign
     }
   }
 
-  last_recv_ = frame->recv_time;
-  have_recv_ = true;
   return n;
 }
 
@@ -301,6 +365,12 @@ void Acquisition::on_data(const udps_frame_t* frame) {
   }
   const std::uint32_t n = build_stamps(frame, time_signal);
   if (n == 0) return;
+  // A change of time source invalidated the recorded history: empty the rings before
+  // anything more goes in, so no reader can find two timelines under one index.
+  if (history_reset_pending_) {
+    history_reset_pending_ = false;
+    model_.reset_live_history_locked();  // this thread holds the model lock
+  }
 
   if (model_.epoch() == 0.0) {
     model_.set_epoch_if_unset(stamps_[0]);