meeting-assistant/MeetingAssistant/Transcription/AzureSpeechStreamingTranscriptionProvider.cs

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<AzureSpeechStreamingTranscriptionProvider> logger;

    public AzureSpeechStreamingTranscriptionProvider(
        IOptions<MeetingAssistantOptions> options,
        ILogger<AzureSpeechStreamingTranscriptionProvider> logger)
    {
        this.options = options.Value;
        this.logger = logger;
    }

    public async IAsyncEnumerable<TranscriptionSegment> TranscribeAsync(
        IAsyncEnumerable<AudioChunk> 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<TranscriptionSegment>();

        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<string>? 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<string> NormalizeDictationWords(IReadOnlyList<string>? 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<string> 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<AudioChunk> audio,
        PushAudioInputStream pushStream,
        ConversationTranscriber recognizer,
        ChannelWriter<TranscriptionSegment> 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) ??
            "";
    }
}