fix: a crossing is counted with hysteresis, so noise is not counted as cycles

period, frequency and pulse duration counted every sign change about a single
level, which is unusable on any real signal: the mid level is exactly where a
waveform moves fastest, so an interferer faster than the fundamental recrosses
it several times per true crossing. A 5 Hz sine with a 500 Hz tone at a tenth of
the amplitude read 79 crossings for 9, period=0.025622 against 0.2.

MeasureLevel gains a hysteresis band sized once in measure_set_level(): a tenth
of the window's peak-to-peak, clamped to half the room to the nearer extreme so
a threshold set near one stays usable. The band moves the edge decision only --
the sum/count split stays at the bare level, since that split is top/bottom and
has a documented exact reading. The bucket fast path tests the band and the
level separately rather than folding them, or it would refuse exactly the
buckets a crossing passes through: clean cost is unchanged at raw=0.1% and
1.41 ms a recompute.

tools/mock_udpstreamer.py gains --noise, defaulting to 0 so the exact probe
assertions still hold.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

tools/mock_udpstreamer.py

@@ -55,8 +55,9 @@ def build_config(rate, time_mode=TIME_FULL_ARRAY, unit="V"):
         # 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(
+            "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),
@@ -66,10 +67,25 @@ def build_config(rate, time_mode=TIME_FULL_ARRAY, unit="V"):
     return header(PKT_CONFIG, 0, len(payload)) + payload
 
 
-def build_data(counter, t0, dt, batch, freq, time_mode=TIME_FULL_ARRAY):
+def build_data(counter, t0, dt, batch, freq, time_mode=TIME_FULL_ARRAY, noise=0.0):
     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]
+    # 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
@@ -90,30 +106,58 @@ def main():
     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")
+    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",
+    )
     args = ap.parse_args()
 
-    dt = args.stream_scale / args.rate          # stream seconds per sample
-    period = args.batch / args.rate             # wall seconds per packet
+    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
 
@@ -121,9 +165,19 @@ def main():
     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)
+    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
@@ -143,7 +197,10 @@ def main():
                     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)
+                    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
@@ -166,11 +223,20 @@ def main():
         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)
+        sock.sendto(
+            build_data(
+                counter, t_stream, pdt, args.batch, args.freq, time_mode, args.noise
+            ),
+            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:
+        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