Rust gtk+portaudio

extern crate chrono;
extern crate gtk;
extern crate gdk;
extern crate portaudio;

use gtk::traits::*;
use gtk::signal::Inhibit;
use std::thread;
use std::collections::VecDeque;
use chrono::UTC;
use std::time::Duration;
use std::sync::mpsc::{Sender, Receiver};
use std::sync::mpsc;
use std::sync::Arc;
use std::sync::Mutex;

const SAMPLE_RATE: i64 = 44_100;
const CHANNELS: i32 = 2;
const FRAMES: u32 = 1024;
const INTERLEAVED: bool = true;
const LATENCY: portaudio::Time = 0.0; // Ignored by PortAudio::is_*_format_supported.

struct IPData {
    terminate: bool
}

impl IPData {
    pub fn new() -> IPData {
        IPData {
            terminate: false
        }
    }
}


fn main() {
    // During construction, call unwrap() on all Results, since there's no
    // alternative to printing an error message and stopping anyway.
    let pa = portaudio::PortAudio::new().unwrap();

    // Share ip_data between the main event loop and the audio thread
    let ip_data = Arc::new(Mutex::new(IPData::new()));

    // GTK initialization
    if gtk::init().is_err() {
        println!("Failed to initialize GTK.");
        return;
    }

    // Construct window,
    let (window, avg_level_bar, peak_level_bar) = create_main_window(&pa);
    window.show_all();
    // Make close window set flag which terminates all loops
    let ip_clone = ip_data.clone();
    window.connect_delete_event(move |_, _| {
        ip_clone.lock().unwrap().terminate = true;
        Inhibit(false)
    });

    // Set up communication channel between audio thread and main loop
    let (tx, rx): (Sender<(f32,f32)>, Receiver<(f32,f32)>) = mpsc::channel();

    let ip_clone = ip_data.clone();
    let analysis_thread = thread::spawn(move || {
        match start_analysis_thread(pa, &ip_clone, tx) {
            Err(err) => {
                println!("PortAudio error: {}", err);
            },
            _ => ()
        }
    });

    // Show peak audio signal with a decay instead of changing it abruptly
    let mut peak_decay: f32 = 0.0;

    // Main event loop.
    'eventloop: while !ip_data.lock().unwrap().terminate {
        while gtk::events_pending() {
            if !gtk::main_iteration() {
                break 'eventloop; // never happens as no-one calls the quit function
            }
        }
        // Receive power and peak value from the other thread. This takes around
        // 23ms, which is an acceptable delay for an event loop.
        let (power, peak) = rx.recv().unwrap();
        if peak > peak_decay {
            peak_decay = peak;
        } else {
            peak_decay = (3.0 * peak_decay + peak) / 4.0;
        }
        avg_level_bar.set_value(power as f64);
        peak_level_bar.set_value(peak_decay as f64);
    }

    // Wait for termination of audio thread
    analysis_thread.join().unwrap();
}

// Creates the gtk window with a few elements and returns a handle to the
// window and two elements that will be updated from the main loop
fn create_main_window(pa: &portaudio::PortAudio) -> (gtk::Window, gtk::LevelBar, gtk::LevelBar) {
    let window = gtk::Window::new(gtk::WindowType::Toplevel).unwrap();

    window.set_title("Realtime FFT Analyzer");
    window.set_default_size(350, 70);
    window.set_border_width(10);
    window.set_window_position(gtk::WindowPosition::Center);

    let avg_level_bar = gtk::LevelBar::new_for_interval(0., 100.).unwrap();
    avg_level_bar.set_value(0.);
    let peak_level_bar = gtk::LevelBar::new_for_interval(0., 100.).unwrap();
    peak_level_bar.set_value(0.);
    let scale_low = gtk::Scale::new_with_range(gtk::Orientation::Horizontal, 0., 2000., 10.).unwrap();
    let scale_high = gtk::Scale::new_with_range(gtk::Orientation::Horizontal, 0., 2000., 10.).unwrap();

    let _box = gtk::Box::new(gtk::Orientation::Vertical, 10).unwrap();
    _box.set_border_width(5);
    _box.add(&create_input_interface_combo_box(&pa));
    _box.add(&avg_level_bar);
    _box.add(&peak_level_bar);
    _box.add(&scale_low);
    _box.add(&scale_high);
    window.add(&_box);
    (window, avg_level_bar, peak_level_bar)
}

// Creates combo box with all audio inputs (ignoring half duplex output only
// interfaces) and selects the system default in the list.
fn create_input_interface_combo_box(pa: &portaudio::PortAudio) -> gtk::ComboBoxText {
    let default_device_index = pa.default_input_device().unwrap();
    let interface_list = gtk::ComboBoxText::new().unwrap();
    let mut def_dev_idx: i32 = -1;
    let mut menu_index: i32 = 0;

    for device in pa.devices().unwrap() {
        let (idx, info) = device.unwrap();
        let in_channels = info.max_input_channels;
        let input_params = portaudio::StreamParameters::<i16>::new(idx, in_channels, INTERLEAVED, LATENCY);
        let out_channels = info.max_output_channels;
        let output_params = portaudio::StreamParameters::<i16>::new(idx, out_channels, INTERLEAVED, LATENCY);
        let default_sample_rate = info.default_sample_rate;
        if pa.is_input_format_supported(input_params, default_sample_rate).is_ok() ||
           pa.is_duplex_format_supported(input_params, output_params, default_sample_rate).is_ok() {
            interface_list.append(info.name, info.name);
            if idx == default_device_index {
                def_dev_idx = menu_index;
            }
            menu_index += 1;
        }
    }
    interface_list.set_active(def_dev_idx);
    interface_list
}

// Open a stream on the default input device, and keep reading until the flag
// in ip_data tells us to stop. For every received buffer, compute some form
// of running average of the power and the current peak and send them to the
// main loop.
// Note: the loop sleeps after receiving a full buffer, since the blocking
// stream doesn't block in reality.
fn start_analysis_thread(pa: portaudio::PortAudio, ip_data: &Arc<Mutex<IPData>>, tx: Sender<(f32,f32)>) -> Result<(), portaudio::Error> {

    let def_input = try!(pa.default_input_device());
    let input_info = try!(pa.device_info(def_input));

    // Construct the input stream parameters.
    let latency = input_info.default_low_input_latency;
    let input_params = portaudio::StreamParameters::<f32>::new(def_input, CHANNELS, INTERLEAVED, latency);

    // Construct the settings with which we'll open our duplex stream.
    let settings = portaudio::InputStreamSettings::new(input_params, SAMPLE_RATE as f64, FRAMES);

    let mut stream = try!(pa.open_blocking_stream(settings));

    // We don't buffer yet.
    // let mut buffer: VecDeque<f32> = VecDeque::with_capacity(FRAMES as usize * CHANNELS as usize);

    // We'll use this function to wait for read/write availability.
    fn wait_for_stream<F>(f: F, name: &str) -> u32
        where F: Fn() -> Result<portaudio::StreamAvailable, portaudio::error::Error>
    {
        'waiting_for_stream: loop {
            match f() {
                Ok(available) => match available {
                    portaudio::StreamAvailable::Frames(frames) => return frames as u32,
                    portaudio::StreamAvailable::InputOverflowed => println!("Input stream has overflowed"),
                    portaudio::StreamAvailable::OutputUnderflowed => println!("Output stream has underflowed"),
                },
                Err(err) => panic!("An error occurred while waiting for the {} stream: {}", name, err),
            }
        }
    };

    let interval = chrono::Duration::microseconds((1000000 as i64 * FRAMES as i64 + SAMPLE_RATE - 1) / SAMPLE_RATE);
    let mut next = UTC::now() + interval + interval;
    let mut do_sleep: bool = true;

    let mut ring_buffer: VecDeque<f32> = VecDeque::with_capacity(25);
    let mut running_sum: f32 = 0.0;

    // let max_noise_power: f32 = -9.5424; // average over power of all possible values from (-32768..+32767)/32768 in dB
    // let max_per_buffer: f32 = (FRAMES as f32).log10() * 20.0 + max_noise_power;

    try!(stream.start());

    'stream: while !ip_data.lock().unwrap().terminate {

        if do_sleep {
            // If the sleep flag is set, sleep until the time indicated by
            // next. This is needed since the blocking stream doesn't block.
            let wait_in_ms = (next - UTC::now()).num_milliseconds() as u64;
            thread::sleep(Duration::from_millis(wait_in_ms + 1));
            do_sleep = false;
        }

        // How many frames are available on the input stream?
        let in_frames = wait_for_stream(|| stream.read_available(), "Read");

        // If there are frames available, let's take them and add them to our buffer.
        if in_frames > 0 {

            let input_samples = try!(stream.read(in_frames));
            // buffer.extend(input_samples.into_iter());
            let mut peak: f32 = 0.0;
            // Compute the average of all x[i]^2
            let energy = input_samples.into_iter().
                fold(0.0, |acc_power, &sample| {
                    let abs = sample.abs();
                    if abs > peak {
                        peak = abs;
                    };
                    sample * sample + acc_power
                });
            let power: f32 = energy / FRAMES as f32;

            // Use a ring buffer to compute the running average over 25 buffers
            // (about .5s with the current constants).
            // TODO: Doesn't seem to work as intended.
            if ring_buffer.len() == 25 {
                running_sum -= ring_buffer.pop_front().unwrap_or(0.0);
            }
            running_sum += power;
            ring_buffer.push_back(power);

            // Convert to dB, but add some large number to make sure it's
            // visible.
            let db: f32 = 120.0 + (running_sum / ring_buffer.len() as f32).log10() * 20.0; // - max_per_buffer ??

            // Send values to main loop so they will be shown in the UI
            match tx.send((db, 100.0 * peak)) {
                Ok(()) => (),
                Err(_) => {
                    break;
                }
            }
            // buffer.clear();

            next = UTC::now() + interval;
            do_sleep = true;

        }

    }

    try!(stream.stop());
    Ok(())
}