I'm working on driving an analog VU
meter</a> from my Raspberry
Pi</a> using whatever audio is going out the
Pi's sound outputs. The de facto Linux sound system,
PulseAudio</a>,
allows any sound output (or "sink" in PulseAudio's nonclementure) to be
monitored. In PulseAudio land, each sink has a corresponding "source"
called the monitor source which can be read just like any other other
PulseAudio input such as a microphone. In fact, to help with volume
meter style applications, PulseAudio even allows you to ask for peak
level measurements, which means you can sample the monitor sink at a low
frequency, with low CPU utilisation, but still produce a useful volume
display. When this feature is used, each sample read indicates the peak
level since the last sample.</p>
The main PulseAudio API is asynchronous and callback based, and the
documentation</a>
is primarly just an API reference. This makes it a little difficult to
figure out how to get everthing to hang together. Using the code from
various open source projects (primarily
veromix-plasmoid</a> and
pavumeter</a>), along with
the API reference, I was able to develop a fairly minimal code example
that will hopefully be useful to others trying to do something similar.
Although this example is written in Python, it is using the PulseAudio C
API directly (via ctypes) so it should hopefully still be relevant if
your application is written in C or another language.</p>
Here's the demo code. The latest version is also available on
GitLab</a>. Note that in order to run
this example you need to install Vincent Breitmoser's ctypes PulseAudio
wrapper available at https://github.com/Valodim/python-pulseaudio</a>.</p>
import</span> sys</span></span>
from</span> Queue</span> import</span> Queue</span></span>
from</span> ctypes</span> import</span> POINTER</span>, c_ubyte, c_void_p, c_ulong, cast</span></span>
</span>
# From https://github.com/Valodim/python-pulseaudio</span></span>
from</span> pulseaudio.lib_pulseaudio</span> import *</span></span>
</span>
# edit to match your sink</span></span>
SINK_NAME</span> =</span> 'alsa_output.pci-0000_00_1b.0.analog-stereo'</span></span>
METER_RATE</span> =</span> 344</span></span>
MAX_SAMPLE_VALUE</span> =</span> 127</span></span>
DISPLAY_SCALE</span> =</span> 2</span></span>
MAX_SPACES</span> =</span> MAX_SAMPLE_VALUE</span> >></span> DISPLAY_SCALE</span></span>
</span>
class</span> PeakMonitor</span>(</span>object</span>):</span></span>
</span>
def</span> __init__</span>(self, sink_name, rate):</span></span>
self</span>.sink_name</span> =</span> sink_name</span></span>
self</span>.rate</span> =</span> rate</span></span>
</span>
# Wrap callback methods in appropriate ctypefunc instances so</span></span>
# that the Pulseaudio C API can call them</span></span>
self</span>._context_notify_cb</span> =</span> pa_context_notify_cb_t(</span>self</span>.context_notify_cb)</span></span>
self</span>._sink_info_cb</span> =</span> pa_sink_info_cb_t(</span>self</span>.sink_info_cb)</span></span>
self</span>._stream_read_cb</span> =</span> pa_stream_request_cb_t(</span>self</span>.stream_read_cb)</span></span>
</span>
# stream_read_cb() puts peak samples into this Queue instance</span></span>
self</span>._samples</span> =</span> Queue()</span></span>
</span>
# Create the mainloop thread and set our context_notify_cb</span></span>
# method to be called when there's updates relating to the</span></span>
# connection to Pulseaudio</span></span>
_mainloop</span> =</span> pa_threaded_mainloop_new()</span></span>
_mainloop_api</span> =</span> pa_threaded_mainloop_get_api(_mainloop)</span></span>
context</span> =</span> pa_context_new(_mainloop_api,</span> 'peak_demo'</span>)</span></span>
pa_context_set_state_callback(context,</span> self</span>._context_notify_cb,</span> None</span>)</span></span>
pa_context_connect(context,</span> None</span>,</span> 0</span>,</span> None</span>)</span></span>
pa_threaded_mainloop_start(_mainloop)</span></span>
</span>
def</span> __iter__</span>(self):</span></span>
while</span> True</span>:</span></span>
yield</span> self</span>._samples.get()</span></span>
</span>
def</span> context_notify_cb</span>(self, context, _):</span></span>
state</span> =</span> pa_context_get_state(context)</span></span>
</span>
if</span> state</span> ==</span> PA_CONTEXT_READY</span>:</span></span>
print</span> "Pulseaudio connection ready..."</span></span>
# Connected to Pulseaudio. Now request that sink_info_cb</span></span>
# be called with information about the available sinks.</span></span>
o</span> =</span> pa_context_get_sink_info_list(context,</span> self</span>._sink_info_cb,</span> None</span>)</span></span>
pa_operation_unref(o)</span></span>
</span>
elif</span> state</span> ==</span> PA_CONTEXT_FAILED</span> :</span></span>
print</span> "Connection failed"</span></span>
</span>
elif</span> state</span> ==</span> PA_CONTEXT_TERMINATED</span>:</span></span>
print</span> "Connection terminated"</span></span>
</span>
def</span> sink_info_cb</span>(self, context, sink_info_p, _, __):</span></span>
if not</span> sink_info_p:</span></span>
return</span></span>
</span>
sink_info</span> =</span> sink_info_p.contents</span></span>
print</span> '-'</span>*</span> 60</span></span>
print</span> 'index:'</span>, sink_info.index</span></span>
print</span> 'name:'</span>, sink_info.name</span></span>
print</span> 'description:'</span>, sink_info.description</span></span>
</span>
if</span> sink_info.name</span> ==</span>
This page cannot be shown here. You can still read it on the original site — the toolbar below keeps your place in the directory.
I'm working on driving an analog VU meter from my Raspberry Pi using whatever audio is going out the Pi's sound outputs. The de facto Linux sound system, PulseAudio , allows any sound output (or "sink" in PulseAudio's nonclementure) to be monitored. In PulseAudio land, each sink has a corresponding "source" called the monitor source which can be read just like any other other…