2020-04-10 17:52:28 +00:00
|
|
|
// Copyright (C) 2020 Mathieu Duponchelle <mathieu@centricular.com>
|
|
|
|
//
|
2022-01-15 18:40:12 +00:00
|
|
|
// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
|
|
|
|
// If a copy of the MPL was not distributed with this file, You can obtain one at
|
|
|
|
// <https://mozilla.org/MPL/2.0/>.
|
2020-04-10 17:52:28 +00:00
|
|
|
//
|
2022-01-15 18:40:12 +00:00
|
|
|
// SPDX-License-Identifier: MPL-2.0
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-06-03 18:20:54 +00:00
|
|
|
use gst::glib;
|
2020-04-09 22:51:03 +00:00
|
|
|
use gst::prelude::*;
|
|
|
|
use gst::subclass::prelude::*;
|
2022-10-09 13:06:59 +00:00
|
|
|
use gst::{element_imp_error, error_msg, loggable_error};
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
use std::default::Default;
|
|
|
|
|
2022-08-04 06:11:41 +00:00
|
|
|
use aws_config::default_provider::credentials::DefaultCredentialsChain;
|
2022-05-17 15:25:08 +00:00
|
|
|
use aws_sig_auth::signer::{self, HttpSignatureType, OperationSigningConfig, RequestConfig};
|
|
|
|
use aws_smithy_http::body::SdkBody;
|
2022-08-04 06:11:41 +00:00
|
|
|
use aws_types::credentials::ProvideCredentials;
|
2022-05-17 15:25:08 +00:00
|
|
|
use aws_types::region::{Region, SigningRegion};
|
|
|
|
use aws_types::{Credentials, SigningService};
|
|
|
|
use std::time::{Duration, SystemTime};
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2022-05-17 15:25:08 +00:00
|
|
|
use chrono::prelude::*;
|
|
|
|
use http::Uri;
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
use async_tungstenite::tungstenite::error::Error as WsError;
|
|
|
|
use async_tungstenite::{tokio::connect_async, tungstenite::Message};
|
|
|
|
use futures::channel::mpsc;
|
|
|
|
use futures::future::{abortable, AbortHandle};
|
|
|
|
use futures::prelude::*;
|
|
|
|
use tokio::runtime;
|
|
|
|
|
2021-06-22 18:53:49 +00:00
|
|
|
use std::cmp::Ordering;
|
2020-04-09 22:51:03 +00:00
|
|
|
use std::collections::VecDeque;
|
|
|
|
use std::pin::Pin;
|
2020-04-10 17:52:28 +00:00
|
|
|
use std::sync::Mutex;
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2020-12-29 15:28:19 +00:00
|
|
|
use atomic_refcell::AtomicRefCell;
|
|
|
|
|
2020-11-14 17:24:01 +00:00
|
|
|
use super::packet::*;
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
use serde_derive::{Deserialize, Serialize};
|
2020-04-10 17:52:28 +00:00
|
|
|
|
|
|
|
use once_cell::sync::Lazy;
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2022-08-03 22:42:50 +00:00
|
|
|
use super::{AwsTranscriberResultStability, AwsTranscriberVocabularyFilterMethod};
|
2021-06-19 01:27:42 +00:00
|
|
|
|
2022-08-04 06:11:41 +00:00
|
|
|
const DEFAULT_TRANSCRIBER_REGION: &str = "us-east-1";
|
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
#[derive(Deserialize, Serialize, Debug)]
|
2020-04-09 22:51:03 +00:00
|
|
|
#[serde(rename_all = "PascalCase")]
|
|
|
|
struct TranscriptItem {
|
|
|
|
content: String,
|
|
|
|
end_time: f32,
|
|
|
|
start_time: f32,
|
|
|
|
#[serde(rename = "Type")]
|
|
|
|
type_: String,
|
2021-06-19 01:27:42 +00:00
|
|
|
stable: bool,
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
#[derive(Deserialize, Serialize, Debug)]
|
2020-04-09 22:51:03 +00:00
|
|
|
#[serde(rename_all = "PascalCase")]
|
|
|
|
struct TranscriptAlternative {
|
|
|
|
items: Vec<TranscriptItem>,
|
|
|
|
transcript: String,
|
|
|
|
}
|
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
#[derive(Deserialize, Serialize, Debug)]
|
2020-04-09 22:51:03 +00:00
|
|
|
#[serde(rename_all = "PascalCase")]
|
|
|
|
struct TranscriptResult {
|
|
|
|
alternatives: Vec<TranscriptAlternative>,
|
|
|
|
end_time: f32,
|
|
|
|
start_time: f32,
|
|
|
|
is_partial: bool,
|
|
|
|
result_id: String,
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Deserialize, Debug)]
|
|
|
|
#[serde(rename_all = "PascalCase")]
|
|
|
|
struct TranscriptTranscript {
|
|
|
|
results: Vec<TranscriptResult>,
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Deserialize, Debug)]
|
|
|
|
#[serde(rename_all = "PascalCase")]
|
|
|
|
struct Transcript {
|
|
|
|
transcript: TranscriptTranscript,
|
|
|
|
}
|
|
|
|
|
2020-04-10 21:51:19 +00:00
|
|
|
#[derive(Deserialize, Debug)]
|
|
|
|
#[serde(rename_all = "PascalCase")]
|
|
|
|
struct ExceptionMessage {
|
|
|
|
message: String,
|
|
|
|
}
|
|
|
|
|
2020-04-10 17:52:28 +00:00
|
|
|
static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
|
|
|
|
gst::DebugCategory::new(
|
|
|
|
"awstranscribe",
|
|
|
|
gst::DebugColorFlags::empty(),
|
|
|
|
Some("AWS Transcribe element"),
|
|
|
|
)
|
|
|
|
});
|
|
|
|
|
|
|
|
static RUNTIME: Lazy<runtime::Runtime> = Lazy::new(|| {
|
2021-01-09 10:14:31 +00:00
|
|
|
runtime::Builder::new_multi_thread()
|
2020-04-09 22:51:03 +00:00
|
|
|
.enable_all()
|
2021-01-09 10:14:31 +00:00
|
|
|
.worker_threads(1)
|
2020-04-09 22:51:03 +00:00
|
|
|
.build()
|
2020-04-10 17:52:28 +00:00
|
|
|
.unwrap()
|
|
|
|
});
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-05-26 13:22:01 +00:00
|
|
|
const DEFAULT_LATENCY: gst::ClockTime = gst::ClockTime::from_seconds(8);
|
2021-07-27 00:36:59 +00:00
|
|
|
const DEFAULT_LATENESS: gst::ClockTime = gst::ClockTime::from_seconds(0);
|
2021-06-19 01:27:42 +00:00
|
|
|
const DEFAULT_STABILITY: AwsTranscriberResultStability = AwsTranscriberResultStability::Low;
|
2022-08-03 22:42:50 +00:00
|
|
|
const DEFAULT_VOCABULARY_FILTER_METHOD: AwsTranscriberVocabularyFilterMethod =
|
|
|
|
AwsTranscriberVocabularyFilterMethod::Mask;
|
2021-05-26 13:22:01 +00:00
|
|
|
const GRANULARITY: gst::ClockTime = gst::ClockTime::from_mseconds(100);
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
#[derive(Debug, Clone)]
|
|
|
|
struct Settings {
|
2021-05-26 13:22:01 +00:00
|
|
|
latency: gst::ClockTime,
|
2021-07-27 00:36:59 +00:00
|
|
|
lateness: gst::ClockTime,
|
2020-04-09 22:51:03 +00:00
|
|
|
language_code: Option<String>,
|
2021-02-19 20:54:08 +00:00
|
|
|
vocabulary: Option<String>,
|
2021-06-16 22:53:04 +00:00
|
|
|
session_id: Option<String>,
|
2021-06-19 01:27:42 +00:00
|
|
|
results_stability: AwsTranscriberResultStability,
|
2021-09-27 13:49:12 +00:00
|
|
|
access_key: Option<String>,
|
|
|
|
secret_access_key: Option<String>,
|
2022-06-16 07:16:28 +00:00
|
|
|
session_token: Option<String>,
|
2022-08-03 22:42:50 +00:00
|
|
|
vocabulary_filter: Option<String>,
|
|
|
|
vocabulary_filter_method: AwsTranscriberVocabularyFilterMethod,
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl Default for Settings {
|
|
|
|
fn default() -> Self {
|
|
|
|
Self {
|
2021-05-26 13:22:01 +00:00
|
|
|
latency: DEFAULT_LATENCY,
|
2021-07-27 00:36:59 +00:00
|
|
|
lateness: DEFAULT_LATENESS,
|
2020-04-09 22:51:03 +00:00
|
|
|
language_code: Some("en-US".to_string()),
|
2021-02-19 20:54:08 +00:00
|
|
|
vocabulary: None,
|
2021-06-16 22:53:04 +00:00
|
|
|
session_id: None,
|
2021-06-19 01:27:42 +00:00
|
|
|
results_stability: DEFAULT_STABILITY,
|
2021-09-27 13:49:12 +00:00
|
|
|
access_key: None,
|
|
|
|
secret_access_key: None,
|
2022-06-16 07:16:28 +00:00
|
|
|
session_token: None,
|
2022-08-03 22:42:50 +00:00
|
|
|
vocabulary_filter: None,
|
|
|
|
vocabulary_filter_method: DEFAULT_VOCABULARY_FILTER_METHOD,
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
struct State {
|
|
|
|
connected: bool,
|
|
|
|
sender: Option<mpsc::Sender<Message>>,
|
|
|
|
recv_abort_handle: Option<AbortHandle>,
|
|
|
|
send_abort_handle: Option<AbortHandle>,
|
|
|
|
in_segment: gst::FormattedSegment<gst::ClockTime>,
|
|
|
|
out_segment: gst::FormattedSegment<gst::ClockTime>,
|
|
|
|
seqnum: gst::Seqnum,
|
|
|
|
buffers: VecDeque<gst::Buffer>,
|
|
|
|
send_eos: bool,
|
|
|
|
discont: bool,
|
2021-06-19 01:27:42 +00:00
|
|
|
partial_index: usize,
|
2021-06-22 18:53:49 +00:00
|
|
|
send_events: bool,
|
|
|
|
start_time: Option<gst::ClockTime>,
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl Default for State {
|
|
|
|
fn default() -> Self {
|
|
|
|
Self {
|
|
|
|
connected: false,
|
|
|
|
sender: None,
|
|
|
|
recv_abort_handle: None,
|
|
|
|
send_abort_handle: None,
|
|
|
|
in_segment: gst::FormattedSegment::new(),
|
|
|
|
out_segment: gst::FormattedSegment::new(),
|
|
|
|
seqnum: gst::Seqnum::next(),
|
|
|
|
buffers: VecDeque::new(),
|
|
|
|
send_eos: false,
|
|
|
|
discont: true,
|
2021-06-19 01:27:42 +00:00
|
|
|
partial_index: 0,
|
2021-06-22 18:53:49 +00:00
|
|
|
send_events: true,
|
|
|
|
start_time: None,
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-12-29 15:28:19 +00:00
|
|
|
type WsSink = Pin<Box<dyn Sink<Message, Error = WsError> + Send + Sync>>;
|
2020-04-10 17:52:28 +00:00
|
|
|
|
2020-11-14 17:24:01 +00:00
|
|
|
pub struct Transcriber {
|
2020-04-09 22:51:03 +00:00
|
|
|
srcpad: gst::Pad,
|
|
|
|
sinkpad: gst::Pad,
|
|
|
|
settings: Mutex<Settings>,
|
|
|
|
state: Mutex<State>,
|
2020-12-29 15:28:19 +00:00
|
|
|
ws_sink: AtomicRefCell<Option<WsSink>>,
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
fn build_packet(payload: &[u8]) -> Vec<u8> {
|
|
|
|
let headers = [
|
|
|
|
Header {
|
2020-04-10 21:51:19 +00:00
|
|
|
name: ":event-type".into(),
|
|
|
|
value: "AudioEvent".into(),
|
2020-04-09 22:51:03 +00:00
|
|
|
value_type: 7,
|
|
|
|
},
|
|
|
|
Header {
|
2020-04-10 21:51:19 +00:00
|
|
|
name: ":content-type".into(),
|
|
|
|
value: "application/octet-stream".into(),
|
2020-04-09 22:51:03 +00:00
|
|
|
value_type: 7,
|
|
|
|
},
|
|
|
|
Header {
|
2020-04-10 21:51:19 +00:00
|
|
|
name: ":message-type".into(),
|
|
|
|
value: "event".into(),
|
2020-04-09 22:51:03 +00:00
|
|
|
value_type: 7,
|
|
|
|
},
|
|
|
|
];
|
|
|
|
|
|
|
|
encode_packet(payload, &headers).expect("foobar")
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Transcriber {
|
2022-10-09 13:06:59 +00:00
|
|
|
fn dequeue(&self) -> bool {
|
2020-04-09 22:51:03 +00:00
|
|
|
/* First, check our pending buffers */
|
2020-04-10 17:52:28 +00:00
|
|
|
let mut items = vec![];
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
let now = match self.instance().current_running_time() {
|
2021-06-22 18:53:49 +00:00
|
|
|
Some(now) => now,
|
|
|
|
None => {
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
let latency = self.settings.lock().unwrap().latency;
|
|
|
|
|
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
|
|
|
|
if state.start_time.is_none() {
|
|
|
|
state.start_time = Some(now);
|
|
|
|
state.out_segment.set_position(now);
|
|
|
|
}
|
2021-06-19 01:27:42 +00:00
|
|
|
|
2021-06-22 18:53:49 +00:00
|
|
|
let start_time = state.start_time.unwrap();
|
|
|
|
let mut last_position = state.out_segment.position().unwrap();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-06-22 18:53:49 +00:00
|
|
|
let send_eos = state.send_eos && state.buffers.is_empty();
|
|
|
|
|
|
|
|
while let Some(buf) = state.buffers.front() {
|
|
|
|
let pts = buf.pts().unwrap();
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::trace!(
|
2021-06-22 18:53:49 +00:00
|
|
|
CAT,
|
2022-10-09 13:06:59 +00:00
|
|
|
imp: self,
|
2021-06-22 18:53:49 +00:00
|
|
|
"Checking now {} if item is ready for dequeuing, PTS {}, threshold {} vs {}",
|
|
|
|
now,
|
|
|
|
pts,
|
|
|
|
pts + latency.saturating_sub(3 * GRANULARITY),
|
|
|
|
now - start_time
|
|
|
|
);
|
|
|
|
|
|
|
|
if pts + latency.saturating_sub(3 * GRANULARITY) < now - start_time {
|
|
|
|
/* Safe unwrap, we know we have an item */
|
|
|
|
let mut buf = state.buffers.pop_front().unwrap();
|
2020-05-28 21:55:00 +00:00
|
|
|
|
2021-05-26 13:22:01 +00:00
|
|
|
{
|
2021-06-22 18:53:49 +00:00
|
|
|
let buf_mut = buf.get_mut().unwrap();
|
|
|
|
|
|
|
|
buf_mut.set_pts(start_time + pts);
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
2021-06-22 18:53:49 +00:00
|
|
|
|
|
|
|
items.push(buf);
|
|
|
|
} else {
|
|
|
|
break;
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
2021-06-22 18:53:49 +00:00
|
|
|
}
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-06-22 18:53:49 +00:00
|
|
|
let seqnum = state.seqnum;
|
|
|
|
|
|
|
|
drop(state);
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
/* We're EOS, we can pause and exit early */
|
|
|
|
if send_eos {
|
2020-04-10 17:52:28 +00:00
|
|
|
let _ = self.srcpad.pause_task();
|
|
|
|
|
2020-04-09 22:51:03 +00:00
|
|
|
return self
|
|
|
|
.srcpad
|
2020-06-24 22:33:57 +00:00
|
|
|
.push_event(gst::event::Eos::builder().seqnum(seqnum).build());
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
for mut buf in items.drain(..) {
|
2021-06-22 18:53:49 +00:00
|
|
|
let mut pts = buf.pts().unwrap();
|
|
|
|
let mut duration = buf.duration().unwrap();
|
|
|
|
|
|
|
|
match pts.cmp(&last_position) {
|
|
|
|
Ordering::Greater => {
|
|
|
|
let gap_event = gst::event::Gap::builder(last_position)
|
|
|
|
.duration(pts - last_position)
|
|
|
|
.seqnum(seqnum)
|
|
|
|
.build();
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::log!(CAT, "Pushing gap: {} -> {}", last_position, pts);
|
2021-06-22 18:53:49 +00:00
|
|
|
if !self.srcpad.push_event(gap_event) {
|
|
|
|
return false;
|
|
|
|
}
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
2021-06-22 18:53:49 +00:00
|
|
|
Ordering::Less => {
|
|
|
|
let delta = last_position - pts;
|
|
|
|
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::warning!(
|
2021-06-22 18:53:49 +00:00
|
|
|
CAT,
|
2022-10-09 13:06:59 +00:00
|
|
|
imp: self,
|
2021-06-22 18:53:49 +00:00
|
|
|
"Updating item PTS ({} < {}), consider increasing latency",
|
|
|
|
pts,
|
|
|
|
last_position
|
|
|
|
);
|
|
|
|
|
|
|
|
pts = last_position;
|
|
|
|
duration = duration.saturating_sub(delta);
|
|
|
|
|
|
|
|
{
|
|
|
|
let buf_mut = buf.get_mut().unwrap();
|
|
|
|
|
|
|
|
buf_mut.set_pts(pts);
|
|
|
|
buf_mut.set_duration(duration);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
_ => (),
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
2021-06-22 18:53:49 +00:00
|
|
|
|
|
|
|
last_position = pts + duration;
|
|
|
|
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::debug!(CAT, "Pushing buffer: {} -> {}", pts, pts + duration);
|
2021-06-22 18:53:49 +00:00
|
|
|
|
2020-04-10 17:52:28 +00:00
|
|
|
if self.srcpad.push(buf).is_err() {
|
2020-04-09 22:51:03 +00:00
|
|
|
return false;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/* next, push a gap if we're lagging behind the target position */
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::trace!(
|
2021-06-22 18:53:49 +00:00
|
|
|
CAT,
|
2022-10-09 13:06:59 +00:00
|
|
|
imp: self,
|
2021-06-22 18:53:49 +00:00
|
|
|
"Checking now: {} if we need to push a gap, last_position: {}, threshold: {}",
|
|
|
|
now,
|
|
|
|
last_position,
|
|
|
|
last_position + latency.saturating_sub(GRANULARITY)
|
|
|
|
);
|
|
|
|
|
|
|
|
if now > last_position + latency.saturating_sub(GRANULARITY) {
|
|
|
|
let duration = now - last_position - latency.saturating_sub(GRANULARITY);
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-06-22 18:53:49 +00:00
|
|
|
let gap_event = gst::event::Gap::builder(last_position)
|
2021-05-26 13:22:01 +00:00
|
|
|
.duration(duration)
|
2020-04-09 22:51:03 +00:00
|
|
|
.seqnum(seqnum)
|
|
|
|
.build();
|
2021-06-22 18:53:49 +00:00
|
|
|
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::log!(
|
2021-06-22 18:53:49 +00:00
|
|
|
CAT,
|
|
|
|
"Pushing gap: {} -> {}",
|
|
|
|
last_position,
|
|
|
|
last_position + duration
|
|
|
|
);
|
|
|
|
|
|
|
|
last_position += duration;
|
|
|
|
|
2020-04-09 22:51:03 +00:00
|
|
|
if !self.srcpad.push_event(gap_event) {
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
self.state
|
|
|
|
.lock()
|
|
|
|
.unwrap()
|
|
|
|
.out_segment
|
|
|
|
.set_position(last_position);
|
|
|
|
|
|
|
|
true
|
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn enqueue(&self, state: &mut State, alternative: &TranscriptAlternative, partial: bool) {
|
2021-07-27 00:36:59 +00:00
|
|
|
let lateness = self.settings.lock().unwrap().lateness;
|
|
|
|
|
2021-11-24 12:57:15 +00:00
|
|
|
if alternative.items.len() <= state.partial_index {
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::error!(
|
2021-11-24 12:57:15 +00:00
|
|
|
CAT,
|
2022-10-09 13:06:59 +00:00
|
|
|
imp: self,
|
2021-11-24 12:57:15 +00:00
|
|
|
"sanity check failed, alternative length {} < partial_index {}",
|
|
|
|
alternative.items.len(),
|
|
|
|
state.partial_index
|
|
|
|
);
|
|
|
|
|
|
|
|
if !partial {
|
|
|
|
state.partial_index = 0;
|
|
|
|
}
|
|
|
|
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
for item in &alternative.items[state.partial_index..] {
|
2021-06-22 18:53:49 +00:00
|
|
|
let start_time =
|
2021-07-27 00:36:59 +00:00
|
|
|
gst::ClockTime::from_nseconds((item.start_time as f64 * 1_000_000_000.0) as u64)
|
|
|
|
+ lateness;
|
2021-06-22 18:53:49 +00:00
|
|
|
let end_time =
|
2021-07-27 00:36:59 +00:00
|
|
|
gst::ClockTime::from_nseconds((item.end_time as f64 * 1_000_000_000.0) as u64)
|
|
|
|
+ lateness;
|
2021-05-26 13:22:01 +00:00
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
if !item.stable {
|
|
|
|
break;
|
|
|
|
}
|
2020-05-28 21:55:00 +00:00
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
/* Should be sent now */
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::debug!(
|
2021-06-22 18:53:49 +00:00
|
|
|
CAT,
|
2022-10-09 13:06:59 +00:00
|
|
|
imp: self,
|
2021-06-22 18:53:49 +00:00
|
|
|
"Item is ready for queuing: {}, PTS {}",
|
|
|
|
item.content,
|
|
|
|
start_time
|
|
|
|
);
|
2021-06-19 01:27:42 +00:00
|
|
|
let mut buf = gst::Buffer::from_mut_slice(item.content.clone().into_bytes());
|
2020-05-28 21:55:00 +00:00
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
{
|
|
|
|
let buf = buf.get_mut().unwrap();
|
|
|
|
|
|
|
|
if state.discont {
|
|
|
|
buf.set_flags(gst::BufferFlags::DISCONT);
|
|
|
|
state.discont = false;
|
|
|
|
}
|
2020-05-28 21:55:00 +00:00
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
buf.set_pts(start_time);
|
|
|
|
buf.set_duration(end_time - start_time);
|
2020-05-28 21:55:00 +00:00
|
|
|
}
|
2021-06-19 01:27:42 +00:00
|
|
|
|
|
|
|
state.partial_index += 1;
|
|
|
|
|
|
|
|
state.buffers.push_back(buf);
|
|
|
|
}
|
|
|
|
|
|
|
|
if !partial {
|
|
|
|
state.partial_index = 0;
|
2020-05-28 21:55:00 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn loop_fn(&self, receiver: &mut mpsc::Receiver<Message>) -> Result<(), gst::ErrorMessage> {
|
2021-06-22 18:53:49 +00:00
|
|
|
let mut events = {
|
|
|
|
let mut events = vec![];
|
|
|
|
|
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
|
|
|
|
if state.send_events {
|
|
|
|
events.push(
|
|
|
|
gst::event::StreamStart::builder("transcription")
|
|
|
|
.seqnum(state.seqnum)
|
|
|
|
.build(),
|
|
|
|
);
|
|
|
|
|
|
|
|
let caps = gst::Caps::builder("text/x-raw")
|
2021-11-06 07:34:10 +00:00
|
|
|
.field("format", "utf8")
|
2021-06-22 18:53:49 +00:00
|
|
|
.build();
|
|
|
|
events.push(
|
|
|
|
gst::event::Caps::builder(&caps)
|
|
|
|
.seqnum(state.seqnum)
|
|
|
|
.build(),
|
|
|
|
);
|
|
|
|
|
|
|
|
events.push(
|
|
|
|
gst::event::Segment::builder(&state.out_segment)
|
|
|
|
.seqnum(state.seqnum)
|
|
|
|
.build(),
|
|
|
|
);
|
|
|
|
|
|
|
|
state.send_events = false;
|
|
|
|
}
|
|
|
|
|
|
|
|
events
|
|
|
|
};
|
|
|
|
|
|
|
|
for event in events.drain(..) {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::info!(CAT, imp: self, "Sending {:?}", event);
|
2021-06-22 18:53:49 +00:00
|
|
|
self.srcpad.push_event(event);
|
|
|
|
}
|
|
|
|
|
2020-04-09 22:51:03 +00:00
|
|
|
let future = async move {
|
|
|
|
let msg = match receiver.next().await {
|
|
|
|
Some(msg) => msg,
|
|
|
|
/* Sender was closed */
|
|
|
|
None => {
|
2020-04-10 17:52:28 +00:00
|
|
|
let _ = self.srcpad.pause_task();
|
2020-04-09 22:51:03 +00:00
|
|
|
return Ok(());
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
match msg {
|
|
|
|
Message::Binary(buf) => {
|
2020-04-10 21:51:19 +00:00
|
|
|
let (_, pkt) = parse_packet(&buf).map_err(|err| {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::error!(CAT, imp: self, "Failed to parse packet: {}", err);
|
2020-12-20 18:43:45 +00:00
|
|
|
error_msg!(
|
2020-04-10 21:51:19 +00:00
|
|
|
gst::StreamError::Failed,
|
|
|
|
["Failed to parse packet: {}", err]
|
|
|
|
)
|
|
|
|
})?;
|
|
|
|
|
2020-04-09 22:51:03 +00:00
|
|
|
let payload = std::str::from_utf8(pkt.payload).unwrap();
|
2020-04-10 21:51:19 +00:00
|
|
|
|
|
|
|
if packet_is_exception(&pkt) {
|
|
|
|
let message: ExceptionMessage =
|
2021-07-30 10:53:35 +00:00
|
|
|
serde_json::from_str(payload).map_err(|err| {
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::error!(
|
2020-04-10 21:51:19 +00:00
|
|
|
CAT,
|
2022-10-09 13:06:59 +00:00
|
|
|
imp: self,
|
2020-04-10 21:51:19 +00:00
|
|
|
"Unexpected exception message: {} ({})",
|
|
|
|
payload,
|
|
|
|
err
|
|
|
|
);
|
2020-12-20 18:43:45 +00:00
|
|
|
error_msg!(
|
2020-04-10 21:51:19 +00:00
|
|
|
gst::StreamError::Failed,
|
|
|
|
["Unexpected exception message: {} ({})", payload, err]
|
|
|
|
)
|
|
|
|
})?;
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::error!(CAT, imp: self, "AWS raised an error: {}", message.message);
|
2020-04-14 08:59:08 +00:00
|
|
|
|
2020-12-20 18:43:45 +00:00
|
|
|
return Err(error_msg!(
|
2020-04-10 21:51:19 +00:00
|
|
|
gst::StreamError::Failed,
|
|
|
|
["AWS raised an error: {}", message.message]
|
2020-04-14 08:59:08 +00:00
|
|
|
));
|
2020-04-10 21:51:19 +00:00
|
|
|
}
|
|
|
|
|
2021-07-30 10:53:35 +00:00
|
|
|
let transcript: Transcript = serde_json::from_str(payload).map_err(|err| {
|
2021-06-19 01:27:42 +00:00
|
|
|
error_msg!(
|
|
|
|
gst::StreamError::Failed,
|
|
|
|
["Unexpected binary message: {} ({})", payload, err]
|
|
|
|
)
|
|
|
|
})?;
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
if let Some(result) = transcript.transcript.results.get(0) {
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::trace!(
|
2021-06-19 01:27:42 +00:00
|
|
|
CAT,
|
2022-10-09 13:06:59 +00:00
|
|
|
imp: self,
|
2021-06-19 01:27:42 +00:00
|
|
|
"result: {}",
|
|
|
|
serde_json::to_string_pretty(&result).unwrap(),
|
|
|
|
);
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-06-19 01:27:42 +00:00
|
|
|
if let Some(alternative) = result.alternatives.get(0) {
|
2020-05-28 21:55:00 +00:00
|
|
|
let mut state = self.state.lock().unwrap();
|
2021-06-19 01:27:42 +00:00
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
self.enqueue(&mut state, alternative, result.is_partial)
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
}
|
2021-06-19 01:27:42 +00:00
|
|
|
|
2020-04-09 22:51:03 +00:00
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
_ => Ok(()),
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
/* Wrap in a timeout so we can push gaps regularly */
|
|
|
|
let future = async move {
|
2021-05-26 13:22:01 +00:00
|
|
|
match tokio::time::timeout(GRANULARITY.into(), future).await {
|
2020-04-09 22:51:03 +00:00
|
|
|
Err(_) => {
|
2022-10-09 13:06:59 +00:00
|
|
|
if !self.dequeue() {
|
|
|
|
gst::info!(CAT, imp: self, "Failed to push gap event, pausing");
|
2020-04-10 17:52:28 +00:00
|
|
|
|
|
|
|
let _ = self.srcpad.pause_task();
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
Ok(res) => {
|
2022-10-09 13:06:59 +00:00
|
|
|
if !self.dequeue() {
|
|
|
|
gst::info!(CAT, imp: self, "Failed to push gap event, pausing");
|
2020-04-10 17:52:28 +00:00
|
|
|
|
|
|
|
let _ = self.srcpad.pause_task();
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
res
|
|
|
|
}
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
2021-01-09 10:14:31 +00:00
|
|
|
let _enter = RUNTIME.enter();
|
|
|
|
futures::executor::block_on(future)
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn start_task(&self) -> Result<(), gst::LoggableError> {
|
2020-04-09 22:51:03 +00:00
|
|
|
let (sender, mut receiver) = mpsc::channel(1);
|
|
|
|
|
|
|
|
{
|
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
state.sender = Some(sender);
|
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
let imp = self.ref_counted();
|
2020-04-09 22:51:03 +00:00
|
|
|
let res = self.srcpad.start_task(move || {
|
2022-10-09 13:06:59 +00:00
|
|
|
if let Err(err) = imp.loop_fn(&mut receiver) {
|
|
|
|
element_imp_error!(imp, gst::StreamError::Failed, ["Streaming failed: {}", err]);
|
|
|
|
let _ = imp.srcpad.pause_task();
|
2021-09-08 12:33:31 +00:00
|
|
|
}
|
2020-04-09 22:51:03 +00:00
|
|
|
});
|
|
|
|
if res.is_err() {
|
2020-12-20 18:43:45 +00:00
|
|
|
return Err(loggable_error!(CAT, "Failed to start pad task"));
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
fn src_activatemode(
|
|
|
|
&self,
|
|
|
|
_pad: &gst::Pad,
|
2020-04-10 17:52:28 +00:00
|
|
|
_mode: gst::PadMode,
|
2020-04-09 22:51:03 +00:00
|
|
|
active: bool,
|
|
|
|
) -> Result<(), gst::LoggableError> {
|
|
|
|
if active {
|
2022-10-09 13:06:59 +00:00
|
|
|
self.start_task()?;
|
2020-04-09 22:51:03 +00:00
|
|
|
} else {
|
|
|
|
{
|
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
state.sender = None;
|
|
|
|
}
|
|
|
|
|
|
|
|
let _ = self.srcpad.stop_task();
|
|
|
|
}
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn src_query(&self, pad: &gst::Pad, query: &mut gst::QueryRef) -> bool {
|
2022-01-19 13:07:45 +00:00
|
|
|
use gst::QueryViewMut;
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::log!(CAT, obj: pad, "Handling query {:?}", query);
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
match query.view_mut() {
|
2022-01-19 13:07:45 +00:00
|
|
|
QueryViewMut::Latency(q) => {
|
2020-06-24 22:33:57 +00:00
|
|
|
let mut peer_query = gst::query::Latency::new();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
let ret = self.sinkpad.peer_query(&mut peer_query);
|
|
|
|
|
|
|
|
if ret {
|
2021-04-12 12:49:54 +00:00
|
|
|
let (_, min, _) = peer_query.result();
|
2021-05-26 13:22:01 +00:00
|
|
|
let our_latency = self.settings.lock().unwrap().latency;
|
|
|
|
q.set(true, our_latency + min, gst::ClockTime::NONE);
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
ret
|
|
|
|
}
|
2022-01-19 13:07:45 +00:00
|
|
|
QueryViewMut::Position(q) => {
|
2021-04-12 12:49:54 +00:00
|
|
|
if q.format() == gst::Format::Time {
|
2020-04-09 22:51:03 +00:00
|
|
|
let state = self.state.lock().unwrap();
|
|
|
|
q.set(
|
|
|
|
state
|
2020-04-10 17:52:28 +00:00
|
|
|
.out_segment
|
2021-04-12 12:49:54 +00:00
|
|
|
.to_stream_time(state.out_segment.position()),
|
2020-04-09 22:51:03 +00:00
|
|
|
);
|
|
|
|
true
|
|
|
|
} else {
|
|
|
|
false
|
|
|
|
}
|
|
|
|
}
|
2022-10-12 14:48:47 +00:00
|
|
|
_ => gst::Pad::query_default(pad, Some(&*self.instance()), query),
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn sink_event(&self, pad: &gst::Pad, event: gst::Event) -> bool {
|
2020-04-09 22:51:03 +00:00
|
|
|
use gst::EventView;
|
|
|
|
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::log!(CAT, obj: pad, "Handling event {:?}", event);
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
match event.view() {
|
2022-10-09 13:06:59 +00:00
|
|
|
EventView::Eos(_) => match self.handle_buffer(pad, None) {
|
2020-04-09 22:51:03 +00:00
|
|
|
Err(err) => {
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::error!(CAT, "Failed to send EOS to AWS: {}", err);
|
2020-04-09 22:51:03 +00:00
|
|
|
false
|
|
|
|
}
|
|
|
|
Ok(_) => true,
|
|
|
|
},
|
2020-04-10 17:52:28 +00:00
|
|
|
EventView::FlushStart(_) => {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::info!(CAT, imp: self, "Received flush start, disconnecting");
|
2022-10-12 14:48:47 +00:00
|
|
|
let mut ret = gst::Pad::event_default(pad, Some(&*self.instance()), event);
|
2021-02-09 16:57:34 +00:00
|
|
|
|
|
|
|
match self.srcpad.stop_task() {
|
2020-04-10 17:52:28 +00:00
|
|
|
Err(err) => {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::error!(CAT, imp: self, "Failed to stop srcpad task: {}", err);
|
2021-06-22 18:53:49 +00:00
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
self.disconnect();
|
2021-06-22 18:53:49 +00:00
|
|
|
|
2021-02-09 16:57:34 +00:00
|
|
|
ret = false;
|
2020-04-10 17:52:28 +00:00
|
|
|
}
|
2021-06-22 18:53:49 +00:00
|
|
|
Ok(_) => {
|
2022-10-09 13:06:59 +00:00
|
|
|
self.disconnect();
|
2021-06-22 18:53:49 +00:00
|
|
|
}
|
2021-02-09 16:57:34 +00:00
|
|
|
};
|
2020-04-10 17:52:28 +00:00
|
|
|
|
2021-02-09 16:57:34 +00:00
|
|
|
ret
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
2020-04-10 17:52:28 +00:00
|
|
|
EventView::FlushStop(_) => {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::info!(CAT, imp: self, "Received flush stop, restarting task");
|
2020-04-10 17:52:28 +00:00
|
|
|
|
2022-10-12 14:48:47 +00:00
|
|
|
if gst::Pad::event_default(pad, Some(&*self.instance()), event) {
|
2022-10-09 13:06:59 +00:00
|
|
|
match self.start_task() {
|
2020-04-10 17:52:28 +00:00
|
|
|
Err(err) => {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::error!(CAT, imp: self, "Failed to start srcpad task: {}", err);
|
2020-04-10 17:52:28 +00:00
|
|
|
false
|
|
|
|
}
|
|
|
|
Ok(_) => true,
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
false
|
|
|
|
}
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
EventView::Segment(e) => {
|
2021-04-12 12:49:54 +00:00
|
|
|
let segment = match e.segment().clone().downcast::<gst::ClockTime>() {
|
2020-04-09 22:51:03 +00:00
|
|
|
Err(segment) => {
|
2022-10-09 13:06:59 +00:00
|
|
|
element_imp_error!(
|
|
|
|
self,
|
2020-04-09 22:51:03 +00:00
|
|
|
gst::StreamError::Format,
|
2021-04-12 12:49:54 +00:00
|
|
|
["Only Time segments supported, got {:?}", segment.format(),]
|
2020-04-09 22:51:03 +00:00
|
|
|
);
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
Ok(segment) => segment,
|
|
|
|
};
|
|
|
|
|
2021-06-22 18:53:49 +00:00
|
|
|
let mut state = self.state.lock().unwrap();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-06-22 18:53:49 +00:00
|
|
|
state.in_segment = segment;
|
|
|
|
state.seqnum = e.seqnum();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-06-22 18:53:49 +00:00
|
|
|
true
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
EventView::Tag(_) => true,
|
|
|
|
EventView::Caps(e) => {
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::info!(CAT, "Received caps {:?}", e);
|
2021-06-22 18:53:49 +00:00
|
|
|
true
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
2021-06-22 18:53:49 +00:00
|
|
|
EventView::StreamStart(_) => true,
|
2022-10-12 14:48:47 +00:00
|
|
|
_ => gst::Pad::event_default(pad, Some(&*self.instance()), event),
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
async fn sync_and_send(
|
|
|
|
&self,
|
|
|
|
buffer: Option<gst::Buffer>,
|
|
|
|
) -> Result<gst::FlowSuccess, gst::FlowError> {
|
|
|
|
let mut delay = None;
|
|
|
|
|
|
|
|
{
|
|
|
|
let state = self.state.lock().unwrap();
|
|
|
|
|
|
|
|
if let Some(buffer) = &buffer {
|
2021-04-12 12:49:54 +00:00
|
|
|
let running_time = state.in_segment.to_running_time(buffer.pts());
|
2022-10-09 13:06:59 +00:00
|
|
|
let now = self.instance().current_running_time();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-10-09 10:17:05 +00:00
|
|
|
delay = running_time.opt_checked_sub(now).ok().flatten();
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if let Some(delay) = delay {
|
2021-05-26 13:22:01 +00:00
|
|
|
tokio::time::sleep(delay.into()).await;
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
2020-12-29 15:28:19 +00:00
|
|
|
if let Some(ws_sink) = self.ws_sink.borrow_mut().as_mut() {
|
2020-04-09 22:51:03 +00:00
|
|
|
if let Some(buffer) = buffer {
|
|
|
|
let data = buffer.map_readable().unwrap();
|
|
|
|
for chunk in data.chunks(8192) {
|
|
|
|
let packet = build_packet(chunk);
|
|
|
|
ws_sink.send(Message::Binary(packet)).await.map_err(|err| {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::error!(CAT, imp: self, "Failed sending packet: {}", err);
|
2020-04-09 22:51:03 +00:00
|
|
|
gst::FlowError::Error
|
|
|
|
})?;
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
// EOS
|
|
|
|
let packet = build_packet(&[]);
|
|
|
|
ws_sink.send(Message::Binary(packet)).await.map_err(|err| {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::error!(CAT, imp: self, "Failed sending packet: {}", err);
|
2020-04-09 22:51:03 +00:00
|
|
|
gst::FlowError::Error
|
|
|
|
})?;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
Ok(gst::FlowSuccess::Ok)
|
|
|
|
}
|
|
|
|
|
|
|
|
fn handle_buffer(
|
|
|
|
&self,
|
2020-04-10 17:52:28 +00:00
|
|
|
_pad: &gst::Pad,
|
2020-04-09 22:51:03 +00:00
|
|
|
buffer: Option<gst::Buffer>,
|
|
|
|
) -> Result<gst::FlowSuccess, gst::FlowError> {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::log!(CAT, imp: self, "Handling {:?}", buffer);
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
self.ensure_connection().map_err(|err| {
|
|
|
|
element_imp_error!(
|
|
|
|
self,
|
2020-04-09 22:51:03 +00:00
|
|
|
gst::StreamError::Failed,
|
|
|
|
["Streaming failed: {}", err]
|
|
|
|
);
|
|
|
|
gst::FlowError::Error
|
|
|
|
})?;
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
let (future, abort_handle) = abortable(self.sync_and_send(buffer));
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
self.state.lock().unwrap().send_abort_handle = Some(abort_handle);
|
|
|
|
|
2021-01-09 10:14:31 +00:00
|
|
|
let res = {
|
|
|
|
let _enter = RUNTIME.enter();
|
|
|
|
futures::executor::block_on(future)
|
|
|
|
};
|
|
|
|
|
|
|
|
match res {
|
2020-04-10 17:52:28 +00:00
|
|
|
Err(_) => Err(gst::FlowError::Flushing),
|
2020-04-09 22:51:03 +00:00
|
|
|
Ok(res) => res,
|
2020-04-10 17:52:28 +00:00
|
|
|
}
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
fn sink_chain(
|
|
|
|
&self,
|
|
|
|
pad: &gst::Pad,
|
|
|
|
buffer: gst::Buffer,
|
|
|
|
) -> Result<gst::FlowSuccess, gst::FlowError> {
|
2022-10-09 13:06:59 +00:00
|
|
|
self.handle_buffer(pad, Some(buffer))
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn ensure_connection(&self) -> Result<(), gst::ErrorMessage> {
|
2021-06-22 18:53:49 +00:00
|
|
|
let state = self.state.lock().unwrap();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2020-04-10 17:52:28 +00:00
|
|
|
if state.connected {
|
2020-04-09 22:51:03 +00:00
|
|
|
return Ok(());
|
|
|
|
}
|
|
|
|
|
2021-04-12 12:49:54 +00:00
|
|
|
let in_caps = self.sinkpad.current_caps().unwrap();
|
2021-04-20 12:58:11 +00:00
|
|
|
let s = in_caps.structure(0).unwrap();
|
2021-04-25 12:41:22 +00:00
|
|
|
let sample_rate = s.get::<i32>("rate").unwrap();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
|
2021-07-27 00:36:59 +00:00
|
|
|
if settings.latency + settings.lateness <= 2 * GRANULARITY {
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::error!(
|
2021-07-27 00:36:59 +00:00
|
|
|
CAT,
|
2022-10-09 13:06:59 +00:00
|
|
|
imp: self,
|
2021-07-27 00:36:59 +00:00
|
|
|
"latency + lateness must be greater than 200 milliseconds"
|
|
|
|
);
|
|
|
|
return Err(error_msg!(
|
|
|
|
gst::LibraryError::Settings,
|
|
|
|
["latency + lateness must be greater than 200 milliseconds"]
|
|
|
|
));
|
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::info!(CAT, imp: self, "Connecting ..");
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2022-08-04 06:11:41 +00:00
|
|
|
let region = Region::new(DEFAULT_TRANSCRIBER_REGION);
|
|
|
|
let access_key = settings.access_key.as_ref();
|
|
|
|
let secret_access_key = settings.secret_access_key.as_ref();
|
|
|
|
let session_token = settings.session_token.clone();
|
|
|
|
|
|
|
|
let credentials = match (access_key, secret_access_key) {
|
|
|
|
(Some(key), Some(secret_key)) => {
|
|
|
|
gst::debug!(
|
|
|
|
CAT,
|
2022-10-09 13:06:59 +00:00
|
|
|
imp: self,
|
2022-08-04 06:11:41 +00:00
|
|
|
"Using provided access and secret access key"
|
|
|
|
);
|
|
|
|
Ok(Credentials::new(
|
|
|
|
key.clone(),
|
|
|
|
secret_key.clone(),
|
|
|
|
session_token,
|
|
|
|
None,
|
|
|
|
"transcribe",
|
|
|
|
))
|
|
|
|
}
|
|
|
|
_ => {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::debug!(CAT, imp: self, "Using default AWS credentials");
|
2022-08-04 06:11:41 +00:00
|
|
|
let cred_future = async {
|
|
|
|
let cred = DefaultCredentialsChain::builder()
|
|
|
|
.region(region.clone())
|
|
|
|
.build()
|
|
|
|
.await;
|
|
|
|
cred.provide_credentials().await
|
|
|
|
};
|
|
|
|
|
|
|
|
RUNTIME.block_on(cred_future)
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
if let Err(e) = credentials {
|
2022-05-17 15:25:08 +00:00
|
|
|
return Err(error_msg!(
|
|
|
|
gst::LibraryError::Settings,
|
2022-08-04 06:11:41 +00:00
|
|
|
["Failed to retrieve credentials with error {}", e]
|
2022-05-17 15:25:08 +00:00
|
|
|
));
|
|
|
|
}
|
|
|
|
|
|
|
|
let current_time = Utc::now();
|
|
|
|
|
|
|
|
let mut query_params = String::from("/stream-transcription-websocket?");
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
let language_code = settings
|
|
|
|
.language_code
|
|
|
|
.as_ref()
|
|
|
|
.expect("Language code is required");
|
|
|
|
|
2022-05-17 15:25:08 +00:00
|
|
|
query_params.push_str(
|
|
|
|
format!(
|
|
|
|
"language-code={}&media-encoding=pcm&sample-rate={}",
|
|
|
|
language_code,
|
|
|
|
&sample_rate.to_string(),
|
|
|
|
)
|
|
|
|
.as_str(),
|
2020-04-09 22:51:03 +00:00
|
|
|
);
|
2021-02-19 20:54:08 +00:00
|
|
|
|
|
|
|
if let Some(ref vocabulary) = settings.vocabulary {
|
2022-05-17 15:25:08 +00:00
|
|
|
query_params.push_str(format!("&vocabulary-name={}", vocabulary).as_str());
|
2021-02-19 20:54:08 +00:00
|
|
|
}
|
|
|
|
|
2022-08-03 22:42:50 +00:00
|
|
|
if let Some(ref vocabulary_filter) = settings.vocabulary_filter {
|
|
|
|
query_params
|
|
|
|
.push_str(format!("&vocabulary-filter-name={}", vocabulary_filter).as_str());
|
|
|
|
|
2022-09-08 20:08:20 +00:00
|
|
|
query_params.push_str(
|
|
|
|
format!(
|
|
|
|
"&vocabulary-filter-method={}",
|
|
|
|
match settings.vocabulary_filter_method {
|
|
|
|
AwsTranscriberVocabularyFilterMethod::Mask => "mask",
|
|
|
|
AwsTranscriberVocabularyFilterMethod::Remove => "remove",
|
|
|
|
AwsTranscriberVocabularyFilterMethod::Tag => "tag",
|
|
|
|
}
|
|
|
|
)
|
|
|
|
.as_str(),
|
|
|
|
);
|
|
|
|
}
|
2022-08-03 22:42:50 +00:00
|
|
|
|
2021-06-16 22:53:04 +00:00
|
|
|
if let Some(ref session_id) = settings.session_id {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::debug!(CAT, imp: self, "Using session ID: {}", session_id);
|
2022-05-17 15:25:08 +00:00
|
|
|
query_params.push_str(format!("&session-id={}", session_id).as_str());
|
2021-06-16 22:53:04 +00:00
|
|
|
}
|
|
|
|
|
2022-05-17 15:25:08 +00:00
|
|
|
query_params.push_str("&enable-partial-results-stabilization=true");
|
|
|
|
|
|
|
|
query_params.push_str(
|
|
|
|
format!(
|
|
|
|
"&partial-results-stability={}",
|
|
|
|
match settings.results_stability {
|
|
|
|
AwsTranscriberResultStability::High => "high",
|
|
|
|
AwsTranscriberResultStability::Medium => "medium",
|
|
|
|
AwsTranscriberResultStability::Low => "low",
|
|
|
|
}
|
|
|
|
)
|
|
|
|
.as_str(),
|
2021-06-19 01:27:42 +00:00
|
|
|
);
|
|
|
|
|
2021-06-22 18:53:49 +00:00
|
|
|
drop(settings);
|
|
|
|
drop(state);
|
|
|
|
|
2022-05-17 15:25:08 +00:00
|
|
|
let signer = signer::SigV4Signer::new();
|
|
|
|
let mut operation_config = OperationSigningConfig::default_config();
|
|
|
|
operation_config.signature_type = HttpSignatureType::HttpRequestQueryParams;
|
|
|
|
operation_config.expires_in = Some(Duration::from_secs(5 * 60)); // See commit a3db85d.
|
|
|
|
|
|
|
|
let request_config = RequestConfig {
|
|
|
|
request_ts: SystemTime::from(current_time),
|
|
|
|
region: &SigningRegion::from(region.clone()),
|
|
|
|
service: &SigningService::from_static("transcribe"),
|
|
|
|
payload_override: None,
|
|
|
|
};
|
|
|
|
let transcribe_uri = Uri::builder()
|
|
|
|
.scheme("https")
|
|
|
|
.authority(format!("transcribestreaming.{}.amazonaws.com:8443", region).as_str())
|
|
|
|
.path_and_query(query_params.clone())
|
|
|
|
.build()
|
|
|
|
.map_err(|err| {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::error!(CAT, imp: self, "Failed to build HTTP request URI: {}", err);
|
2022-06-25 06:50:02 +00:00
|
|
|
error_msg!(
|
2022-05-17 15:25:08 +00:00
|
|
|
gst::CoreError::Failed,
|
|
|
|
["Failed to build HTTP request URI: {}", err]
|
2022-06-25 06:50:02 +00:00
|
|
|
)
|
2022-05-17 15:25:08 +00:00
|
|
|
})?;
|
|
|
|
let mut request = http::Request::builder()
|
|
|
|
.uri(transcribe_uri)
|
|
|
|
.body(SdkBody::empty())
|
|
|
|
.expect("Failed to build valid request");
|
|
|
|
let _signature = signer
|
|
|
|
.sign(
|
|
|
|
&operation_config,
|
|
|
|
&request_config,
|
2022-08-04 06:11:41 +00:00
|
|
|
&credentials.unwrap(),
|
2022-05-17 15:25:08 +00:00
|
|
|
&mut request,
|
|
|
|
)
|
|
|
|
.map_err(|err| {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::error!(CAT, imp: self, "Failed to sign HTTP request: {}", err);
|
2022-05-17 15:25:08 +00:00
|
|
|
error_msg!(
|
|
|
|
gst::CoreError::Failed,
|
|
|
|
["Failed to sign HTTP request: {}", err]
|
|
|
|
)
|
|
|
|
})?;
|
|
|
|
let url = request.uri().to_string();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-01-09 10:14:31 +00:00
|
|
|
let (ws, _) = {
|
|
|
|
let _enter = RUNTIME.enter();
|
|
|
|
futures::executor::block_on(connect_async(format!("wss{}", &url[5..]))).map_err(
|
|
|
|
|err| {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::error!(CAT, imp: self, "Failed to connect: {}", err);
|
2021-01-09 10:14:31 +00:00
|
|
|
error_msg!(gst::CoreError::Failed, ["Failed to connect: {}", err])
|
|
|
|
},
|
|
|
|
)?
|
|
|
|
};
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
let (ws_sink, mut ws_stream) = ws.split();
|
|
|
|
|
2020-12-29 15:28:19 +00:00
|
|
|
*self.ws_sink.borrow_mut() = Some(Box::pin(ws_sink));
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
let imp_weak = self.downgrade();
|
2020-04-09 22:51:03 +00:00
|
|
|
let future = async move {
|
2022-10-09 13:06:59 +00:00
|
|
|
while let Some(transcribe) = imp_weak.upgrade() {
|
2020-04-09 22:51:03 +00:00
|
|
|
let msg = match ws_stream.next().await {
|
|
|
|
Some(msg) => msg,
|
|
|
|
None => {
|
|
|
|
let mut state = transcribe.state.lock().unwrap();
|
|
|
|
state.send_eos = true;
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
let msg = match msg {
|
|
|
|
Ok(msg) => msg,
|
|
|
|
Err(err) => {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::error!(CAT, imp: transcribe, "Failed to receive data: {}", err);
|
|
|
|
element_imp_error!(
|
|
|
|
transcribe,
|
2020-04-09 22:51:03 +00:00
|
|
|
gst::StreamError::Failed,
|
|
|
|
["Streaming failed: {}", err]
|
|
|
|
);
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
let mut sender = transcribe.state.lock().unwrap().sender.clone();
|
|
|
|
|
|
|
|
if let Some(sender) = sender.as_mut() {
|
2020-04-10 17:52:28 +00:00
|
|
|
if sender.send(msg).await.is_err() {
|
|
|
|
break;
|
|
|
|
}
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
2021-06-22 18:53:49 +00:00
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
|
2020-04-09 22:51:03 +00:00
|
|
|
let (future, abort_handle) = abortable(future);
|
|
|
|
|
|
|
|
state.recv_abort_handle = Some(abort_handle);
|
|
|
|
|
|
|
|
RUNTIME.spawn(future);
|
|
|
|
|
|
|
|
state.connected = true;
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::info!(CAT, imp: self, "Connected");
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn disconnect(&self) {
|
2020-04-09 22:51:03 +00:00
|
|
|
let mut state = self.state.lock().unwrap();
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::info!(CAT, imp: self, "Unpreparing");
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
if let Some(abort_handle) = state.recv_abort_handle.take() {
|
|
|
|
abort_handle.abort();
|
|
|
|
}
|
|
|
|
|
|
|
|
if let Some(abort_handle) = state.send_abort_handle.take() {
|
|
|
|
abort_handle.abort();
|
|
|
|
}
|
|
|
|
|
|
|
|
*state = State::default();
|
|
|
|
|
2022-02-21 17:43:46 +00:00
|
|
|
gst::info!(
|
2020-04-09 22:51:03 +00:00
|
|
|
CAT,
|
2022-10-09 13:06:59 +00:00
|
|
|
imp: self,
|
2020-04-09 22:51:03 +00:00
|
|
|
"Unprepared, connected: {}!",
|
|
|
|
state.connected
|
|
|
|
);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-03-07 16:22:24 +00:00
|
|
|
#[glib::object_subclass]
|
2020-04-09 22:51:03 +00:00
|
|
|
impl ObjectSubclass for Transcriber {
|
|
|
|
const NAME: &'static str = "RsAwsTranscriber";
|
2020-11-14 17:24:01 +00:00
|
|
|
type Type = super::Transcriber;
|
2020-04-09 22:51:03 +00:00
|
|
|
type ParentType = gst::Element;
|
|
|
|
|
2020-11-14 17:24:01 +00:00
|
|
|
fn with_class(klass: &Self::Class) -> Self {
|
2021-04-20 12:58:11 +00:00
|
|
|
let templ = klass.pad_template("sink").unwrap();
|
2020-06-22 08:03:52 +00:00
|
|
|
let sinkpad = gst::Pad::builder_with_template(&templ, Some("sink"))
|
|
|
|
.chain_function(|pad, parent, buffer| {
|
|
|
|
Transcriber::catch_panic_pad_function(
|
|
|
|
parent,
|
|
|
|
|| Err(gst::FlowError::Error),
|
2022-10-09 13:06:59 +00:00
|
|
|
|transcriber| transcriber.sink_chain(pad, buffer),
|
2020-06-22 08:03:52 +00:00
|
|
|
)
|
|
|
|
})
|
|
|
|
.event_function(|pad, parent, event| {
|
|
|
|
Transcriber::catch_panic_pad_function(
|
|
|
|
parent,
|
|
|
|
|| false,
|
2022-10-09 13:06:59 +00:00
|
|
|
|transcriber| transcriber.sink_event(pad, event),
|
2020-06-22 08:03:52 +00:00
|
|
|
)
|
|
|
|
})
|
|
|
|
.build();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-04-20 12:58:11 +00:00
|
|
|
let templ = klass.pad_template("src").unwrap();
|
2020-06-22 08:03:52 +00:00
|
|
|
let srcpad = gst::Pad::builder_with_template(&templ, Some("src"))
|
|
|
|
.activatemode_function(|pad, parent, mode, active| {
|
|
|
|
Transcriber::catch_panic_pad_function(
|
|
|
|
parent,
|
2020-12-20 18:43:45 +00:00
|
|
|
|| Err(loggable_error!(CAT, "Panic activating src pad with mode")),
|
2022-10-09 13:06:59 +00:00
|
|
|
|transcriber| transcriber.src_activatemode(pad, mode, active),
|
2020-06-22 08:03:52 +00:00
|
|
|
)
|
|
|
|
})
|
|
|
|
.query_function(|pad, parent, query| {
|
|
|
|
Transcriber::catch_panic_pad_function(
|
|
|
|
parent,
|
|
|
|
|| false,
|
2022-10-09 13:06:59 +00:00
|
|
|
|transcriber| transcriber.src_query(pad, query),
|
2020-06-22 08:03:52 +00:00
|
|
|
)
|
|
|
|
})
|
|
|
|
.flags(gst::PadFlags::FIXED_CAPS)
|
|
|
|
.build();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
|
|
|
let settings = Mutex::new(Settings::default());
|
|
|
|
|
|
|
|
Self {
|
|
|
|
srcpad,
|
|
|
|
sinkpad,
|
|
|
|
settings,
|
|
|
|
state: Mutex::new(State::default()),
|
2020-12-29 15:28:19 +00:00
|
|
|
ws_sink: AtomicRefCell::new(None),
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
}
|
2021-01-21 18:21:29 +00:00
|
|
|
}
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-01-21 18:21:29 +00:00
|
|
|
impl ObjectImpl for Transcriber {
|
|
|
|
fn properties() -> &'static [glib::ParamSpec] {
|
|
|
|
static PROPERTIES: Lazy<Vec<glib::ParamSpec>> = Lazy::new(|| {
|
|
|
|
vec![
|
2022-08-18 12:04:15 +00:00
|
|
|
glib::ParamSpecString::builder("language-code")
|
|
|
|
.nick("Language Code")
|
|
|
|
.blurb("The Language of the Stream, see \
|
2021-01-21 18:21:29 +00:00
|
|
|
<https://docs.aws.amazon.com/transcribe/latest/dg/how-streaming-transcription.html> \
|
2022-08-18 12:04:15 +00:00
|
|
|
for an up to date list of allowed languages")
|
|
|
|
.default_value(Some("en-US"))
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
|
|
|
glib::ParamSpecUInt::builder("latency")
|
|
|
|
.nick("Latency")
|
|
|
|
.blurb("Amount of milliseconds to allow AWS transcribe")
|
|
|
|
.default_value(DEFAULT_LATENCY.mseconds() as u32)
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
|
|
|
glib::ParamSpecUInt::builder("lateness")
|
|
|
|
.nick("Lateness")
|
|
|
|
.blurb("Amount of milliseconds to introduce as lateness")
|
|
|
|
.default_value(DEFAULT_LATENESS.mseconds() as u32)
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
|
|
|
glib::ParamSpecString::builder("vocabulary-name")
|
|
|
|
.nick("Vocabulary Name")
|
|
|
|
.blurb("The name of a custom vocabulary, see \
|
2021-02-19 20:54:08 +00:00
|
|
|
<https://docs.aws.amazon.com/transcribe/latest/dg/how-vocabulary.html> \
|
2022-08-18 12:04:15 +00:00
|
|
|
for more information")
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
|
|
|
glib::ParamSpecString::builder("session-id")
|
|
|
|
.nick("Session ID")
|
|
|
|
.blurb("The ID of the transcription session, must be length 36")
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
2022-09-05 08:45:47 +00:00
|
|
|
glib::ParamSpecEnum::builder::<AwsTranscriberResultStability>("results-stability", DEFAULT_STABILITY)
|
2022-08-18 12:04:15 +00:00
|
|
|
.nick("Results stability")
|
|
|
|
.blurb("Defines how fast results should stabilize")
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
|
|
|
glib::ParamSpecString::builder("access-key")
|
|
|
|
.nick("Access Key")
|
|
|
|
.blurb("AWS Access Key")
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
|
|
|
glib::ParamSpecString::builder("secret-access-key")
|
|
|
|
.nick("Secret Access Key")
|
|
|
|
.blurb("AWS Secret Access Key")
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
|
|
|
glib::ParamSpecString::builder("session-token")
|
|
|
|
.nick("Session Token")
|
|
|
|
.blurb("AWS temporary Session Token from STS")
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
|
|
|
glib::ParamSpecString::builder("vocabulary-filter-name")
|
|
|
|
.nick("Vocabulary Filter Name")
|
|
|
|
.blurb("The name of a custom filter vocabulary, see \
|
2022-08-03 22:42:50 +00:00
|
|
|
<https://docs.aws.amazon.com/transcribe/latest/help-panel/vocab-filter.html> \
|
2022-08-18 12:04:15 +00:00
|
|
|
for more information")
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
2022-09-05 08:45:47 +00:00
|
|
|
glib::ParamSpecEnum::builder::<AwsTranscriberVocabularyFilterMethod>("vocabulary-filter-method", DEFAULT_VOCABULARY_FILTER_METHOD)
|
2022-08-18 12:04:15 +00:00
|
|
|
.nick("Vocabulary Filter Method")
|
|
|
|
.blurb("Defines how filtered words will be edited, has no effect when vocabulary-filter-name isn't set")
|
|
|
|
.mutable_ready()
|
|
|
|
.build(),
|
2021-01-21 18:21:29 +00:00
|
|
|
]
|
|
|
|
});
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2021-01-21 18:21:29 +00:00
|
|
|
PROPERTIES.as_ref()
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn constructed(&self) {
|
|
|
|
self.parent_constructed();
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
let obj = self.instance();
|
2020-11-14 17:24:01 +00:00
|
|
|
obj.add_pad(&self.sinkpad).unwrap();
|
|
|
|
obj.add_pad(&self.srcpad).unwrap();
|
|
|
|
obj.set_element_flags(gst::ElementFlags::PROVIDE_CLOCK | gst::ElementFlags::REQUIRE_CLOCK);
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn set_property(&self, _id: usize, value: &glib::Value, pspec: &glib::ParamSpec) {
|
2021-04-12 12:49:54 +00:00
|
|
|
match pspec.name() {
|
2022-09-08 20:07:49 +00:00
|
|
|
"language-code" => {
|
2020-04-09 22:51:03 +00:00
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
settings.language_code = value.get().expect("type checked upstream");
|
|
|
|
}
|
2021-01-21 18:21:29 +00:00
|
|
|
"latency" => {
|
2020-04-09 22:51:03 +00:00
|
|
|
let mut settings = self.settings.lock().unwrap();
|
2021-05-26 13:22:01 +00:00
|
|
|
settings.latency = gst::ClockTime::from_mseconds(
|
|
|
|
value.get::<u32>().expect("type checked upstream").into(),
|
|
|
|
);
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
2021-07-27 00:36:59 +00:00
|
|
|
"lateness" => {
|
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
settings.lateness = gst::ClockTime::from_mseconds(
|
|
|
|
value.get::<u32>().expect("type checked upstream").into(),
|
|
|
|
);
|
|
|
|
}
|
2021-02-19 20:54:08 +00:00
|
|
|
"vocabulary-name" => {
|
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
settings.vocabulary = value.get().expect("type checked upstream");
|
|
|
|
}
|
2021-06-16 22:53:04 +00:00
|
|
|
"session-id" => {
|
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
settings.session_id = value.get().expect("type checked upstream");
|
|
|
|
}
|
2021-06-19 01:27:42 +00:00
|
|
|
"results-stability" => {
|
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
settings.results_stability = value
|
|
|
|
.get::<AwsTranscriberResultStability>()
|
|
|
|
.expect("type checked upstream");
|
|
|
|
}
|
2021-09-27 13:49:12 +00:00
|
|
|
"access-key" => {
|
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
settings.access_key = value.get().expect("type checked upstream");
|
|
|
|
}
|
|
|
|
"secret-access-key" => {
|
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
settings.secret_access_key = value.get().expect("type checked upstream");
|
|
|
|
}
|
2022-06-16 07:16:28 +00:00
|
|
|
"session-token" => {
|
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
settings.session_token = value.get().expect("type checked upstream");
|
|
|
|
}
|
2022-08-03 22:42:50 +00:00
|
|
|
"vocabulary-filter-name" => {
|
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
settings.vocabulary_filter = value.get().expect("type checked upstream");
|
|
|
|
}
|
|
|
|
"vocabulary-filter-method" => {
|
|
|
|
let mut settings = self.settings.lock().unwrap();
|
|
|
|
settings.vocabulary_filter_method = value
|
|
|
|
.get::<AwsTranscriberVocabularyFilterMethod>()
|
|
|
|
.expect("type checked upstream");
|
|
|
|
}
|
2020-04-09 22:51:03 +00:00
|
|
|
_ => unimplemented!(),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn property(&self, _id: usize, pspec: &glib::ParamSpec) -> glib::Value {
|
2021-04-12 12:49:54 +00:00
|
|
|
match pspec.name() {
|
2021-01-21 18:21:29 +00:00
|
|
|
"language-code" => {
|
2020-04-09 22:51:03 +00:00
|
|
|
let settings = self.settings.lock().unwrap();
|
2020-11-19 15:55:57 +00:00
|
|
|
settings.language_code.to_value()
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
2021-01-21 18:21:29 +00:00
|
|
|
"latency" => {
|
2020-04-09 22:51:03 +00:00
|
|
|
let settings = self.settings.lock().unwrap();
|
2021-05-26 13:22:01 +00:00
|
|
|
(settings.latency.mseconds() as u32).to_value()
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
2021-07-27 00:36:59 +00:00
|
|
|
"lateness" => {
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
(settings.lateness.mseconds() as u32).to_value()
|
|
|
|
}
|
2021-02-19 20:54:08 +00:00
|
|
|
"vocabulary-name" => {
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
settings.vocabulary.to_value()
|
|
|
|
}
|
2021-06-16 22:53:04 +00:00
|
|
|
"session-id" => {
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
settings.session_id.to_value()
|
|
|
|
}
|
2021-06-19 01:27:42 +00:00
|
|
|
"results-stability" => {
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
settings.results_stability.to_value()
|
|
|
|
}
|
2021-09-27 13:49:12 +00:00
|
|
|
"access-key" => {
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
settings.access_key.to_value()
|
|
|
|
}
|
|
|
|
"secret-access-key" => {
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
settings.secret_access_key.to_value()
|
|
|
|
}
|
2022-06-16 07:16:28 +00:00
|
|
|
"session-token" => {
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
settings.session_token.to_value()
|
|
|
|
}
|
2022-08-03 22:42:50 +00:00
|
|
|
"vocabulary-filter-name" => {
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
settings.vocabulary_filter.to_value()
|
|
|
|
}
|
|
|
|
"vocabulary-filter-method" => {
|
|
|
|
let settings = self.settings.lock().unwrap();
|
|
|
|
settings.vocabulary_filter_method.to_value()
|
|
|
|
}
|
2020-04-09 22:51:03 +00:00
|
|
|
_ => unimplemented!(),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-10-23 08:57:31 +00:00
|
|
|
impl GstObjectImpl for Transcriber {}
|
|
|
|
|
2020-04-09 22:51:03 +00:00
|
|
|
impl ElementImpl for Transcriber {
|
2021-01-21 18:21:29 +00:00
|
|
|
fn metadata() -> Option<&'static gst::subclass::ElementMetadata> {
|
|
|
|
static ELEMENT_METADATA: Lazy<gst::subclass::ElementMetadata> = Lazy::new(|| {
|
|
|
|
gst::subclass::ElementMetadata::new(
|
|
|
|
"Transcriber",
|
|
|
|
"Audio/Text/Filter",
|
|
|
|
"Speech to Text filter, using AWS transcribe",
|
|
|
|
"Jordan Petridis <jordan@centricular.com>, Mathieu Duponchelle <mathieu@centricular.com>",
|
|
|
|
)
|
|
|
|
});
|
|
|
|
|
|
|
|
Some(&*ELEMENT_METADATA)
|
|
|
|
}
|
|
|
|
|
|
|
|
fn pad_templates() -> &'static [gst::PadTemplate] {
|
|
|
|
static PAD_TEMPLATES: Lazy<Vec<gst::PadTemplate>> = Lazy::new(|| {
|
|
|
|
let src_caps = gst::Caps::builder("text/x-raw")
|
2021-11-06 07:34:10 +00:00
|
|
|
.field("format", "utf8")
|
2021-01-21 18:21:29 +00:00
|
|
|
.build();
|
|
|
|
let src_pad_template = gst::PadTemplate::new(
|
|
|
|
"src",
|
|
|
|
gst::PadDirection::Src,
|
|
|
|
gst::PadPresence::Always,
|
|
|
|
&src_caps,
|
|
|
|
)
|
|
|
|
.unwrap();
|
|
|
|
|
2022-10-13 18:02:04 +00:00
|
|
|
let sink_caps = gst_audio::AudioCapsBuilder::new()
|
|
|
|
.format(gst_audio::AudioFormat::S16le)
|
|
|
|
.rate_range(8000..=48000)
|
|
|
|
.channels(1)
|
2021-01-21 18:21:29 +00:00
|
|
|
.build();
|
|
|
|
let sink_pad_template = gst::PadTemplate::new(
|
|
|
|
"sink",
|
|
|
|
gst::PadDirection::Sink,
|
|
|
|
gst::PadPresence::Always,
|
|
|
|
&sink_caps,
|
|
|
|
)
|
|
|
|
.unwrap();
|
|
|
|
|
|
|
|
vec![src_pad_template, sink_pad_template]
|
|
|
|
});
|
|
|
|
|
|
|
|
PAD_TEMPLATES.as_ref()
|
|
|
|
}
|
|
|
|
|
2020-04-09 22:51:03 +00:00
|
|
|
fn change_state(
|
|
|
|
&self,
|
|
|
|
transition: gst::StateChange,
|
|
|
|
) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
|
2022-10-09 13:06:59 +00:00
|
|
|
gst::info!(CAT, imp: self, "Changing state {:?}", transition);
|
2020-04-09 22:51:03 +00:00
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
let mut success = self.parent_change_state(transition)?;
|
2021-06-22 18:53:49 +00:00
|
|
|
|
2020-04-09 22:51:03 +00:00
|
|
|
match transition {
|
|
|
|
gst::StateChange::PausedToReady => {
|
2022-10-09 13:06:59 +00:00
|
|
|
self.disconnect();
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|
|
|
|
gst::StateChange::ReadyToPaused => {
|
|
|
|
success = gst::StateChangeSuccess::NoPreroll;
|
|
|
|
}
|
|
|
|
gst::StateChange::PlayingToPaused => {
|
|
|
|
success = gst::StateChangeSuccess::NoPreroll;
|
|
|
|
}
|
|
|
|
_ => (),
|
|
|
|
}
|
|
|
|
|
|
|
|
Ok(success)
|
|
|
|
}
|
2020-06-02 17:31:58 +00:00
|
|
|
|
2022-10-09 13:06:59 +00:00
|
|
|
fn provide_clock(&self) -> Option<gst::Clock> {
|
2020-06-02 17:31:58 +00:00
|
|
|
Some(gst::SystemClock::obtain())
|
|
|
|
}
|
2020-04-09 22:51:03 +00:00
|
|
|
}
|