feat: the mock can inject a narrow pulse and stop a signal, so a width and a dropout condition have something to find

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,
         )