Add "plot input" example

Author: mgeier

doc/examples.rst

@@ -17,6 +17,13 @@ Input to Output Pass-Through
 
 .. literalinclude:: ../examples/wire.py
 
+Plot Microphone Signal(s) in Real-Time
+--------------------------------------
+
+:download:`plot_input.py <../examples/plot_input.py>`
+
+.. literalinclude:: ../examples/plot_input.py
+
 Real-Time Text-Mode Spectrogram
 -------------------------------
 

examples/plot_input.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""Plot the live microphone signal(s) with matplotlib."""
+import argparse
+from queue import Queue, Empty
+
+
+def int_or_str(text):
+    """Helper function for argument parsing."""
+    try:
+        return int(text)
+    except ValueError:
+        return text
+
+
+parser = argparse.ArgumentParser(description=__doc__)
+parser.add_argument(
+    '-l', '--list-devices', action='store_true',
+    help='show list of audio devices and exit')
+parser.add_argument(
+    '-d', '--device', type=int_or_str,
+    help='input device (numeric ID or substring)')
+parser.add_argument(
+    '-w', '--window', type=float, default=200, metavar='DURATION',
+    help='visible time slot (default: %(default)s ms)')
+parser.add_argument(
+    '-i', '--interval', type=float, default=30,
+    help='minimum time between plot updates (default: %(default)s ms)')
+parser.add_argument(
+    '-b', '--blocksize', type=int, help='block size (in samples)')
+parser.add_argument(
+    '-r', '--samplerate', type=float, help='sampling rate of audio device')
+parser.add_argument(
+    '-n', '--downsample', type=int, default=10, metavar='N',
+    help='display every Nth sample (default: %(default)s)')
+parser.add_argument(
+    'channels', type=int, default=[1], nargs='*', metavar='CHANNEL',
+    help='input channels to plot (default: the first)')
+args = parser.parse_args()
+if any(c < 1 for c in args.channels):
+    parser.error('argument CHANNEL: must be >= 1')
+mapping = [c - 1 for c in args.channels]  # Channel numbers start with 1
+queue = Queue()
+
+
+def audio_callback(indata, frames, time, status):
+    """This is called (from a separate thread) for each audio block."""
+    if status:
+        print(status, flush=True)
+    # Fancy indexing with mapping creates a (necessary!) copy:
+    queue.put(indata[::args.downsample, mapping])
+
+
+def update_plot(frame):
+    """This is called by matplotlib for each plot update.
+
+    Typically, audio callbacks happen more frequently than plot updates,
+    therefore the queue tends to contain multiple blocks of audio data.
+
+    """
+    global plotdata
+    block = True  # The first read from the queue is blocking ...
+    while True:
+        try:
+            data = queue.get(block=block)
+        except Empty:
+            break
+        shift = len(data)
+        plotdata = np.roll(plotdata, -shift, axis=0)
+        plotdata[-shift:, :] = data
+        block = False  # ... all further reads are non-blocking
+    for column, line in enumerate(lines):
+        line.set_ydata(plotdata[:, column])
+    return lines
+
+
+try:
+    from matplotlib.animation import FuncAnimation
+    import matplotlib.pyplot as plt
+    import numpy as np
+    import sounddevice as sd
+
+    if args.list_devices:
+        print(sd.query_devices())
+        parser.exit()
+    if args.samplerate is None:
+        device_info = sd.query_devices(args.device, 'input')
+        args.samplerate = device_info['default_samplerate']
+
+    length = np.ceil(args.window * args.samplerate / (1000 * args.downsample))
+    plotdata = np.zeros((length, len(args.channels)))
+
+    fig, ax = plt.subplots()
+    lines = ax.plot(plotdata)
+    if len(args.channels) > 1:
+        ax.legend(['channel {}'.format(c) for c in args.channels],
+                  loc='lower left', ncol=len(args.channels))
+    ax.axis((0, len(plotdata), -1, 1))
+    ax.set_yticks([0])
+    ax.yaxis.grid(True)
+    ax.tick_params(bottom='off', top='off', labelbottom='off',
+                   right='off', left='off', labelleft='off')
+    fig.tight_layout(pad=0)
+
+    stream = sd.InputStream(
+        device=args.device, channels=max(args.channels),
+        samplerate=args.samplerate, callback=audio_callback)
+    ani = FuncAnimation(fig, update_plot, interval=args.interval, blit=True)
+    with stream:
+        plt.show()
+except Exception as e:
+    parser.exit(type(e).__name__ + ': ' + str(e))