using System.Threading.Channels; using MeetingAssistant.Recording; using Microsoft.CognitiveServices.Speech; using Microsoft.CognitiveServices.Speech.Audio; using Microsoft.CognitiveServices.Speech.Transcription; using Microsoft.Extensions.Options; namespace MeetingAssistant.Transcription; public sealed class AzureSpeechStreamingTranscriptionProvider : IStreamingTranscriptionProvider { private readonly MeetingAssistantOptions options; private readonly ILogger logger; public AzureSpeechStreamingTranscriptionProvider( IOptions options, ILogger logger) { this.options = options.Value; this.logger = logger; } public async IAsyncEnumerable TranscribeAsync( IAsyncEnumerable audio, SpeechRecognitionPipelineOptions pipelineOptions, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken) { await using var enumerator = audio.GetAsyncEnumerator(cancellationToken); if (!await enumerator.MoveNextAsync()) { yield break; } var firstChunk = enumerator.Current; using var pushStream = AudioInputStream.CreatePushStream( AudioStreamFormat.GetWaveFormatPCM( (uint)firstChunk.SampleRate, 16, (byte)firstChunk.Channels)); using var audioConfig = AudioConfig.FromStreamInput(pushStream); var speechConfig = CreateSpeechConfig(); using var recognizer = CreateTranscriber(speechConfig, audioConfig); AddPhraseList(recognizer, pipelineOptions.DictationWords); var segments = Channel.CreateUnbounded(); recognizer.Transcribed += (_, args) => { if (args.Result.Reason != ResultReason.RecognizedSpeech || string.IsNullOrWhiteSpace(args.Result.Text)) { return; } var start = TimeSpan.FromTicks(args.Result.OffsetInTicks); var segment = new TranscriptionSegment( start, start + args.Result.Duration, NormalizeSpeakerId(args.Result.SpeakerId), args.Result.Text.Trim()); logger.LogInformation( "Azure Speech emitted segment at {Start} from {Speaker}: {Text}", segment.Start, segment.Speaker, segment.Text); segments.Writer.TryWrite(segment); }; recognizer.Canceled += (_, args) => { if (args.Reason == CancellationReason.Error) { segments.Writer.TryComplete(new InvalidOperationException( $"Azure Speech recognition canceled: {args.ErrorCode} {args.ErrorDetails}")); return; } segments.Writer.TryComplete(); }; await recognizer.StartTranscribingAsync(); var pumpTask = Task.Run( () => PumpAudioAsync( firstChunk, enumerator, pushStream, recognizer, segments.Writer, options.AzureSpeech.RecognitionStopTimeout, cancellationToken), CancellationToken.None); await foreach (var segment in segments.Reader.ReadAllAsync(cancellationToken)) { yield return segment; } await pumpTask.WaitAsync(cancellationToken); } internal ConversationTranscriber CreateTranscriber(SpeechConfig speechConfig, AudioConfig audioConfig) { var languages = GetAutoDetectLanguages(options.AzureSpeech); if (languages.Count == 0) { return new ConversationTranscriber(speechConfig, audioConfig); } logger.LogInformation( "Azure Speech auto language detection enabled for {Languages}", string.Join(", ", languages)); return new ConversationTranscriber( speechConfig, AutoDetectSourceLanguageConfig.FromLanguages(languages.ToArray()), audioConfig); } internal void AddPhraseList(ConversationTranscriber recognizer, IReadOnlyList? words) { var phrases = NormalizeDictationWords(words); if (phrases.Count == 0) { return; } var phraseList = PhraseListGrammar.FromRecognizer(recognizer); foreach (var phrase in phrases) { phraseList.AddPhrase(phrase); } phraseList.SetWeight(options.AzureSpeech.PhraseListWeight); logger.LogInformation("Configured Azure Speech phrase list with {PhraseCount} phrase(s)", phrases.Count); } internal static IReadOnlyList NormalizeDictationWords(IReadOnlyList? words) { return (words ?? []) .Select(word => word.Trim()) .Where(word => !string.IsNullOrWhiteSpace(word)) .Distinct(StringComparer.OrdinalIgnoreCase) .ToList(); } internal SpeechConfig CreateSpeechConfig() { var azure = options.AzureSpeech; var key = ResolveKey(azure); if (string.IsNullOrWhiteSpace(key)) { throw new InvalidOperationException( $"Azure Speech key is not configured. Set AzureSpeech:Key or environment variable {azure.KeyEnv}."); } var speechConfig = SpeechConfig.FromEndpoint(GetEndpoint(azure), key); var autoDetectLanguages = GetAutoDetectLanguages(azure); if (autoDetectLanguages.Count == 0) { speechConfig.SpeechRecognitionLanguage = azure.Language; } else { speechConfig.SetProperty( PropertyId.SpeechServiceConnection_LanguageIdMode, NormalizeLanguageIdMode(azure.LanguageIdMode)); } speechConfig.OutputFormat = OutputFormat.Detailed; speechConfig.SetProperty( PropertyId.SpeechServiceResponse_DiarizeIntermediateResults, azure.DiarizeIntermediateResults ? "true" : "false"); return speechConfig; } internal static IReadOnlyList GetAutoDetectLanguages(AzureSpeechOptions options) { if (!options.Language.Equals("auto", StringComparison.OrdinalIgnoreCase)) { return []; } return options.AutoDetectLanguages .Select(language => language.Trim()) .Where(language => !string.IsNullOrWhiteSpace(language)) .Distinct(StringComparer.OrdinalIgnoreCase) .ToList(); } internal static Uri GetEndpoint(AzureSpeechOptions options) { if (!string.IsNullOrWhiteSpace(options.Endpoint)) { return new Uri(options.Endpoint); } if (string.IsNullOrWhiteSpace(options.Region)) { throw new InvalidOperationException("Azure Speech region or endpoint must be configured."); } return new Uri($"wss://{options.Region}.stt.speech.microsoft.com/speech/universal/v2"); } internal static string NormalizeLanguageIdMode(string? value) { return value?.Trim().Equals("AtStart", StringComparison.OrdinalIgnoreCase) == true ? "AtStart" : "Continuous"; } private static async Task PumpAudioAsync( AudioChunk firstChunk, IAsyncEnumerator audio, PushAudioInputStream pushStream, ConversationTranscriber recognizer, ChannelWriter segments, TimeSpan recognitionStopTimeout, CancellationToken cancellationToken) { try { WriteChunk(pushStream, firstChunk, firstChunk); while (await audio.MoveNextAsync()) { cancellationToken.ThrowIfCancellationRequested(); WriteChunk(pushStream, firstChunk, audio.Current); } pushStream.Close(); try { var stopTask = recognizer.StopTranscribingAsync(); if (recognitionStopTimeout > TimeSpan.Zero) { await stopTask.WaitAsync(recognitionStopTimeout, cancellationToken); } else { await stopTask.WaitAsync(cancellationToken); } } catch (TimeoutException) { // The SDK can outlive short diagnostic streams while waiting for final service events. } segments.TryComplete(); } catch (Exception exception) { segments.TryComplete(exception); } } private static void WriteChunk(PushAudioInputStream pushStream, AudioChunk expectedFormat, AudioChunk chunk) { if (chunk.SampleRate != expectedFormat.SampleRate || chunk.Channels != expectedFormat.Channels) { return; } pushStream.Write(chunk.Pcm); } private static string NormalizeSpeakerId(string? speakerId) { return string.IsNullOrWhiteSpace(speakerId) || speakerId.Equals("Unknown", StringComparison.OrdinalIgnoreCase) ? "Unknown" : speakerId; } internal static string ResolveKey(AzureSpeechOptions options) { if (!string.IsNullOrWhiteSpace(options.Key)) { return options.Key; } if (string.IsNullOrWhiteSpace(options.KeyEnv)) { return ""; } return Environment.GetEnvironmentVariable(options.KeyEnv) ?? Environment.GetEnvironmentVariable(options.KeyEnv, EnvironmentVariableTarget.User) ?? Environment.GetEnvironmentVariable(options.KeyEnv, EnvironmentVariableTarget.Machine) ?? ""; } }