Analyzing data from continuous input stream (recording) in Python, multiprocessing?

Question

I am analyzing data coming from recording with my microphone in real-time. So far I have been doing this in a linear fashion:

• Record one second (takes 1s)
• Analyze the data (takes for example 50 ms)
• Record one second
• Analyze the data

And so forth. This obviously means that while I'm analyzing the data from the past second, I am losing this 50 ms of time, I won't be recording the sounds during it.

I thought multiprocessing would be the solution: I start a seperate proces that non-stop records in certain length chunks and each time sends it through a pipe to the the main proces, which then analyzes the data. Unfortunately, sending a lot of data through a pipe (or in general, sending a lot of data from one proces to another), is far from ideal apparently. Is there any other way to do this? I just want my computer to record data and import it into python (all of which I'm already doing), while it's also analyzing data.

If I need to add any more details, let me know!

Thanks!

Answer 1

Simple producer/consumer implementation.

While it's true moving data back and forth induces overhead and increases memory use, as long as the same data is not needed by more than one process that overhead is minimal. Try it and find out :) Can adjust memory footprint by changing the queue and pool size numbers.

Threading is another option to reduce memory use but at the expense of being blocked on the GIL and effectively single threaded if the processing is in python bytecode.

import multiprocessing
# Some fixed size to avoid run away memory use
recorded_data = multiprocessing.Queue(100)

def process(recorded_data):
    while True:
        data = recorded_data.get()
        <process data>

def record(recorded_data):
    for data in input_stream:
        recorded_data.put(data)

producer = multiprocessing.Process(target=record, 
                                   args=(recorded_data,))
producer.start()

# Pool of 10 processes
num_proc = 10
consumer_pool = multiprocessing.Pool(num_proc)
results = []
for _ in xrange(num_proc):
    results.append(
      consumer_pool.apply_async(process,
                                args=(recorded_data,)))

producer.join()

# If processing actually returns something
for result in results:
    print result
# Consumers wait for data from queue forever
#  so terminate them when done
consumer_pool.terminate()