MARTe-Integrated-Components/Common/Client/go/wshub/capture_coverage_test.go

package wshub

import (
	"encoding/binary"
	"math"
	"testing"
)

// decodeCaptureSpan pulls the time extent of one signal out of a v2 frame.
func decodeCaptureSpan(t *testing.T, buf []byte, key string) (first, last float64, n int) {
	t.Helper()
	off := 1 + 8 + 8 + 8
	nSig := int(binary.LittleEndian.Uint32(buf[off:]))
	off += 4
	for i := 0; i < nSig; i++ {
		kl := int(binary.LittleEndian.Uint16(buf[off:]))
		off += 2
		k := string(buf[off : off+kl])
		off += kl
		cnt := int(binary.LittleEndian.Uint32(buf[off:]))
		off += 4
		if k == key && cnt > 0 {
			first = math.Float64frombits(binary.LittleEndian.Uint64(buf[off:]))
			last = math.Float64frombits(binary.LittleEndian.Uint64(buf[off+(cnt-1)*8:]))
			n = cnt
		}
		off += cnt * 16
	}
	return
}

// The rings only reach back over the window once they have rolled over at the
// current bucket, which takes as long as the window itself — so a window widened
// mid-run leaves the first captures asking for history the rings never stored.
// The archive kept it, and the capture must come back whole.
func TestCaptureBackfillsItsHeadFromTheArchive(t *testing.T) {
	h := NewHub()
	hw, key := newTestHistory(t, HistoryConfig{
		WindowSec: 60, Decimation: 1, MinDiskFreeMB: -1,
	}, 1000)
	h.hist = hw

	// 20 s of 1 kSps, archived in full…
	ts, vs := ramp(1000, 0.001, 20000)
	hw.write(key, ts, vs)
	// …but a ring that only ever holds the last 5 s of it.
	rb := newSigRing(5000)
	rb.write(ts, vs)
	h.rings[key] = rb

	// A 15 s window, of which the ring has the newest third.
	const t0, t1 = 1005.0, 1020.0
	buf := h.buildTriggerCapture(1015, 10, 5)
	if buf == nil {
		t.Fatal("no capture frame built")
	}
	first, last, n := decodeCaptureSpan(t, buf, key)
	if first > t0+0.05 {
		t.Errorf("capture starts at %.3f, want the window's start %.3f — the archive holds it",
			first, t0)
	}
	if last < t1-0.05 {
		t.Errorf("capture ends at %.3f, want %.3f", last, t1)
	}
	if n < 100 {
		t.Errorf("capture has %d points, too few for a 15 s window at 1 kSps", n)
	}

	// The join between the two sources must not break time order, or every
	// binary search over the capture — client-side and in the hold — misreads it.
	ct, _, ok := h.capture.slice(key, t0, t1)
	if !ok {
		t.Fatal("the hold declined the window it just published")
	}
	for i := 1; i < len(ct); i++ {
		if ct[i] < ct[i-1] {
			t.Fatalf("capture time goes backwards at %d: %.6f then %.6f", i, ct[i-1], ct[i])
		}
	}
}

// A capture that neither source could fill must not answer for the stretch it is
// missing: the client has to fall through to the archive instead of redrawing
// the same hole on every zoom.
func TestHoldDeclinesTheStretchACaptureNeverGot(t *testing.T) {
	h := NewHub()
	ts, vs := ramp(1000, 0.001, 20000)
	rb := newSigRing(5000) // the newest 5 s only, and no archive to fill from
	rb.write(ts, vs)
	h.rings["src:sig"] = rb

	if buf := h.buildTriggerCapture(1015, 10, 5); buf == nil {
		t.Fatal("no capture frame built")
	}
	if _, _, ok := h.capture.slice("src:sig", 1005, 1020); ok {
		t.Error("the hold answered for 15 s it only has the last 5 s of")
	}
	// What it does hold, it still serves.
	if _, _, ok := h.capture.slice("src:sig", 1016, 1019); !ok {
		t.Error("the hold declined a range well inside its data")
	}
}

// TestCaptureCoverageAcrossShots walks a whole acquisition the way Run() does —
// ingest, retune, dueCapture, rearm — and reports how much of each window the
// capture actually came back with.
func TestCaptureCoverageAcrossShots(t *testing.T) {
	const (
		key      = "s1:Ch1"
		rate     = 100e3 // scaled 10x down from the 1 MSps producer
		budget   = 400_000
		window   = 120.0
		prePct   = 20.0
		batchSec = 1.0 / 30.0
		simSec   = 900.0
	)

	h := NewHub()
	h.SetRingBudget(budget)
	h.rings[key] = newSigRing(ringCapInitial)

	h.trigger.SetConfig(trigConfig{signalKey: key, edge: "rising", threshold: 0,
		windowSec: window, prePercent: prePct, mode: "normal", holdoffSec: 0.2})

	rateHz := float64(rate)
	nBatch := int(rateHz * batchSec)
	ts := make([]float64, nBatch)
	vs := make([]float64, nBatch)

	armed := false
	shots := 0
	for now := 0.0; now < simSec; now += batchSec {
		for i := range ts {
			ts[i] = now + float64(i)/rateHz
			// 0.05 Hz sine: one rising zero crossing every 20 s.
			vs[i] = math.Sin(2 * math.Pi * 0.05 * ts[i])
		}
		h.ingest(key, 1, ts, vs)
		h.retuneRings(now)

		// Arm once the stream is going, as a user would.
		if !armed && now > 5 {
			h.trigger.Arm()
			armed = true
		}

		if trigTime, pre, post, ok := h.trigger.dueCapture(now + batchSec); ok {
			buf := h.buildTriggerCapture(trigTime, pre, post)
			if buf == nil {
				t.Fatalf("shot at t=%.1f produced no frame", trigTime)
			}
			first, last, n := decodeCaptureSpan(t, buf, key)
			t0, t1 := trigTime-pre, trigTime+post
			_, ringSpan := h.rings[key].stats()
			shots++
			t.Logf("shot %d fired t=%.1f window [%.1f,%.1f] got [%.1f,%.1f] "+
				"= %.0f%% (%d pts, bucket %d, ring span %.1f s)",
				shots, trigTime, t0, t1, first, last,
				100*(last-first)/(t1-t0), n, h.rings[key].bucketSize(), ringSpan)
			h.trigger.markTriggered(now + batchSec)
		} else if h.trigger.dueRearm(now + batchSec) {
			h.trigger.Arm()
		}
	}
	if shots < 3 {
		t.Fatalf("only %d shots in %.0f s", shots, simSec)
	}
}