Episode 12: Crossing a Threshold

Turn a feature into a yes-or-no decision with a Threshold, and stop it chattering with release and dwell.

At some point a measurement has to become a decision: is the feature high enough, yes or no. That step is where a well-behaved pipeline usually starts misbehaving, because a bare comparison flips on every sample that grazes the level, and a noisy signal grazes it a lot.

A slow swing in noise

A 0.5 Hz sine starts at zero and crosses it three more times in four seconds; with the comparison’s initial state that is four transitions. Noise of standard deviation 0.4 on top of it crosses zero many more times by accident.

27    # A slow swing that crosses zero four times in four seconds, in noise
28    # big enough to cross it many more times by accident
29    source = gp.Generator(sampling_rate=SAMPLING_RATE, channel_count=1,
30                          signal_frequency=SLOW_HZ, signal_amplitude=1.0,
31                          noise_amplitude=0.4)
Four decisions about it

A bare Threshold(level=0). Two with hysteresis: release gives leaving the state its own, lower level, so getting out takes more than the noise that got you in. One with dwell: the new state must hold for 25 samples, a tenth of a second, before it is believed.

33    # Four decisions about the same signal
34    plain = gp.Threshold(level=0.0)
35    # Hysteresis: leaving the state needs the signal well back below it.
36    # A gap inside the noise barely helps; two standard deviations do
37    narrow = gp.Threshold(level=0.2, release=-0.2)
38    wide = gp.Threshold(level=0.8, release=-0.8)
39    # Dwell: the new state must hold for 25 samples before it is believed
40    dwell = gp.Threshold(level=0.0, dwell=25)
41
42    kept = {}
43    for name, node in (("plain", plain), ("narrow", narrow),
44                       ("wide", wide), ("dwell", dwell)):
45        kept[name] = gp.Collector(name=name)
46        p.connect(source, node)
47        p.connect(node, kept[name])
Result

Run the script. A helper counts how often each decision changed state:

a 0.5 Hz swing over 4 s, four crossings of zero, noise of standard deviation 0.4
  Threshold(level=0)                    ... transitions
  Threshold(level=0.2, release=-0.2)    ... transitions
  Threshold(level=0.8, release=-0.8)    ... transitions
  Threshold(level=0, dwell=25)            4 transitions

Four is the answer the signal deserves. The bare comparison is not wrong about any single sample, every one of those flips really did cross zero; it is answering a question nobody asked. Hysteresis only helps if the gap is sized against the noise: plus or minus 0.2 sits inside a standard deviation of 0.4 and barely helps, plus or minus 0.8 works. Hysteresis chosen by eye is a knob that looks tuned and is not. Dwell costs latency by construction: 25 samples at 250 Hz is 100 ms on every decision at the least, and in noise more, since the state flips only once the signal has been clear of the level by a few standard deviations for 25 samples running, a trade to make on purpose. The noise is random, so the first three counts change from run to run.

Summary

In this episode, you learned how to:

  • make a decision with Threshold

  • size release against the noise, not by eye

  • pay for dwell in latency, knowingly

  • read the output, 0.0 or 1.0, as a channel like any other

Proceed to Episode 13, where GenericFilter runs a filter of your own design.

Nodes in this episode: Generator, Threshold, Collector

File s2e12_crossing_a_threshold.py – View file on GitHub

 1"""Season 2, Episode 12: a Threshold turns a feature into a decision."""
 2
 3import time
 4
 5import numpy as np
 6
 7import gpype as gp
 8
 9SAMPLING_RATE = 250
10SECONDS = 4.0
11SLOW_HZ = 0.5  # 2 full cycles in 4 s, so 4 honest crossings of zero
12
13
14def transitions(collector):
15    """How many times the held state changed."""
16    result = collector.result
17    if result is None:
18        return -1
19    state = np.asarray(result.data).reshape(-1)
20    return int(np.count_nonzero(np.diff(state)))
21
22
23if __name__ == "__main__":
24
25    p = gp.Pipeline()
26
27    # A slow swing that crosses zero four times in four seconds, in noise
28    # big enough to cross it many more times by accident
29    source = gp.Generator(sampling_rate=SAMPLING_RATE, channel_count=1,
30                          signal_frequency=SLOW_HZ, signal_amplitude=1.0,
31                          noise_amplitude=0.4)
32
33    # Four decisions about the same signal
34    plain = gp.Threshold(level=0.0)
35    # Hysteresis: leaving the state needs the signal well back below it.
36    # A gap inside the noise barely helps; two standard deviations do
37    narrow = gp.Threshold(level=0.2, release=-0.2)
38    wide = gp.Threshold(level=0.8, release=-0.8)
39    # Dwell: the new state must hold for 25 samples before it is believed
40    dwell = gp.Threshold(level=0.0, dwell=25)
41
42    kept = {}
43    for name, node in (("plain", plain), ("narrow", narrow),
44                       ("wide", wide), ("dwell", dwell)):
45        kept[name] = gp.Collector(name=name)
46        p.connect(source, node)
47        p.connect(node, kept[name])
48
49    p.start()
50    time.sleep(SECONDS)
51    p.stop()
52    p.close()
53
54    print(f"a {SLOW_HZ} Hz swing over {SECONDS:.0f} s, four crossings of "
55          "zero, noise of standard deviation 0.4")
56    print(f"  Threshold(level=0)                  "
57          f"{transitions(kept['plain']):5d} transitions")
58    print(f"  Threshold(level=0.2, release=-0.2)  "
59          f"{transitions(kept['narrow']):5d} transitions")
60    print(f"  Threshold(level=0.8, release=-0.8)  "
61          f"{transitions(kept['wide']):5d} transitions")
62    print(f"  Threshold(level=0, dwell=25)        "
63          f"{transitions(kept['dwell']):5d} transitions")