udpscope/tools/mock_udpstreamer.py

#!/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):
    stamps = [t0 + i * dt for i in range(batch)]
    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 = [t0] * batch if time_mode == TIME_FIRST_SAMPLE else 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("--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")
    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
    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
                    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))
        sock.sendto(build_data(counter, t_stream, pdt, args.batch, args.freq, time_mode), 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()