Generating Audio Spectrograms in Python

A spectrogram is a visual representation of the spectrum of frequencies in a sound sample.

more info: wikipedia spectrogram

Spectrogram code in Python, using Matplotlib:
(source on GitHub)

"""Generate a Spectrogram image for a given WAV audio sample.

A spectrogram, or sonogram, is a visual representation of the spectrum
of frequencies in a sound.  Horizontal axis represents time, Vertical axis
represents frequency, and color represents amplitude.
"""


import os
import wave

import pylab


def graph_spectrogram(wav_file):
    sound_info, frame_rate = get_wav_info(wav_file)
    pylab.figure(num=None, figsize=(19, 12))
    pylab.subplot(111)
    pylab.title('spectrogram of %r' % wav_file)
    pylab.specgram(sound_info, Fs=frame_rate)
    pylab.savefig('spectrogram.png')


def get_wav_info(wav_file):
    wav = wave.open(wav_file, 'r')
    frames = wav.readframes(-1)
    sound_info = pylab.fromstring(frames, 'Int16')
    frame_rate = wav.getframerate()
    wav.close()
    return sound_info, frame_rate


if __name__ == '__main__':
    wav_file = 'sample.wav'
    graph_spectrogram(wav_file)

Spectrogram code in Python, using timeside:
(source on GitHub)

"""Generate a Spectrogram image for a given audio sample.

Compatible with several audio formats: wav, flac, mp3, etc.
Requires: https://code.google.com/p/timeside/

A spectrogram, or sonogram, is a visual representation of the spectrum
of frequencies in a sound.  Horizontal axis represents time, Vertical axis
represents frequency, and color represents amplitude.
"""


import timeside


audio_file = 'sample.wav'

decoder = timeside.decoder.FileDecoder(audio_file)
grapher = timeside.grapher.Spectrogram(width=1920, height=1080)
(decoder | grapher).run()
grapher.render('spectrogram.png')

happy audio hacking.

Python - concurrencytest: Running Concurrent Tests

Add parallel testing to your unit test framework.

In my previous post, I described running concurrent tests using nose as a loader and runner.

On a similar note, let's look at building concurrency into your own test framework built on Python's unittest.

Have a look at this module: concurrencytest

(Thanks to bits and concepts taken from testtools and bzrlib)

---

An Example:

Say you have a 'TestSuite' of tests loaded. You could run them with the standard 'TextTestRunner' like this:

runner = unittest.TextTestRunner()
runner.run(suite)

That would run the tests in your suite sequentially in a single process.

By adding the concurrencytest module, you can use a 'ConcurrentTestSuite' instead, by adding:

from concurrencytest import ConcurrentTestSuite, fork_for_tests

concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(4))
runner.run(concurrent_suite)

That would run the same tests split across 4 processes (workers).

Note: this relies on 'os.fork()' which only works on Unix systems.

---

There's no way to understand this better than looking at some contrived examples!

This first example is totally unrealistic, but shows off concurrency perfectly. The test cases it loads each sleep for 0.5 seconds and then exit.

The Code:

	#!/usr/bin/env python
	#
	# Example using `concurrencytest`:
	# https://github.com/cgoldberg/concurrencytest
	import time
	import unittest
	from concurrencytest import ConcurrentTestSuite, fork_for_tests
	class SampleTestCase(unittest.TestCase):
	def test_it(self):
	time.sleep(0.5)
	if __name__ == '__main__':
	# load a TestSuite with 50x TestCases for demo
	loader = unittest.TestLoader()
	suite = unittest.TestSuite()
	for _ in range(50):
	suite.addTests(loader.loadTestsFromTestCase(SampleTestCase))
	print('Loaded %d test cases...' % suite.countTestCases())
	runner = unittest.TextTestRunner()
	print('\nRun tests sequentially:')
	runner.run(suite)
	print('\nRun same tests across 50 processes:')
	concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(50))
	runner.run(concurrent_suite)

view raw concurrencytest_example_1.py hosted with ❤ by GitHub

Output:

Loaded 50 test cases...

Run tests sequentially:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 25.031s

OK

Run same tests across 50 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 0.525s

OK

nice!

Now another example that shows concurrency with CPU-bound test cases. The test cases it loads each calculate fibonacci of 31 (recursively!) and then exit. We can see how it performs on my 8-core machine (Core2 i7 quad, hyperthreaded).

The Code:

	#!/usr/bin/env python
	#
	# Example using `concurrencytest`:
	# https://github.com/cgoldberg/concurrencytest
	import unittest
	from concurrencytest import ConcurrentTestSuite, fork_for_tests
	def fib(n):
	if n == 0:
	return 0
	elif n == 1:
	return 1
	return fib(n - 1) + fib(n - 2)
	class FibonacciTestCase(unittest.TestCase):
	def test_fib(self):
	self.assertEqual(fib(31), 1346269)
	if __name__ == '__main__':
	# load a TestSuite with 50x TestCases for demo
	loader = unittest.TestLoader()
	suite = unittest.TestSuite()
	for _ in range(50):
	suite.addTests(loader.loadTestsFromTestCase(FibonacciTestCase))
	print('Loaded %d test cases...' % suite.countTestCases())
	runner = unittest.TextTestRunner()
	print('\nRun tests sequentially:')
	runner.run(suite)
	print('\nRun same tests with 2 processes:')
	concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(2))
	runner.run(concurrent_suite)
	print('\nRun same tests with 4 processes:')
	concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(4))
	runner.run(concurrent_suite)
	print('\nRun same tests with 8 processes:')
	concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(8))
	runner.run(concurrent_suite)

view raw concurrencytest_example_2.py hosted with ❤ by GitHub

Output:

Loaded 50 test cases...

Run tests sequentially:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 21.941s

OK

Run same tests with 2 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 11.081s

OK

Run same tests with 4 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 5.862s

OK

Run same tests with 8 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 4.743s

OK

---

happy hacking.

Python - Nose: Running Concurrent Tests

TLDR:
To enable multiprocessing with N workers,
run nose with:

$ nosetests --processes=N

---

When writing tests in Python, I start with TestCase's derived from unittest.TestCase, and standard test discovery. When I need more complex test discovery/loading or output reports, I often use nose and its assortment of plugins as my test loader/runner.

One nice feature of nose is the multiprocess plugin. It allows you to run your tests suites concurrently rather than sequentially, spread across a number of worker processes. Running tests in parallel like this can potentially give you a large speedup in your test run times.

from the nose multiprocess docs:

"You can parallelize a test run across a configurable number of worker processes. While this can speed up CPU-bound test runs, it is mainly useful for IO-bound tests that spend most of their time waiting for data to arrive from someplace else and can benefit from parallelization."

Normally, you run tests from nose with:

$ nosetests

To run the same tests split across 4 processes (workers), you would just do:

$ nosetests --processes=4

Assuming your tests are properly isolated, everything should run normally, and you can benefit from a speedup on a multiprocessor machine.

However, Beware.

"Not all test suites will benefit from, or even operate correctly using, this plugin. For example, CPU-bound tests will run more slowly if you don't have multiple processors."

"But the biggest issue you will face is probably concurrency. Unless you have kept your tests as religiously pure unit tests, with no side-effects, no ordering issues, and no external dependencies, chances are you will experience odd, intermittent and unexplainable failures and errors when using this plugin. This doesn't necessarily mean the plugin is broken; it may mean that your test suite is not safe for concurrency."