Martino FerrariandClaude Opus 4.6
f179fbbb44
Two defects behind one report ("a trigger during a capture overrides the first,
and the hold-off does not apply"); the second is also what made a spike in one
channel look like the acquisition resetting.
The trigger engine keyed its invalidation on the trigger channel's *index*, which
every rebuild of the channel list renumbers -- enabling a signal that sorts before
it is enough. Any rebuild therefore abandoned an open capture and set no hold-off,
so the next edge inside the post window became t0. It keys on the channel's name
now, which changes only when the source genuinely does.
A packet whose stamps jump more than 256 packet spans ahead is now rejected as a
fault and *bridged* at the declared cadence, instead of poisoning the monotonic
mark (which pinned every later packet flat at the glitched value) and condemning
the next good packet. Only four consecutive failures abandon the declared time
signal, and that fallback no longer wipes the rings: the arrival clock is anchored
to continue from the monotonic mark, so the history is still on one timeline --
the wipe is what made one glitching packet look like the stream restarting from 0.
The declared unit is resolved before those checks, since they weigh the stamps
against a cadence in seconds.
Guards: UDPSCOPE_PROBE_TRIGRENUM_MS prints t0_held=1 through the real widgets,
and the mock's --time-glitch-at/--time-glitch-by injects the clock fault.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
287 lines
11 KiB
Python
287 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: MIT
|
|
# SPDX-FileCopyrightText: 2026 Martino Ferrari
|
|
"""A minimal UDPStreamer (UDPS) producer, for exercising udpscope without MARTe2.
|
|
|
|
Unicast only: binds `--port`, waits for a CONNECT datagram, then streams CONFIG
|
|
once and DATA packets to whoever connected. Publish mode is Accumulate, so each
|
|
packet carries `--batch` consecutive samples per (scalar) signal -- the shape the
|
|
viewer's per-sample path expects.
|
|
|
|
Signals: `time` (the stream's own clock, seconds), `sine`, `square`, `ramp`.
|
|
Everything is float64 and unquantised.
|
|
|
|
./tools/mock_udpstreamer.py --port 8000 --rate 10000 --batch 100
|
|
"""
|
|
|
|
import argparse
|
|
import math
|
|
import random
|
|
import socket
|
|
import struct
|
|
import time
|
|
|
|
MAGIC = 0x53504455
|
|
PKT_DATA, PKT_CONFIG, PKT_ACK, PKT_CONNECT, PKT_DISCONNECT = 0, 1, 2, 3, 4
|
|
T_FLOAT64 = 9
|
|
QUANT_NONE = 0
|
|
TIME_PACKET, TIME_FULL_ARRAY, TIME_FIRST_SAMPLE = 0, 1, 2
|
|
PUBLISH_ACCUMULATE = 1
|
|
NO_TIME_SIGNAL = 0xFFFFFFFF
|
|
|
|
|
|
def header(pkt_type, counter, payload_bytes):
|
|
return struct.pack("<IBIHHI", MAGIC, pkt_type, counter, 0, 1, payload_bytes)
|
|
|
|
|
|
def signal_desc(name, *, rate, time_mode, time_idx, unit, rows=1, cols=1, dims=0):
|
|
"""One 136-byte signal descriptor."""
|
|
d = bytearray(136)
|
|
d[0:64] = name.encode()[:64].ljust(64, b"\0")
|
|
d[64] = T_FLOAT64
|
|
d[65] = QUANT_NONE
|
|
d[66] = dims
|
|
struct.pack_into("<II", d, 67, rows, cols)
|
|
struct.pack_into("<dd", d, 75, -10.0, 10.0)
|
|
d[91] = time_mode
|
|
struct.pack_into("<d", d, 92, rate)
|
|
struct.pack_into("<I", d, 100, time_idx)
|
|
d[104:136] = unit.encode()[:32].ljust(32, b"\0")
|
|
return bytes(d)
|
|
|
|
|
|
def build_config(rate, time_mode=TIME_FULL_ARRAY, unit="V"):
|
|
sigs = [
|
|
# The stream's clock: referenced by every data signal, so udpscope treats
|
|
# it as the time base rather than a plottable channel. Its *own* time_mode
|
|
# is what udpscope switches on, so `--time-mode first` has to land here.
|
|
signal_desc(
|
|
"time", rate=0.0, time_mode=time_mode, time_idx=NO_TIME_SIGNAL, unit="s"
|
|
),
|
|
signal_desc("sine", rate=rate, time_mode=time_mode, time_idx=0, unit=unit),
|
|
signal_desc("square", rate=rate, time_mode=time_mode, time_idx=0, unit=unit),
|
|
signal_desc("ramp", rate=rate, time_mode=time_mode, time_idx=0, unit=unit),
|
|
]
|
|
payload = struct.pack("<I", len(sigs)) + b"".join(sigs)
|
|
payload += bytes([PUBLISH_ACCUMULATE])
|
|
return header(PKT_CONFIG, 0, len(payload)) + payload
|
|
|
|
|
|
def build_data(
|
|
counter, t0, dt, batch, freq, time_mode=TIME_FULL_ARRAY, noise=0.0, time_spike=0.0
|
|
):
|
|
stamps = [t0 + i * dt for i in range(batch)]
|
|
# `time_spike` models a device whose clock glitches for one packet: the stamps jump
|
|
# far ahead and then the next packet carries on from where it would have been. The
|
|
# waveforms are still computed from the *true* stamps, so only the time signal is
|
|
# wrong -- which is what a glitching clock looks like on the wire.
|
|
wire_stamps = [t + time_spike for t in stamps] if time_spike > 0.0 else stamps
|
|
# The interferer is a tone at 100x the fundamental with a randomly scaled amplitude, which
|
|
# is the shape that is hardest on anything counting crossings: near the sine's own mid level
|
|
# the fundamental's slope is 2*pi*f while the interferer's is 100x that, so the signal
|
|
# recrosses the level several times per true crossing. `--noise 0` is the ideal waveform, and
|
|
# it is the default because the documented probe assertions are *exact* (`peak to peak=2`,
|
|
# `top=0.636620`) and any interferer moves the extremes.
|
|
if noise > 0.0:
|
|
sine = [
|
|
math.sin(2.0 * math.pi * freq * t)
|
|
+ random.random() * noise * math.sin(2.0 * math.pi * 100.0 * freq * t)
|
|
for t in stamps
|
|
]
|
|
else:
|
|
sine = [math.sin(2.0 * math.pi * freq * t) for t in stamps]
|
|
square = [
|
|
1.0 if math.sin(2.0 * math.pi * freq * 0.5 * t) >= 0.0 else -1.0 for t in stamps
|
|
]
|
|
ramp = [(freq * t) % 1.0 * 2.0 - 1.0 for t in stamps]
|
|
# In FIRST_SAMPLE mode only element 0 of the time signal is meaningful: the producer
|
|
# stamps the first sample of the batch and says nothing about the rest. A scalar
|
|
# signal in Accumulate mode still occupies `batch` slots on the wire (the client
|
|
# reads `num_samples` values for any 1-element signal), so the anchor is repeated --
|
|
# which is exactly the shape that made udpscope derive a step of zero.
|
|
wire_time = (
|
|
[wire_stamps[0]] * batch if time_mode == TIME_FIRST_SAMPLE else wire_stamps
|
|
)
|
|
payload = struct.pack("<QI", int(t0 * 1e9), batch)
|
|
for arr in (wire_time, sine, square, ramp):
|
|
payload += struct.pack("<%dd" % batch, *arr)
|
|
return header(PKT_DATA, counter, len(payload)) + payload
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--port", type=int, default=8000)
|
|
ap.add_argument("--rate", type=float, default=10000.0, help="samples per second")
|
|
ap.add_argument("--batch", type=int, default=100, help="samples per packet")
|
|
ap.add_argument("--freq", type=float, default=5.0, help="waveform frequency, Hz")
|
|
ap.add_argument("--duration", type=float, default=0.0, help="0 = forever")
|
|
ap.add_argument(
|
|
"--stream-scale",
|
|
type=float,
|
|
default=1.0,
|
|
help="stream seconds per wall second: 0.1 mimics a producer whose own "
|
|
"clock advances ten times slower than the wire delivers samples",
|
|
)
|
|
ap.add_argument(
|
|
"--declared-rate",
|
|
type=float,
|
|
default=-1.0,
|
|
help="sampling_rate to put in CONFIG; -1 = the real rate, 0 = 'unknown'",
|
|
)
|
|
ap.add_argument(
|
|
"--time-mode",
|
|
choices=("full", "first"),
|
|
default="full",
|
|
help="how the stream carries its clock: 'full' stamps every sample "
|
|
"(UDPS_TIME_FULL_ARRAY), 'first' stamps only the batch's first "
|
|
"sample (UDPS_TIME_FIRST_SAMPLE) and leaves the viewer to space the "
|
|
"rest by the declared cadence -- the shape a real MARTe2-style "
|
|
"producer with a single u64 clock has",
|
|
)
|
|
ap.add_argument(
|
|
"--unit",
|
|
default="V",
|
|
help="the unit the data signals declare in CONFIG; '' declares none, which "
|
|
"is an ordinary stream (and what every math channel looks like) and is "
|
|
"the only way a probe can see the unitless readout formatting",
|
|
)
|
|
ap.add_argument(
|
|
"--noise",
|
|
type=float,
|
|
default=0.0,
|
|
help="amplitude of an interferer added to `sine`: a tone at 100x the fundamental, "
|
|
"scaled by a fresh uniform random factor per sample. 0 = the ideal sine, which is "
|
|
"what the exact probe assertions (peak to peak = 2, top = 2/pi) are read against; "
|
|
"0.1 is the figure the measurement hysteresis is sized against",
|
|
)
|
|
ap.add_argument(
|
|
"--clock-jitter",
|
|
type=float,
|
|
default=0.0,
|
|
help="per-packet stamp jitter, as a fraction of one packet's span: a device "
|
|
"that stamps each batch from its own clock does not land on a perfect "
|
|
"grid, so the stream's timestamps wobble about one. Non-cumulative "
|
|
"(the mean rate is unchanged); 0 = the ideal grid",
|
|
)
|
|
ap.add_argument(
|
|
"--time-glitch-at",
|
|
type=float,
|
|
default=-1.0,
|
|
help="stream time (s) at which ONE packet's time signal jumps ahead by "
|
|
"--time-glitch-by and then carries on normally: a device whose clock "
|
|
"glitches. -1 = never",
|
|
)
|
|
ap.add_argument(
|
|
"--time-glitch-by",
|
|
type=float,
|
|
default=10.0,
|
|
help="how far ahead the glitching packet's stamps jump, in stream seconds",
|
|
)
|
|
args = ap.parse_args()
|
|
|
|
dt = args.stream_scale / args.rate # stream seconds per sample
|
|
period = args.batch / args.rate # wall seconds per packet
|
|
declared = args.rate if args.declared_rate < 0.0 else args.declared_rate
|
|
time_mode = TIME_FIRST_SAMPLE if args.time_mode == "first" else TIME_FULL_ARRAY
|
|
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
sock.bind(("0.0.0.0", args.port))
|
|
sock.settimeout(0.001)
|
|
print(
|
|
"mock udpstreamer: waiting for CONNECT on :%d (%.0f S/s wire, %d per packet, "
|
|
"stream clock x%.3g, declared %.0f S/s, time-mode %s)"
|
|
% (
|
|
args.port,
|
|
args.rate,
|
|
args.batch,
|
|
args.stream_scale,
|
|
declared,
|
|
args.time_mode,
|
|
),
|
|
flush=True,
|
|
)
|
|
|
|
peer = None
|
|
counter = 1
|
|
glitched = False
|
|
t_stream = 0.0
|
|
started = None
|
|
next_send = time.monotonic()
|
|
|
|
while True:
|
|
try:
|
|
buf, addr = sock.recvfrom(2048)
|
|
if len(buf) >= 17:
|
|
magic, ptype = struct.unpack_from("<IB", buf, 0)
|
|
if magic == MAGIC and ptype == PKT_CONNECT:
|
|
peer = addr
|
|
counter = 1
|
|
glitched = False
|
|
t_stream = 0.0
|
|
started = time.monotonic()
|
|
next_send = started
|
|
sock.sendto(build_config(declared, time_mode, args.unit), peer)
|
|
print(
|
|
"mock udpstreamer: CONNECT from %s:%d, CONFIG sent" % peer,
|
|
flush=True,
|
|
)
|
|
elif magic == MAGIC and ptype == PKT_DISCONNECT:
|
|
print("mock udpstreamer: DISCONNECT", flush=True)
|
|
peer = None
|
|
except socket.timeout:
|
|
pass
|
|
|
|
if peer is None:
|
|
continue
|
|
|
|
now = time.monotonic()
|
|
if now < next_send:
|
|
time.sleep(min(next_send - now, 0.002))
|
|
continue
|
|
|
|
# The jitter varies each *packet's* own spacing rather than offsetting its start, because
|
|
# the stream's timestamps have to stay strictly increasing -- a jittered t0 can overlap the
|
|
# packet before it, and the buffer, the trigger and every binary search here assume a
|
|
# monotonic clock. The mean spacing is still `dt`, so the declared rate stays honest; what
|
|
# moves is the *local* spacing, which is what a uniform grid over the window cannot model.
|
|
pdt = dt
|
|
if args.clock_jitter > 0.0:
|
|
pdt = dt * (1.0 + random.uniform(-args.clock_jitter, args.clock_jitter))
|
|
spike = 0.0
|
|
if args.time_glitch_at >= 0.0 and not glitched and t_stream >= args.time_glitch_at:
|
|
spike = args.time_glitch_by
|
|
glitched = True
|
|
print(
|
|
"mock udpstreamer: time-signal glitch at t=%.3f, +%.3f s for one packet"
|
|
% (t_stream, spike),
|
|
flush=True,
|
|
)
|
|
sock.sendto(
|
|
build_data(
|
|
counter,
|
|
t_stream,
|
|
pdt,
|
|
args.batch,
|
|
args.freq,
|
|
time_mode,
|
|
args.noise,
|
|
spike,
|
|
),
|
|
peer,
|
|
)
|
|
counter += 1
|
|
t_stream += args.batch * pdt
|
|
next_send += period
|
|
if (
|
|
args.duration > 0.0
|
|
and started is not None
|
|
and now - started > args.duration
|
|
):
|
|
print("mock udpstreamer: done (%d packets)" % counter, flush=True)
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|