diff --git a/examples/transcribe.rs b/examples/transcribe.rs index 8b6c63f..d7fc2f3 100644 --- a/examples/transcribe.rs +++ b/examples/transcribe.rs @@ -321,20 +321,20 @@ async fn start_conversation( .or_else(|| env::var("GOOGLE_TRANSCRIBE_REGION").ok()); let region = match region.as_deref() { - Some("global") => google_transcribe::transcribe::Region::Global, - Some("eu") => google_transcribe::transcribe::Region::Eu, - Some("us") => google_transcribe::transcribe::Region::Us, + Some("global") => google_transcribe::Region::Global, + Some("eu") => google_transcribe::Region::Eu, + Some("us") => google_transcribe::Region::Us, Some(invalid) => bail!( "Invalid GOOGLE_TRANSCRIBE_REGION '{}'. Must be one of: global, eu, us", invalid ), - None => google_transcribe::transcribe::Region::default(), + None => google_transcribe::Region::default(), }; // Check model/language/region feature support (including diarization): // https://docs.cloud.google.com/speech-to-text/docs/speech-to-text-supported-languages - let params = google_transcribe::transcribe::Params { + let params = google_transcribe::Params { model: provider_args.model.map(str::to_owned).unwrap_or_else(|| { env::var("GOOGLE_TRANSCRIBE_MODEL").unwrap_or_else(|_| "latest_long".to_owned()) }), diff --git a/services/google-transcribe/src/client.rs b/services/google-transcribe/src/client.rs index 3468f84..2959b26 100644 --- a/services/google-transcribe/src/client.rs +++ b/services/google-transcribe/src/client.rs @@ -1,157 +1,25 @@ //! Tonic usage inspiration from: //! -use std::error; -use std::{env, sync::Arc}; - -use anyhow::{Context, Result, anyhow}; +use anyhow::Result; use async_stream::{stream, try_stream}; -use context_switch_core::{AudioFormat, audio}; use futures::Stream; -use google_cloud_auth::credentials::AccessTokenCredentials; -use google_cloud_auth::credentials::service_account; -use google_cloud_token::TokenSource; +use tokio::sync::mpsc::UnboundedReceiver; +use tracing::debug; + use googleapis_tonic_google_cloud_speech_v2::google::cloud::speech::v2::recognition_config::DecodingConfig; -use googleapis_tonic_google_cloud_speech_v2::google::cloud::speech::v2::speech_client::SpeechClient; use googleapis_tonic_google_cloud_speech_v2::google::cloud::speech::v2::{ ExplicitDecodingConfig, RecognitionConfig, RecognitionFeatures, StreamingRecognitionConfig, StreamingRecognitionFeatures, StreamingRecognizeRequest, StreamingRecognizeResponse, SpeakerDiarizationConfig, - explicit_decoding_config, }; +use googleapis_tonic_google_cloud_speech_v2::google::cloud::speech::v2::explicit_decoding_config; use googleapis_tonic_google_cloud_speech_v2::google::cloud::speech::v2::streaming_recognize_request::StreamingRequest; -use tokio::sync::mpsc::UnboundedReceiver; -use tonic::transport; -use tracing::debug; -use crate::transcribe::Region; - -type Client = - googleapis_tonic_google_cloud_speech_v2::google::cloud::speech::v2::speech_client::SpeechClient< - tonic::service::interceptor::InterceptedService, - >; - -#[derive(Default)] -pub(crate) struct Config { - endpoint: &'static str, - location: &'static str, -} - -impl From for Config { - fn from(value: Region) -> Self { - match value { - Region::Global => Self { - endpoint: "https://speech.googleapis.com", - location: "global", - }, - Region::Eu => Self { - endpoint: "https://eu-speech.googleapis.com", - location: "eu", - }, - Region::Us => Self { - endpoint: "https://us-speech.googleapis.com", - location: "us", - }, - } - } -} - -#[derive(Clone)] -pub(crate) struct Host { - channel: tonic::transport::Channel, - token_source: Arc, - project_id: String, - location: String, -} - -#[derive(Debug)] -struct ServiceAccountTokenSource { - credentials: AccessTokenCredentials, -} - -#[async_trait::async_trait] -impl TokenSource for ServiceAccountTokenSource { - async fn token(&self) -> std::result::Result> { - let access_token = self.credentials.access_token().await?; - Ok(format!("Bearer {}", access_token.token)) - } -} +use context_switch_core::AudioFormat; +use context_switch_core::audio; -impl Host { - pub(crate) async fn new(params: Config) -> Result { - let credentials_path = env::var("GOOGLE_APPLICATION_CREDENTIALS") - .context("GOOGLE_APPLICATION_CREDENTIALS is not set")?; - let credentials_json = tokio::fs::read_to_string(&credentials_path) - .await - .with_context(|| { - format!( - "Failed to read GOOGLE_APPLICATION_CREDENTIALS from path: {credentials_path}" - ) - })?; - let credentials_value: serde_json::Value = serde_json::from_str(&credentials_json) - .with_context(|| { - format!( - "GOOGLE_APPLICATION_CREDENTIALS does not contain valid JSON: {credentials_path}" - ) - })?; - - let project_id = credentials_value - .get("project_id") - .and_then(serde_json::Value::as_str) - .context("project_id missing in GOOGLE_APPLICATION_CREDENTIALS JSON")? - .to_owned(); - - let credentials = service_account::Builder::new(credentials_value) - .build_access_token_credentials() - .context("Failed to build Google service-account credentials")?; - - let token_source: Arc = - Arc::new(ServiceAccountTokenSource { credentials }); - - let channel = transport::Channel::from_static(params.endpoint) - .tls_config(transport::ClientTlsConfig::new().with_webpki_roots())? - .connect() - .await?; - - Ok(Self { - channel, - token_source, - project_id, - location: params.location.to_owned(), - }) - } - - pub async fn client(&self) -> Result { - let inner = self.channel.clone(); - let token = self.token_source.token().await.map_err(|e| anyhow!(e))?; - let mut metadata_value = tonic::metadata::AsciiMetadataValue::try_from(token)?; - metadata_value.set_sensitive(true); - let interceptor = AuthInterceptor { metadata_value }; - let client = SpeechClient::with_interceptor(inner, interceptor); - Ok(TranscribeClient { - client, - project_id: self.project_id.clone(), - location: self.location.clone(), - }) - } -} - -#[derive(Clone)] -struct AuthInterceptor { - metadata_value: tonic::metadata::AsciiMetadataValue, -} - -impl tonic::service::Interceptor for AuthInterceptor { - fn call( - &mut self, - mut request: tonic::Request<()>, - ) -> std::result::Result, tonic::Status> { - request - .metadata_mut() - .insert("authorization", self.metadata_value.clone()); - Ok(request) - } -} +use crate::host::Client; /// A google transcribe client. Capable of streaming audio data in and transcribe results out. #[derive(Debug)] @@ -162,6 +30,14 @@ pub struct TranscribeClient { } impl TranscribeClient { + pub fn new(client: Client, project_id: String, location: String) -> Self { + Self { + client, + project_id, + location, + } + } + pub async fn transcribe<'a>( &mut self, model: &str, diff --git a/services/google-transcribe/src/host.rs b/services/google-transcribe/src/host.rs new file mode 100644 index 0000000..85f8b5b --- /dev/null +++ b/services/google-transcribe/src/host.rs @@ -0,0 +1,144 @@ +use std::env; +use std::error; +use std::sync::Arc; + +use anyhow::{Context, Result, anyhow}; +use google_cloud_auth::credentials::AccessTokenCredentials; +use google_cloud_auth::credentials::service_account; +use google_cloud_token::TokenSource; +use googleapis_tonic_google_cloud_speech_v2::google::cloud::speech::v2::speech_client::SpeechClient; +use tonic::service::interceptor; +use tonic::transport; + +use crate::Region; +use crate::client::TranscribeClient; + +pub type Client = + SpeechClient>; + +#[derive(Default)] +pub struct Config { + endpoint: &'static str, + location: &'static str, +} + +impl From for Config { + fn from(value: Region) -> Self { + match value { + Region::Global => Self { + endpoint: "https://speech.googleapis.com", + location: "global", + }, + Region::Eu => Self { + endpoint: "https://eu-speech.googleapis.com", + location: "eu", + }, + Region::Us => Self { + endpoint: "https://us-speech.googleapis.com", + location: "us", + }, + } + } +} + +#[derive(Clone)] +pub struct Host { + channel: transport::Channel, + token_source: Arc, + project_id: String, + location: String, +} + +impl Host { + pub async fn new(config: Config) -> Result { + let credentials_path = env::var("GOOGLE_APPLICATION_CREDENTIALS") + .context("GOOGLE_APPLICATION_CREDENTIALS is not set")?; + let credentials_json = tokio::fs::read_to_string(&credentials_path) + .await + .with_context(|| { + format!( + "Failed to read GOOGLE_APPLICATION_CREDENTIALS from path: {credentials_path}" + ) + })?; + let credentials_value: serde_json::Value = serde_json::from_str(&credentials_json) + .with_context(|| { + format!( + "GOOGLE_APPLICATION_CREDENTIALS does not contain valid JSON: {credentials_path}" + ) + })?; + + let project_id = credentials_value + .get("project_id") + .and_then(serde_json::Value::as_str) + .context("project_id missing in GOOGLE_APPLICATION_CREDENTIALS JSON")? + .to_owned(); + + let credentials = service_account::Builder::new(credentials_value) + .build_access_token_credentials() + .context("Failed to build Google service-account credentials")?; + + let token_source: Arc = + Arc::new(ServiceAccountTokenSource { credentials }); + + let channel = transport::Channel::from_static(config.endpoint) + .tls_config(transport::ClientTlsConfig::new().with_webpki_roots())? + .connect() + .await?; + + Ok(Self { + channel, + token_source, + project_id, + location: config.location.to_owned(), + }) + } + + pub async fn client(&self) -> Result { + let token = self + .token_source + .token() + .await + .map_err(|error| anyhow!(error))?; + let mut metadata_value = tonic::metadata::AsciiMetadataValue::try_from(token)?; + metadata_value.set_sensitive(true); + let client = SpeechClient::with_interceptor( + self.channel.clone(), + AuthInterceptor { metadata_value }, + ); + Ok(TranscribeClient::new( + client, + self.project_id.clone(), + self.location.clone(), + )) + } +} + +#[derive(Debug)] +struct ServiceAccountTokenSource { + credentials: AccessTokenCredentials, +} + +#[async_trait::async_trait] +impl TokenSource for ServiceAccountTokenSource { + async fn token(&self) -> std::result::Result> { + let access_token = self.credentials.access_token().await?; + Ok(format!("Bearer {}", access_token.token)) + } +} + +#[derive(Clone)] +pub struct AuthInterceptor { + metadata_value: tonic::metadata::AsciiMetadataValue, +} + +impl tonic::service::Interceptor for AuthInterceptor { + fn call( + &mut self, + mut request: tonic::Request<()>, + ) -> std::result::Result, tonic::Status> { + request + .metadata_mut() + .insert("authorization", self.metadata_value.clone()); + Ok(request) + } +} diff --git a/services/google-transcribe/src/lib.rs b/services/google-transcribe/src/lib.rs index ca1c013..cc0e5c6 100644 --- a/services/google-transcribe/src/lib.rs +++ b/services/google-transcribe/src/lib.rs @@ -1,6 +1,37 @@ //! A Google Speech to Text V2 service. +use serde::Deserialize; mod client; +mod host; pub mod transcribe; -pub(crate) use client::Host; + pub use transcribe::GoogleTranscribe; + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Params { + /// Google Cloud Speech-to-Text `V2` recognition model (for example, `latest_long`). + pub model: String, + /// One or more comma-separated BCP 47 locale codes sent as `language_codes`. + pub language: String, + /// Enable speaker diarization. Google determines the number of speakers; support depends on + /// the selected model, language, and region. + #[serde(default)] + pub diarization: bool, + /// Google Cloud location and API endpoint. Only `global`, `eu`, and `us` are supported. + /// Defaults to `global`. + #[serde(default)] + pub region: Region, +} + +#[derive(Debug, Clone, Copy, Default, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Region { + /// Google Cloud global endpoint. + #[default] + Global, + /// Google Cloud European endpoint. + Eu, + /// Google Cloud United States endpoint. + Us, +} diff --git a/services/google-transcribe/src/transcribe.rs b/services/google-transcribe/src/transcribe.rs index 86bfd38..2ab3e9e 100644 --- a/services/google-transcribe/src/transcribe.rs +++ b/services/google-transcribe/src/transcribe.rs @@ -1,50 +1,25 @@ +use std::collections::HashMap; + use anyhow::{Context, Result}; use async_trait::async_trait; use futures::{Stream, StreamExt}; +use tokio::sync::mpsc::UnboundedReceiver; +use tracing::{info, warn}; + use googleapis_tonic_google_cloud_speech_v2::google::cloud::speech::v2::{ StreamingRecognizeResponse, WordInfo, streaming_recognize_response::SpeechEventType, }; -use serde::Deserialize; -use std::collections::HashMap; -use tokio::sync::mpsc::UnboundedReceiver; use tonic::Code; +use context_switch_core::language::Languages; use context_switch_core::{ AudioFormat, AudioFrame, AudioProducer, BillingRecord, BillingSchedule, Conversation, - ConversationOutput, Input, OutputModality, Service, language::Languages, + ConversationOutput, Input, OutputModality, Service, }; -use tracing::{info, warn}; -use crate::{Host, client::TranscribeClient}; - -#[derive(Debug, Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct Params { - /// Google Cloud Speech-to-Text V2 recognition model (for example, `latest_long`). - pub model: String, - /// One or more comma-separated BCP 47 locale codes sent as `language_codes`. - pub language: String, - /// Enable speaker diarization. Google determines the number of speakers; support depends on - /// the selected model, language, and region. - #[serde(default)] - pub diarization: bool, - /// Google Cloud location and API endpoint. Only `global`, `eu`, and `us` are supported. - /// Defaults to `global`. - #[serde(default)] - pub region: Region, -} - -#[derive(Debug, Clone, Copy, Default, Deserialize)] -#[serde(rename_all = "lowercase")] -pub enum Region { - /// Google Cloud global endpoint. - #[default] - Global, - /// Google Cloud European endpoint. - Eu, - /// Google Cloud United States endpoint. - Us, -} +use crate::Params; +use crate::client::TranscribeClient; +use crate::host::Host; #[derive(Debug)] pub struct GoogleTranscribe;