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)
    {
        var enumerator = audio.GetAsyncEnumerator(cancellationToken);
        try
        {
            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);
        }
        finally
        {
            try
            {
                await enumerator.DisposeAsync();
            }
            catch (NotSupportedException)
            {
                // Some diagnostic async enumerable sources expose DisposeAsync but throw after the stream has been read.
            }
        }
    }

    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 = CreateSpeechConfig(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");
        if (!string.IsNullOrWhiteSpace(azure.PostProcessingOption))
        {
            logger.LogWarning(
                "Azure Speech post-processing option {PostProcessingOption} is configured but skipped because the live Azure backend uses ConversationTranscriber; Speech SDK post-processing is documented for SpeechRecognizer.",
                azure.PostProcessingOption.Trim());
        }

        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)
    {
        return string.IsNullOrWhiteSpace(options.Endpoint)
            ? throw new InvalidOperationException("Azure Speech endpoint must be configured.")
            : new Uri(options.Endpoint);
    }

    internal static SpeechConfig CreateSpeechConfig(AzureSpeechOptions options, string key)
    {
        if (!string.IsNullOrWhiteSpace(options.Endpoint))
        {
            return SpeechConfig.FromEndpoint(GetEndpoint(options), key);
        }

        if (string.IsNullOrWhiteSpace(options.Region))
        {
            throw new InvalidOperationException("Azure Speech region or endpoint must be configured.");
        }

        return SpeechConfig.FromSubscription(key, options.Region.Trim());
    }

    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) ??
            "";
    }
}