Compare commits

...

3 commits

3 changed files with 200 additions and 1256 deletions

1256
Cargo.lock generated

File diff suppressed because it is too large Load diff

View file

@ -10,14 +10,12 @@ gst = { package = "gstreamer", version = "0.18.3" }
gstreamer-app = "0.18.0" gstreamer-app = "0.18.0"
gstreamer-base = "0.18.0" gstreamer-base = "0.18.0"
gstreamer-video = { version = "0.18.5", features = ["v1_16"] } 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" ctrlc = "3.2.1"
signal-hook = "0.3.13" 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" log = "0.4.14"
pretty_env_logger = "0.4.0" pretty_env_logger = "0.4.0"
tungstenite = "0.17"
serde = "1"
serde_derive = "1"
serde_json = "1"
url = "2"

View file

@ -1,82 +1,146 @@
use gst::prelude::*; use gst::prelude::*;
use gstreamer_app as gst_app; use gstreamer_app as gst_app;
use log::{debug, error, info, trace, warn}; 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::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<()> { fn main() -> eyre::Result<()> {
pretty_env_logger::init(); pretty_env_logger::init_timed();
gst::init()?; gst::init()?;
gstrusoto::plugin_register_static()?;
gstrsclosedcaption::plugin_register_static()?;
gstrstextwrap::plugin_register_static()?;
let pipeline = gst::parse_launch( let pipeline = gst::parse_launch(
r#" r#"
souphttpsrc location="https://playertest.longtailvideo.com/adaptive/elephants_dream_v4/redundant.m3u8" ! hlsdemux name=demuxer 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
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
"#, "#,
)? )?
.downcast::<gst::Pipeline>() .downcast::<gst::Pipeline>()
.unwrap(); .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..."); info!("Starting pipeline...");
let demuxer = pipeline.by_name("demuxer").unwrap(); let demuxer = pipeline.by_name("dec").unwrap();
demuxer.connect_pad_added(|_, pad| { demuxer.connect_pad_added(|_, pad| {
let name = pad.name(); let name = pad.name();
let caps = pad.caps().unwrap(); let caps = pad.caps().unwrap();
let caps_type = caps.structure(0).unwrap().name(); let caps_type = caps.structure(0).unwrap().name();
// dbg!(name);
debug!("Pad {} added with caps {}", name, caps_type); debug!("Pad {} added with caps {}", name, caps_type);
}); });
// let app_sink = pipeline
// .by_name("sink") let (mut socket, response) = connect(Url::parse("ws://localhost:2700").unwrap())?;
// .unwrap()
// .downcast::<gst_app::AppSink>() let config = Configuration::new(16_000);
// .unwrap(); info!(
// app_sink.set_sync(false); "config payload: {}",
// app_sink.set_callbacks( serde_json::to_string_pretty(&config).unwrap()
// gst_app::AppSinkCallbacks::builder() );
// .new_sample(move |app| { let packet = serde_json::to_string(&config).unwrap();
// let sample = app.pull_sample().unwrap(); socket.write_message(Message::Text(packet)).unwrap();
// let buffer = sample.buffer().unwrap();
// let shared_socket = Arc::new(Mutex::new(socket));
// // We don't care about buffers that are not video
// if buffer let app_sink = pipeline
// .flags() .by_name("sink")
// .contains(gst::BufferFlags::DECODE_ONLY | gst::BufferFlags::GAP) .unwrap()
// { .downcast::<gst_app::AppSink>()
// return Ok(gst::FlowSuccess::Ok); .unwrap();
// } app_sink.set_sync(false);
// app_sink.set_callbacks(
// // let data = buffer.map_readable().unwrap(); gst_app::AppSinkCallbacks::builder()
// // let text = std::str::from_utf8(&data).unwrap(); .new_sample({
// // println!("Subtext = {}", text); let shared_socket = shared_socket.clone();
// dbg!(buffer); move |app| {
// let sample = app.pull_sample().unwrap();
// Ok(gst::FlowSuccess::Ok) let buffer = sample.buffer().unwrap();
// }) let data = buffer.map_readable().unwrap();
// .build(),
// ); 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 context = glib::MainContext::default();
let main_loop = glib::MainLoop::new(Some(&context), false); let main_loop = glib::MainLoop::new(Some(&context), false);
pipeline.set_state(gst::State::Playing)?; pipeline.set_state(gst::State::Playing)?;
let bus = pipeline.bus().unwrap(); let bus = pipeline.bus().unwrap();
bus.add_watch({ bus.add_watch({
let main_loop = main_loop.clone(); let main_loop = main_loop.clone();
@ -116,5 +180,27 @@ fn main() -> eyre::Result<()> {
pipeline.set_state(gst::State::Null)?; 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(()) Ok(())
} }