Write plain transcribed text

Cargo.toml

@@ -10,14 +10,12 @@ gst = { package = "gstreamer", version = "0.18.3" }
 gstreamer-app = "0.18.0"
 gstreamer-base = "0.18.0"
 gstreamer-video = { version = "0.18.5", features = ["v1_16"] }
-gst-plugin-rusoto = { git = "https://gitlab.freedesktop.org/rafaelcaricio/gst-plugins-rs.git", branch = "transbin-accept-any-video-caps", version = "0.9.0" }
-gst-plugin-closedcaption = { git = "https://gitlab.freedesktop.org/rafaelcaricio/gst-plugins-rs.git", branch = "transbin-accept-any-video-caps", version = "0.9.0" }
-gst-plugin-textwrap = { git = "https://gitlab.freedesktop.org/rafaelcaricio/gst-plugins-rs.git", branch = "transbin-accept-any-video-caps", version = "0.9.0" }
 ctrlc = "3.2.1"
 signal-hook = "0.3.13"
-#tokio = { version = "1.17", features = ["full"] }
-#axum = "0.4.5"
-#tower = "0.4.12"
-#tower-http = { version = "0.2.2", features = ["add-extension"] }
 log = "0.4.14"
 pretty_env_logger = "0.4.0"
+tungstenite = "0.17"
+serde = "1"
+serde_derive = "1"
+serde_json = "1"
+url = "2"

src/main.rs

@@ -1,82 +1,146 @@
 use gst::prelude::*;
 use gstreamer_app as gst_app;
 use log::{debug, error, info, trace, warn};
+use serde_derive::{Deserialize, Serialize};
+use std::fs::File;
+use std::io::{Cursor, Write};
 use std::process;
+use std::sync::{Arc, Mutex};
+use tungstenite::{connect, Message};
+use url::Url;
+
+#[derive(Deserialize, Serialize, Debug)]
+pub struct Configuration {
+    config: ConfigInner,
+}
+
+#[derive(Deserialize, Serialize, Debug)]
+struct ConfigInner {
+    /// Sample rate the audio will be provided at.
+    sample_rate: i32,
+
+    /// Show time ranges of each word in the transcription.
+    words: bool,
+}
+
+impl Configuration {
+    pub fn new(sample_rate: i32) -> Self {
+        Self {
+            config: ConfigInner {
+                sample_rate,
+                // We always want to receive the words with their time ranges.
+                words: true,
+            },
+        }
+    }
+}
+
+#[derive(Deserialize, Serialize, Debug)]
+pub struct Transcript {
+    pub result: Vec<WordInfo>,
+    pub text: String,
+}
+
+#[derive(Deserialize, Serialize, Debug)]
+pub struct WordInfo {
+    #[serde(rename = "conf")]
+    pub confidence: f64,
+    pub word: String,
+    pub start: f64,
+    pub end: f64,
+}
 
 fn main() -> eyre::Result<()> {
-    pretty_env_logger::init();
+    pretty_env_logger::init_timed();
     gst::init()?;
-    gstrusoto::plugin_register_static()?;
-    gstrsclosedcaption::plugin_register_static()?;
-    gstrstextwrap::plugin_register_static()?;
 
     let pipeline = gst::parse_launch(
         r#"
-        souphttpsrc location="https://playertest.longtailvideo.com/adaptive/elephants_dream_v4/redundant.m3u8" ! hlsdemux name=demuxer
-
-        demuxer.src_0 ! decodebin ! cccombiner name=ccc_fr ! videoconvert ! x264enc ! video/x-h264,profile=main ! muxer.video_0
-        demuxer.src_1 ! decodebin ! audioconvert ! audioresample ! opusenc ! audio/x-opus,rate=48000,channels=2 ! muxer.audio_0
-        demuxer.src_2 ! decodebin ! audioconvert ! audioresample ! opusenc ! audio/x-opus,rate=48000,channels=2 ! muxer.audio_1
-        demuxer.src_3 ! decodebin ! audioconvert ! audioresample ! opusenc ! audio/x-opus,rate=48000,channels=2 ! muxer.audio_2
-
-        souphttpsrc location="https://playertest.longtailvideo.com/adaptive/elephants_dream_v4/media_b/french/ed.m3u8" ! hlsdemux ! subparse ! tttocea608 ! ccconverter ! closedcaption/x-cea-708,format=cc_data ! ccc_fr.caption
-
-        qtmux name=muxer ! filesink location=output_cae708_only_fr.mp4
+        uridecodebin uri=file:///Users/rafael.caricio/video.mkv name=dec dec.src_1 ! audio/x-raw !
+        audioconvert ! audiorate ! audioresample ! audio/x-raw,format=S16LE,rate=16000,channels=1 ! appsink name=sink
     "#,
     )?
         .downcast::<gst::Pipeline>()
         .unwrap();
-    pipeline.set_async_handling(true);
-
-    // souphttpsrc location="https://playertest.longtailvideo.com/adaptive/elephants_dream_v4/media_b/chinese/ed.m3u8" ! hlsdemux ! subparse ! tttocea608 ! ccconverter ! closedcaption/x-cea-708,format=cc_data ! ccc_ch.caption
-    // souphttpsrc location="https://playertest.longtailvideo.com/adaptive/elephants_dream_v4/media_b/french/ed.m3u8" ! hlsdemux ! subparse ! tttocea608 ! appsink name=sink
 
     info!("Starting pipeline...");
 
-    let demuxer = pipeline.by_name("demuxer").unwrap();
+    let demuxer = pipeline.by_name("dec").unwrap();
     demuxer.connect_pad_added(|_, pad| {
         let name = pad.name();
         let caps = pad.caps().unwrap();
         let caps_type = caps.structure(0).unwrap().name();
-        // dbg!(name);
         debug!("Pad {} added with caps {}", name, caps_type);
     });
-    // let app_sink = pipeline
-    //     .by_name("sink")
-    //     .unwrap()
-    //     .downcast::<gst_app::AppSink>()
-    //     .unwrap();
-    // app_sink.set_sync(false);
-    // app_sink.set_callbacks(
-    //     gst_app::AppSinkCallbacks::builder()
-    //         .new_sample(move |app| {
-    //             let sample = app.pull_sample().unwrap();
-    //             let buffer = sample.buffer().unwrap();
-    //
-    //             // We don't care about buffers that are not video
-    //             if buffer
-    //                 .flags()
-    //                 .contains(gst::BufferFlags::DECODE_ONLY | gst::BufferFlags::GAP)
-    //             {
-    //                 return Ok(gst::FlowSuccess::Ok);
-    //             }
-    //
-    //             // let data = buffer.map_readable().unwrap();
-    //             // let text = std::str::from_utf8(&data).unwrap();
-    //             // println!("Subtext = {}", text);
-    //             dbg!(buffer);
-    //
-    //             Ok(gst::FlowSuccess::Ok)
-    //         })
-    //         .build(),
-    // );
+
+    let (mut socket, response) = connect(Url::parse("ws://localhost:2700").unwrap())?;
+
+    let config = Configuration::new(16_000);
+    info!(
+        "config payload: {}",
+        serde_json::to_string_pretty(&config).unwrap()
+    );
+    let packet = serde_json::to_string(&config).unwrap();
+    socket.write_message(Message::Text(packet)).unwrap();
+
+    let shared_socket = Arc::new(Mutex::new(socket));
+
+    let app_sink = pipeline
+        .by_name("sink")
+        .unwrap()
+        .downcast::<gst_app::AppSink>()
+        .unwrap();
+    app_sink.set_sync(false);
+    app_sink.set_callbacks(
+        gst_app::AppSinkCallbacks::builder()
+            .new_sample({
+                let shared_socket = shared_socket.clone();
+                move |app| {
+                    let sample = app.pull_sample().unwrap();
+                    let buffer = sample.buffer().unwrap();
+                    let data = buffer.map_readable().unwrap();
+
+                    let mut socket = shared_socket.lock().unwrap();
+
+                    for chunk in data.chunks(8_000) {
+                        socket
+                            .write_message(Message::Binary(chunk.to_vec()))
+                            .unwrap();
+
+                        let msg = socket.read_message().unwrap();
+                        match msg {
+                            Message::Text(payload) => {
+                                match serde_json::from_str::<Transcript>(&payload) {
+                                    Ok(transcript) => {
+                                        let text = transcript
+                                            .result
+                                            .iter()
+                                            .map(|p| p.word.to_string())
+                                            .collect::<Vec<_>>()
+                                            .join(" ");
+                                        info!("result: {}", text);
+                                    }
+                                    Err(_) => {
+                                        // The payload is still not a final transcript, so we just ignore it
+                                        // info!("No results...");
+                                    }
+                                }
+                            }
+                            _ => {}
+                        };
+                    }
+
+                    Ok(gst::FlowSuccess::Ok)
+                }
+            })
+            .build(),
+    );
 
     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();
@@ -101,7 +165,7 @@ fn main() -> eyre::Result<()> {
             glib::Continue(true)
         }
     })
-        .expect("Failed to add bus watch");
+    .expect("Failed to add bus watch");
 
     ctrlc::set_handler({
         let pipeline_weak = pipeline.downgrade();
@@ -116,5 +180,27 @@ fn main() -> eyre::Result<()> {
 
     pipeline.set_state(gst::State::Null)?;
 
+    // let mut out_wav_buffer = Cursor::new(Vec::new());
+    // let mut writer = WavWriter::new(
+    //     &mut out_wav_buffer,
+    //     WavSpec {
+    //         channels: 1,
+    //         sample_rate: 48000,
+    //         bits_per_sample: 16,
+    //         sample_format: hound::SampleFormat::Int,
+    //     },
+    // )
+    // .unwrap();
+    //
+    // let mut raw_audio_content = Cursor::new(raw_audio_content.lock().unwrap().to_vec());
+    //
+    // while let Ok(sample) = raw_audio_content.read_i16::<LittleEndian>() {
+    //     writer.write_sample(sample).unwrap();
+    // }
+    //
+    // drop(writer);
+    // let mut file = File::create("out.raw")?;
+    // file.write_all(&raw_audio_content.into_inner())?;
+
     Ok(())
 }