test: the mock can inject an alternating-sign impulse train on sine

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Martino FerrariandClaude Sonnet 4.6 committed 2026-10-13 14:55:57 +02:00
1 parent be71a1d4bb
commit d5a9b3c5d5
2 files changed
+41

No files matched your search

+19
View File
@@ -60,6 +60,25 @@ the value is required, and the popover button uses 1000 when the flag was absent
The flag is **not** persisted in the session file: persistence of the enable state
belongs with `todo.md` item 3, alongside the relative-time toggle.
`tools/mock_udpstreamer.py` has waveform-injection options used by the probes:
- `--glitch-at S` / `--glitch-width W` invert the `square` signal for `W` seconds at the
start of every square period from stream second `S` onwards — a runt pulse of a known
width, which is what a `<` pulse-width trigger and a glitch search are checked against.
- `--quiet-at S` freezes `sine` at its held value from stream second `S` onwards, so it
stops crossing zero: the dropout trigger's input.
- `--spike-every S` / `--spike-amp A` / `--spike-width N` inject an alternating-sign
impulse train onto `sine` only — the other four signals (`time`, `square`, `ramp`,
`noise`) are untouched, which is what lets a probe compare a spiked channel against a
clean one in the same run. Every `S` seconds a spike of `N` samples at amplitude `A` is
added to `sine`; the sign alternates per period because a one-signed train is a DC offset
a median would partly track rather than reject. `N` is in **samples**, not seconds, so its
duration follows `--rate`. `0` (the default for both `S` and `A`) injects nothing, so
every existing invocation is unchanged. **`--batch` must stay at or below 1000** with or
without these options: 2000 samples × 5 signals × `f64` is 80 kB and exceeds UDP's
~65507-byte payload limit, and the mock dies on its first `sendto` with
`OSError: [Errno 90] Message too long` while the run still exits 0.
Run with no `--server` and udpscope asks which source to open: the sessions it has stored, plus a
form for a new host. Every probe below passes `--server`, so none of them sees it.
+22
View File
@@ -82,6 +82,9 @@ def build_data(
glitch_at=-1.0,
glitch_width=0.0,
quiet_at=-1.0,
spike_every=0.0,
spike_amp=0.0,
spike_width=0,
):
stamps = [t0 + i * dt for i in range(batch)]
# `time_spike` models a device whose clock glitches for one packet: the stamps jump
@@ -103,6 +106,16 @@ def build_data(
]
else:
sine = [math.sin(2.0 * math.pi * freq * t) for t in stamps]
if spike_every > 0.0 and spike_amp != 0.0 and spike_width > 0:
# An impulse of `spike_width` samples at the start of every `spike_every`
# window, the sign alternating window to window so the train carries no
# DC a median could track. `stamps` are absolute stream seconds.
stamps_arr = np.array(stamps)
period = np.floor(stamps_arr / spike_every)
into = stamps_arr - period * spike_every
hit = into < (spike_width * dt)
sign = np.where((period.astype(np.int64) & 1) == 0, 1.0, -1.0)
sine = np.array(sine) + np.where(hit, sign * spike_amp, 0.0)
square = [
1.0 if math.sin(2.0 * math.pi * freq * 0.5 * t) >= 0.0 else -1.0 for t in stamps
]
@@ -240,6 +253,12 @@ def main():
"opposite side of zero, so --quiet-at is its last zero crossing: the shape a "
"dropout trigger is read against. -1 = never",
)
ap.add_argument("--spike-every", type=float, default=0.0,
help="seconds between impulses on sine (0 disables)")
ap.add_argument("--spike-amp", type=float, default=0.0,
help="impulse amplitude, alternating sign window to window")
ap.add_argument("--spike-width", type=int, default=0,
help="impulse length in samples")
ap.add_argument(
"--blackout-at",
type=float,
@@ -435,6 +454,9 @@ def main():
args.glitch_at,
args.glitch_width,
args.quiet_at,
args.spike_every,
args.spike_amp,
args.spike_width,
)
burst = max(1, args.drop_burst)
if args.drop_every > 0 and drop_left == 0 and counter % args.drop_every == 0: