dtmf-audio-commands/src/main.rs

mod server;

use anyhow::Result;
use gst::glib;
use gst::glib::once_cell::sync::Lazy;
use gst::prelude::*;
use std::thread;
use std::thread::sleep;
use rand::prelude::*;
use tokio::runtime::Builder;

static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
    gst::DebugCategory::new("main", gst::DebugColorFlags::empty(), Some("Main function"))
});

#[derive(Debug, Clone)]
struct DtmfEvent {
    number: Option<i32>,
    volume: Option<i32>,
    start: bool,
}

impl DtmfEvent {
    fn from_number(number: i32) -> Self {
        Self { number: Some(number), volume: Some(36), start: true }
    }

    fn end(&self) -> Self {
        Self { number: None, volume: None, start: false }
    }
}

impl TryFrom<&gst::StructureRef> for DtmfEvent {
    type Error = anyhow::Error;

    fn try_from(structure: &gst::StructureRef) -> anyhow::Result<Self> {
        let name = structure.name().to_string();
        if !name.starts_with("dtmf-event") {
            anyhow::bail!("Not a dtmf-event structure: {name}");
        }
        let number = structure.get_optional::<i32>("number")?;
        let volume = structure.get_optional::<i32>("volume")?;
        let start = structure.get::<bool>("start")?;
        Ok(Self { number, volume, start })
    }
}

impl From<DtmfEvent> for gst::Structure {
    fn from(event: DtmfEvent) -> Self {
        let mut st = gst::Structure::builder("dtmf-event")
            .field("type", 1)
            .field("start", event.start)
            .build();
        if let Some(number) = event.number {
            st.set("number", number);
        }
        if let Some(volume) = event.volume {
            st.set("volume", volume);
        }
        st
    }
}

fn main() -> Result<()> {
    gst::init()?;
    //videotestsrc ! videoconvert ! timeoverlay shaded-background=true ! gtksink

    let pipeline = gst::parse_launch(
        r#"

    dtmfsrc name=src ! mix.
    audiotestsrc freq=20 ! audiomixer name=mix ! tee name=t ! queue ! audioconvert ! autoaudiosink name=audiosink

    "#,
    )?
    .downcast::<gst::Pipeline>()
    .unwrap();
    // t.! queue ! audioconvert ! dtmfdetect

    let context = glib::MainContext::default();
    let main_loop = glib::MainLoop::new(Some(&context), false);

    pipeline.set_state(gst::State::Playing)?;

    let bus = pipeline.bus().unwrap();
    bus.add_watch({
        let main_loop = main_loop.clone();
        move |_, msg| {
            use gst::MessageView;
            let main_loop = &main_loop;
            match msg.view() {
                MessageView::Eos(..) => main_loop.quit(),
                MessageView::Error(err) => {
                    gst::error!(CAT, obj: &err.src().unwrap(),
                        "Error from {:?}: {} ({:?})",
                        err.src().map(|s| s.path_string()),
                        err.error(),
                        err.debug()
                    );
                    main_loop.quit();
                },
                MessageView::Element(element) => {
                    let dtmf_event = match DtmfEvent::try_from(element.structure().unwrap()) {
                        Ok(ev) => ev,
                        Err(err) => {
                            gst::error!(CAT, "Failed to parse DTMF event {:?} with error: {:?}", element.structure().unwrap(), err);
                            return glib::Continue(true);
                        }
                    };
                    match element.structure().unwrap().name().as_str() {
                        "dtmf-event" => {
                            gst::info!(CAT, "Detected DTMF event");
                        }
                        "dtmf-event-processed" => {
                            if dtmf_event.start {
                                gst::info!(CAT, "DTMF event processed: {:?}", dtmf_event);
                            } else {
                                gst::info!(CAT, "Ending DTMF event processed");
                            }
                        }
                        _ => {
                            gst::info!(CAT, "Received unknown event: {:?}", element.structure().unwrap().name());
                        }
                    }
                }
                _ => (),
            };
            glib::Continue(true)
        }
    })
    .expect("Failed to add bus watch");

    thread::spawn({
        let pipeline_weak = pipeline.downgrade();
        move || {
            let Some(pipeline) = pipeline_weak.upgrade() else {
                gst::error!(CAT, "Pipeline gone...");
                return;
            };

            let source = pipeline.by_name("src").unwrap();

            let mut rng = rand::thread_rng();
            loop {
                let dtmf_event = DtmfEvent::from_number(rng.gen_range(0..15));

                source.send_event(gst::event::CustomUpstream::new(dtmf_event.clone().into()));
                gst::info!(CAT, "Sent DTMF event {:?}", dtmf_event);

                sleep(std::time::Duration::from_millis(10));

                source.send_event(gst::event::CustomUpstream::new(dtmf_event.end().into()));

                sleep(std::time::Duration::from_millis(1000));
            }
        }
    });

    thread::spawn({
        let pipeline_weak = pipeline.downgrade();
        move || {
            let runtime = Builder::new_multi_thread()
                .worker_threads(2)
                .thread_name("http-server")
                .enable_all()
                .build()
                .unwrap();
            runtime.block_on(server::run(8080, pipeline_weak))
        }
    });

    ctrlc::set_handler({
        let main_loop = main_loop.clone();
        move || {
            main_loop.quit();
        }
    })?;

    main_loop.run();

    pipeline.set_state(gst::State::Null)?;
    bus.remove_watch().unwrap();

    Ok(())
}