Skip to content

Commit

Permalink
Simple dependency-free ALSA test rig for PCM capture analysis.
Browse files Browse the repository at this point in the history 
Just drop this script on a test device to run it.  No tools to build,
no dependencies to install.  Confirmed to run on Python 3.8+ with
nothing more than the core libraries and a working libasound.so.2
visible to the runtime linker.

When run without arguments, the tool will record from the capture
device for the specified duration, then emit the resulting samples
back out the playback device without processing (except potentially to
convert the sample format from s32_le to s16_le if needed, and to
discard any channels beyond those supported by the playback device).

Passing --chirp-test enables a playback-to-capture latency detector:
the tool will emit a short ~6 kHz wave packet via ALSA's mmap
interface (which allows measuring and correcting for the buffer
latency from the userspace process) and simultaneously loop on short
reads from the capture device looking for the moment it arrives.

Passing --echo-test enables a capture-while-playback test.  The script
will play a specified .wav file ("noise.wav" by default) for the
specified duration, while simultaneously capturing, and report the
"power" (in essentially arbitrary units, but it's linear with actual
signal energy assuming the sample space is itself linear) of the
captured data to stdout at the end of the test.

Signed-off-by: Andy Ross <[email protected]>
  • Loading branch information
@andyross @marc-hb
andyross authored and marc-hb committed Jan 10, 2024
1 parent a4d3f24 commit bc3ced5
Showing 1 changed file with 397 additions and 0 deletions.
397 changes: 397 additions & 0 deletions tools/capture-test.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,397 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: BSD-3-Clause
# Copyright 2024 Google LLC
# Author: Andy Ross <[email protected]>
import os
import re
import sys
import time
import struct
import random
import argparse
import ctypes as C

HELP_TEXT="""
Simple dependency-free ALSA test rig for PCM capture analysis.
Just drop this script on a test device to run it. No tools to build,
no dependencies to install. Confirmed to run on Python 3.8+ with
nothing more than the core libraries and a working libasound.so.2
visible to the runtime linker.
When run without arguments, the tool will record from the capture
device for the specified duration, then emit the resulting samples
back out the playback device without processing (except potentially to
convert the sample format from s32_le to s16_le if needed, and to
discard any channels beyond those supported by the playback device).
Passing --chirp-test enables a playback-to-capture latency detector:
the tool will emit a short ~6 kHz wave packet via ALSA's mmap
interface (which allows measuring and correcting for the buffer
latency from the userspace process) and simultaneously loop on short
reads from the capture device looking for the moment it arrives.
Passing --echo-test enables a capture-while-playback test. The script
will play a specified .wav file ("noise.wav" by default) for the
specified duration, while simultaneously capturing, and report the
"power" (in essentially arbitrary units, but it's linear with actual
signal energy assuming the sample space is itself linear) of the
captured data to stdout at the end of the test.
The tool supports a "--disable-rtnr" for the specific case of RTNR
noise reduction, which has been observed to suppress the chirp test
occasionally. It does not otherwise change the runtime configuration
of the ALSA device and expects it to be configured by the user for the
specific case under test.
"""

def parse_opts():
global opts
ap = argparse.ArgumentParser(description=HELP_TEXT,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--disable-rtnr", action="store_true", help="Disable RTNR noise reduction")
ap.add_argument("-c", "--card", type=int, default=0, help="ALSA card index")
ap.add_argument("--pcm", type=int, default=16, help="Output ALSA PCM index")
ap.add_argument("--cap", type=int, default=18, help="Capture ALSA PCM index")
ap.add_argument("--rate", type=int, default=48000, help="Sample rate")
ap.add_argument("--chan", type=int, default=2, help="Output channel count")
ap.add_argument("--capchan", type=int,
help="Capture channel count (if different from output)")
ap.add_argument("--capbits", type=int, default=16, help="Capture sample bits (16 or 32)")
ap.add_argument("--noise", default="noise.wav",
help="WAV file containing 'noise' for capture")
ap.add_argument("--duration", type=int, default=3, help="Capture duration (seconds)")
ap.add_argument("--chirpcyc", type=int, default=120, help="Repetitions of chirp waveform")
ap.add_argument("--chirp-test", action="store_true",
help="Test latency with synthesized audio")
ap.add_argument("--echo-test", action="store_true", help="Test simultaneous capture/playback")

opts = ap.parse_args()
if not opts.capchan:
opts.capchan = opts.chan
opts.base_test = not (opts.chirp_test or opts.echo_test)

class ALSA:
"""
Tiny ctypes stub. Wraps the alsa API such that errno returns (at
least ones that look like an errno) become OSErrors and don't need
to be checked. Includes a generalized alloc() that wraps all the
_sizeof() predicates and allocates from the (safe/collected) python
heap. Provides a simple spot for putting (manually-derived)
constants. The ALSA C API is mostly-structless and quite simple, so
this tends to work well without a lot of ctypes use except for an
occasional constructed integer or byref() pointer.
"""
PCM_STREAM_PLAYBACK = 0
PCM_STREAM_CAPTURE = 1
PCM_FORMAT_S16_LE = 2
PCM_FORMAT_S32_LE = 10
PCM_ACCESS_MMAP_INTERLEAVED = 0
PCM_ACCESS_RW_INTERLEAVED = 3
def __init__(self):
self.lib = C.cdll.LoadLibrary("libasound.so.2")
def __getattr__(self, name):
fn = getattr(self.lib, name)
if name.endswith("_name"): # These return strings!
fn.restype = C.c_char_p
return lambda *args: fn(*args).decode("utf-8")
return lambda *args: ALSA.err_wrap(fn(*args))
@staticmethod
def err_wrap(ret):
if -200 < ret < 0:
raise OSError(os.strerror(-ret))
return ret
def alloc(self, typ):
return (C.c_byte * getattr(self.lib, f"snd_{typ}_sizeof")())()
class pcm_channel_area_t(C.Structure):
_fields_ = [("addr", C.c_ulong), ("first", C.c_int), ("step", C.c_int)]

def pcm_init_stream(pcm, rate, chans, fmt, access):
hwp = alsa.alloc("pcm_hw_params")
alsa.snd_pcm_hw_params_any(pcm, hwp)
alsa.snd_pcm_hw_params_set_format(pcm, hwp, fmt)
alsa.snd_pcm_hw_params_set_channels(pcm, hwp, chans)
alsa.snd_pcm_hw_params_set_rate(pcm, hwp, rate, alsa.PCM_STREAM_PLAYBACK)
alsa.snd_pcm_hw_params_set_access(pcm, hwp, access)
alsa.snd_pcm_hw_params(pcm, hwp)

def ctl_disable_rtnr():
"""
Noise reduction likes to squash our chirp on capture. Walk the list
of controls, looking for an RTNR enable control, if one exists, and
set it to false. Unbelievably cumbersome API to do this: call
elem_list once on an empty struct to get the element count, then
allocate, then call it again. Then for each element we can check
the name directly, but need to allocate an "id" struct to query an
abstract identifier, that we use with a separately-allocated "value"
(on which we set the dyncmically typed data) to send the command to
the kernel.
"""
dev = f"hw:{opts.card}".encode("ascii")
ctl = C.c_ulong()
alsa.snd_ctl_open(C.byref(ctl), dev, 0)
elist = alsa.alloc("ctl_elem_list")
alsa.snd_ctl_elem_list(ctl, elist)
nelem = alsa.snd_ctl_elem_list_get_count(elist)
alsa.snd_ctl_elem_list_alloc_space(elist, nelem)
alsa.snd_ctl_elem_list(ctl, elist)
for i in range(nelem):
name = alsa.snd_ctl_elem_list_get_name(elist, i)
if re.match(r'RTNR.*\s+rtnr_enable.*', name):
print(f"Disabling control: {name}")
eid = alsa.alloc("ctl_elem_id")
val = alsa.alloc("ctl_elem_value")
alsa.snd_ctl_elem_list_get_id(elist, i, C.byref(eid))
alsa.snd_ctl_elem_value_set_id(val, eid)
alsa.snd_ctl_elem_value_set_boolean(val, 0, False)
alsa.snd_ctl_elem_write(ctl, val)
alsa.snd_ctl_close(ctl)

def pcm_play_buf(data):
data = bytearray(data)
addr = C.addressof((C.c_byte * 1).from_buffer(data))
off = 0
n = int(len(data) / (2 * opts.chan))
n = min(n, opts.rate * opts.duration)

pcm = C.c_long(0)
dev = f"hw:{opts.card},{opts.pcm}".encode("ascii")
alsa.snd_pcm_open(C.byref(pcm), dev, alsa.PCM_STREAM_PLAYBACK, 0)
pcm_init_stream(pcm, opts.rate, opts.chan, alsa.PCM_FORMAT_S16_LE,
alsa.PCM_ACCESS_RW_INTERLEAVED)
while n > 0:
f = alsa.snd_pcm_writei(pcm, C.c_ulong(addr + off), n)
n -= f
off += f
alsa.snd_pcm_drain(pcm)
alsa.snd_pcm_close(pcm)

def pcm_play_chirp():
pcm = C.c_long(0)
dev = f"hw:{opts.card},{opts.pcm}".encode("ascii")
alsa.snd_pcm_open(C.byref(pcm), dev, alsa.PCM_STREAM_PLAYBACK, 0)
pcm_init_stream(pcm, opts.rate, opts.chan, alsa.PCM_FORMAT_S16_LE,
alsa.PCM_ACCESS_MMAP_INTERLEAVED)

(chirp, chirp_frames) = gen_chirp_s16le()

# Reset the stream and queue up as much data as will fit in the
# ring buffer
area = alsa.pcm_channel_area_t()
offset = C.c_ulong()
frames = C.c_ulong(opts.rate)
ring_frames = 0
alsa.snd_pcm_prepare(pcm)
alsa.snd_pcm_reset(pcm)
while True:
alsa.snd_pcm_avail_update(pcm)
alsa.snd_pcm_mmap_begin(pcm, C.byref(area), C.byref(offset), C.byref(frames))
committed = alsa.snd_pcm_mmap_commit(pcm, offset, frames)
ring_frames += committed
if committed == 0:
break

silence = bytes(2 * opts.chan * ring_frames)

# Start up the stream, spin until there is space in the buffer,
# write the chirp. This minimizes client-side overhead like
# stream startup. Then immediately take a timestamp and write
# silence for one full cycle (to be 100% sure the buffer can't
# wrap and chirp twice).
alsa.snd_pcm_start(pcm)
while alsa.snd_pcm_avail(pcm) < chirp_frames:
pass
pre_buffered = ring_frames - alsa.snd_pcm_avail(pcm)
f = alsa.snd_pcm_mmap_writei(pcm, chirp, chirp_frames)
chirp_sent = time.perf_counter()
assert f == chirp_frames

n = 0
while n < ring_frames:
n += alsa.snd_pcm_mmap_writei(pcm, silence, ring_frames)
alsa.snd_pcm_drain(pcm)
alsa.snd_pcm_close(pcm)

# Correct chirp_sent for buffered data!
chirp_sent += pre_buffered / opts.rate
return chirp_sent

def pcm_do_capture(duration):
"""
Returns an array of tuples of (timestamp, bytes), no processing done
here for performance reasons, just one heap allocation and copy.
"""
pcm = C.c_long(0)
fmt = alsa.PCM_FORMAT_S32_LE if opts.capbits == 32 else alsa.PCM_FORMAT_S16_LE
capsz = 4 if opts.capbits == 32 else 2
dev = f"hw:{opts.card},{opts.cap}".encode("ascii")
alsa.snd_pcm_open(C.byref(pcm), dev, alsa.PCM_STREAM_CAPTURE, 0)
pcm_init_stream(pcm, opts.rate, opts.capchan, fmt, alsa.PCM_ACCESS_RW_INTERLEAVED)
frames_remaining = duration * opts.rate
buf_frames = int(opts.rate / 1000) # 1ms blocks
fsz = opts.capchan * capsz
buf = bytearray(fsz * buf_frames)
addr = C.c_ulong(C.addressof((C.c_byte * 1).from_buffer(buf)))
buflist = []
buf_frames = C.c_ulong(buf_frames)
while frames_remaining > 0:
f = alsa.snd_pcm_readi(pcm, addr, buf_frames)
t = time.perf_counter()
frames_remaining -= f
buflist.append((t, bytes(buf[0:f * fsz])))
return buflist

def gen_chirp_s16le():
"""
A programmatically-detectable chirp/pop signal for testing latency.
To minimize latency, we want the chirp to be low duration, high
energy and high frequency. This repeats an 8-sample square wave (6
kHz at 48k sample rate). Some devices can reproduce this well with
as few as 8 repetitions (1.3ms), but on at least one mt8195 device
it's unreliably audible unless repeated 128 times! It's not caused
by software in the DSP, more like a codec/amp feature (possibly
related to power management, if we don't play other audio
immediately before, it's even less reliable).
"""
reps = 4
chirp = b''
for _ in range(opts.chirpcyc):
n = opts.chan * reps
vals = [-0x8000] * n + [0x7fff] * n
chirp += struct.pack(f"{2*n}h", *vals)
return (chirp, opts.chirpcyc * reps)

def cap_to_playback(buf):
"""
Converts a byte array containing capture frames (which can have
different sample format and channel count) to the playback format
(always s16_le). Also computes an "energy" value as the sum of
absolute sample differences (in units of +/-1.0) over all result
channels. Returns both as a tuple.
"""
capfmt = ('i' if opts.capbits == 32 else 'h') * opts.capchan
capsz = opts.capchan * (4 if opts.capbits == 32 else 2)
scale = 1 / (1 << (opts.capbits - 1))
last_frame = []
delta_sum = 0
out_frames = []

# NOTE: should consider low-passing the energy computation by
# averaging ~N recent samples. Otherwise high frequency noise can
# dominate, which we don't really care about measuring (AEC can't
# treat it, and it can plausibly create false positive chirp signals
# loud enough).
for i in range(0, len(buf), capsz):
frame = [scale * x for x in struct.unpack(capfmt, buf[i:i+capsz])[0:opts.chan]]
if last_frame:
delta_sum += sum(abs(last_frame[x] - frame[x]) for x in range(opts.chan))
last_frame = frame
iframe = [int(min(0x7fff, max(-0x8000, (1 << 15) * e))) for e in frame]
out_frames.append(struct.pack(f'{opts.chan}h', *iframe))
return (b''.join(out_frames), delta_sum)

def chirp_child(wpipe):
for rec in pcm_do_capture(opts.duration):
t = rec[0]
(buf, energy) = cap_to_playback(rec[1])
frames = len(buf) / (2 * opts.chan)

# Normalize energy as "half-swing per sample" and check vs. a
# threshold that will trigger if we get a 0.1 unit swing over
# the 8-sample chirp waveform.
#
# NOTE: would be possible to do this analysis at the
# individual sample layer for better time fidelity instead of
# in 1ms chunks.
energy = energy / (frames * opts.chan)
if energy > (0.1/8):
os.write(wpipe, f"{t}".encode("ascii"))
return

def echo_child(wpipe):
energy = 0
for rec in pcm_do_capture(opts.duration):
energy += cap_to_playback(rec[1])[1]

# Normalize energy to "half-swing per second" here, just to make
# essentially arbitrary numbers prettier (e.g. a typical pop music
# track results in ~few-hundred values for "energy")
energy /= (opts.duration * opts.chan)
os.write(wpipe, f"{energy:.3f}".encode("ascii"))

def chirp_test():
"""
Forks a child process to listen for the chirp and write back a
time.perf_counter() value (which is an invariant clock across
processes) through a pipe.
"""
(rfd, wfd) = os.pipe()
pid = os.fork()
if pid == 0:
chirp_child(wfd)
sys.exit(0)

# Randomly sleep for a bit to make aliasing bugs (e.g. noise being
# detected as a chirp) visible as unreliable output.
time.sleep(random.randint(1000, 2000)/1000)
chirp_sent = pcm_play_chirp()

os.waitpid(pid, 0)
msg = os.read(rfd, 9999).decode("ascii")
chirp_detected = float(msg)

lat_ms = (chirp_detected - chirp_sent) * 1000
print(f"Chirp latency: {lat_ms:.1f} ms")

def echo_test():
"""
Similar to chirp test, but plays a .wav file while the child
captures, and reports average capture energy (useful for testing mic
gain and echo cancellation performance)
"""
# Just slurps in the wav file and chops off the header, assuming
# the user got the format and sampling rate correct.
WAV_HDR_LEN = 44
buf = open(opts.noise, "rb").read()[WAV_HDR_LEN:]

(rfd, wfd) = os.pipe()
pid = os.fork()
if pid == 0:
echo_child(wfd)
sys.exit(0)

pcm_play_buf(buf)

os.waitpid(pid, 0)
msg = os.read(rfd, 9999).decode("ascii")
print("Capture energy:", msg)

def base_test():
"""
Simplest test: Just capture opts.duration seconds worth of data,
convert to playback format, and play it.
"""
bufs = []
energy = 0
for rec in pcm_do_capture(opts.duration):
crec = cap_to_playback(rec[1])
bufs.append(crec[0])
energy += crec[1]
pcm_play_buf(b''.join(bufs))
print(f"Energy {energy}")

def main():
parse_opts()
if opts.disable_rtnr:
ctl_disable_rtnr()
if opts.base_test:
base_test()
if opts.chirp_test:
chirp_test()
if opts.echo_test:
echo_test()

opts = None
alsa = ALSA()
if __name__ == "__main__":
main()

0 comments on commit bc3ced5

Please sign in to comment.