diff --git a/tools/mock_udpstreamer.py b/tools/mock_udpstreamer.py index 1b404a1..fd882ee 100644 --- a/tools/mock_udpstreamer.py +++ b/tools/mock_udpstreamer.py @@ -68,7 +68,17 @@ def build_config(rate, time_mode=TIME_FULL_ARRAY, unit="V"): def build_data( - counter, t0, dt, batch, freq, time_mode=TIME_FULL_ARRAY, noise=0.0, time_spike=0.0 + counter, + t0, + dt, + batch, + freq, + time_mode=TIME_FULL_ARRAY, + noise=0.0, + time_spike=0.0, + glitch_at=-1.0, + glitch_width=0.0, + quiet_at=-1.0, ): stamps = [t0 + i * dt for i in range(batch)] # `time_spike` models a device whose clock glitches for one packet: the stamps jump @@ -94,6 +104,30 @@ def build_data( 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] + # A glitch is an *inversion* of the square for `glitch_width` seconds, repeated once per + # square period (2/freq -- the period of sin(2*pi*freq*0.5*t)). Inverting rather than + # forcing a level makes it a narrow pulse whatever the square's polarity is, and repeating + # it lets a probe assert the *spacing* of the hits a width condition finds instead of an + # absolute count, which depends on when the probe happened to start looking. + if glitch_at >= 0.0 and glitch_width > 0.0: + sq_period = 2.0 / freq + for i, t in enumerate(stamps): + if t < glitch_at: + continue + k = math.floor((t - glitch_at) / sq_period) + start = glitch_at + k * sq_period + if t < start + glitch_width: + square[i] = -square[i] + # `--quiet-at` is for the dropout condition: the sine stops at a level on the *opposite* + # side of zero from where it was, so `quiet_at` itself is its last zero crossing and a + # dropout at level 0 fires at exactly `quiet_at + timeout`. Holding at the same sign would + # leave the last crossing at whichever earlier zero happened to precede it, which is not a + # figure a probe can assert. Choose a `quiet_at` away from the sine's own zeros. + if quiet_at >= 0.0: + held = -0.5 if math.sin(2.0 * math.pi * freq * quiet_at) >= 0.0 else 0.5 + for i, t in enumerate(stamps): + if t >= quiet_at: + sine[i] = held # 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 @@ -177,6 +211,28 @@ def main(): default=10.0, help="how far ahead the glitching packet's stamps jump, in stream seconds", ) + ap.add_argument( + "--glitch-at", + type=float, + default=-1.0, + help="stream time (s) of the first narrow pulse injected into `square`: the square " + "is inverted for --glitch-width seconds, once per square period (2/freq), so a " + "pulse-width or glitch trigger has a repeating event to find. -1 = never", + ) + ap.add_argument( + "--glitch-width", + type=float, + default=0.0, + help="width of each injected pulse, in stream seconds", + ) + ap.add_argument( + "--quiet-at", + type=float, + default=-1.0, + help="stream time (s) after which `sine` is held at a constant level on the " + "opposite side of zero, so --quiet-at is its last zero crossing: the shape a " + "dropout trigger is read against. -1 = never", + ) args = ap.parse_args() dt = args.stream_scale / args.rate # stream seconds per sample @@ -267,6 +323,9 @@ def main(): time_mode, args.noise, spike, + args.glitch_at, + args.glitch_width, + args.quiet_at, ), peer, )