using MeetingAssistant.LaunchProfiles; using MeetingAssistant.MeetingNotes; using MeetingAssistant.Summary; using MeetingAssistant.Transcription; using MeetingAssistant.Workflow; using Microsoft.Extensions.Options; using NAudio.Wave; namespace MeetingAssistant.Recording; public sealed class OfflineTranscriptionBacklogProcessor { private const int MaxChunkDurationMilliseconds = 1000; private readonly IOfflineTranscriptionBacklog backlog; private readonly ISpeechRecognitionPipelineFactory pipelineFactory; private readonly ITranscriptStore transcriptStore; private readonly IMeetingNoteStore meetingNoteStore; private readonly IMeetingArtifactStore meetingArtifactStore; private readonly IMeetingSummaryPipeline summaryPipeline; private readonly IMeetingWorkflowEngine workflowEngine; private readonly IRecordingDictationWordProvider dictationWordProvider; private readonly MeetingAssistantOptions options; private readonly ILogger logger; private readonly ILaunchProfileOptionsProvider? launchProfiles; public OfflineTranscriptionBacklogProcessor( IOfflineTranscriptionBacklog backlog, ISpeechRecognitionPipelineFactory pipelineFactory, ITranscriptStore transcriptStore, IMeetingNoteStore meetingNoteStore, IMeetingArtifactStore meetingArtifactStore, IMeetingSummaryPipeline summaryPipeline, IMeetingWorkflowEngine workflowEngine, IRecordingDictationWordProvider dictationWordProvider, IOptions options, ILogger logger, ILaunchProfileOptionsProvider? launchProfiles = null) { this.backlog = backlog; this.pipelineFactory = pipelineFactory; this.transcriptStore = transcriptStore; this.meetingNoteStore = meetingNoteStore; this.meetingArtifactStore = meetingArtifactStore; this.summaryPipeline = summaryPipeline; this.workflowEngine = workflowEngine; this.dictationWordProvider = dictationWordProvider; this.options = options.Value; this.logger = logger; this.launchProfiles = launchProfiles; } public async Task ProcessPendingAsync(CancellationToken cancellationToken) { var processed = 0; foreach (var item in await backlog.ListAsync(cancellationToken)) { cancellationToken.ThrowIfCancellationRequested(); if (await TryProcessAsync(item, cancellationToken)) { processed++; } } return processed; } private async Task TryProcessAsync( OfflineTranscriptionBacklogItem item, CancellationToken cancellationToken) { try { await ProcessAsync(item, cancellationToken); await backlog.CompleteAsync(item.Id, cancellationToken); logger.LogInformation( "Completed offline transcription backlog item {BacklogItemId} for recording {RecordingPath}", item.Id, item.AudioPath); return true; } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { throw; } catch (Exception exception) { logger.LogWarning( exception, "Could not process offline transcription backlog item {BacklogItemId}; it will be retried later", item.Id); return false; } } private async Task ProcessAsync( OfflineTranscriptionBacklogItem item, CancellationToken cancellationToken) { var runOptions = ResolveOptions(item.LaunchProfileName); var artifacts = new MeetingSessionArtifacts( item.MeetingNotePath, item.TranscriptPath, item.AssistantContextPath, item.SummaryPath); var session = new TranscriptSession(item.TranscriptPath); var meetingNote = await meetingNoteStore.ReadAsync(item.MeetingNotePath, cancellationToken); var pipelineOptions = await RecordingSpeechRecognitionPipelineOptions.BuildAsync( dictationWordProvider, runOptions, meetingNote, cancellationToken); await using var pipeline = pipelineFactory.Create(item.LaunchProfileName); await pipeline.InitializeAsync(pipelineOptions, cancellationToken); await pipeline.WaitUntilReadyAsync(cancellationToken); await WriteRecordingToPipelineAsync(item.AudioPath, pipeline, cancellationToken); var finishedSegments = await pipeline.ReadFinishedTranscriptAsync( item.AudioPath, pipelineOptions, cancellationToken); var lines = await TransformTranscriptLinesAsync(artifacts, runOptions, finishedSegments, cancellationToken); await transcriptStore.ReplaceLinesAsync(session, lines, cancellationToken); meetingNote.Frontmatter.EndTime = item.InferredEndTime ?? DateTimeOffset.Now; var completedMeetingNote = await meetingNoteStore.SaveAsync(meetingNote, runOptions, cancellationToken); await transcriptStore.UpdateMetadataAsync(session, artifacts, completedMeetingNote, cancellationToken); await meetingArtifactStore.UpdateAssistantContextMeetingAsync(artifacts, completedMeetingNote, cancellationToken); await TransitionMeetingAsync( artifacts, runOptions, AssistantContextState.Transcribing, AssistantContextState.Summarizing, cancellationToken); var summaryResult = await summaryPipeline.RunAsync(artifacts, runOptions, cancellationToken); await TransitionMeetingAsync( artifacts, runOptions, AssistantContextState.Summarizing, summaryResult.Succeeded ? AssistantContextState.Finished : AssistantContextState.Error, cancellationToken); } private MeetingAssistantOptions ResolveOptions(string? launchProfileName) { if (launchProfiles is not null) { return launchProfiles.GetRequiredProfile(launchProfileName).Options; } return options; } private async Task> TransformTranscriptLinesAsync( MeetingSessionArtifacts artifacts, MeetingAssistantOptions runOptions, IReadOnlyList segments, CancellationToken cancellationToken) { var lines = new List(segments.Count); foreach (var segment in segments) { var formatted = TranscriptLineFormatter.Format(segment); lines.Add(await TransformTranscriptLineAsync(artifacts, runOptions, formatted, cancellationToken)); } return lines; } private async Task TransformTranscriptLineAsync( MeetingSessionArtifacts artifacts, MeetingAssistantOptions runOptions, FormattedTranscriptLine formatted, CancellationToken cancellationToken) { try { return await workflowEngine.TransformTranscriptLineAsync( MeetingWorkflowEvent.TranscriptLine(artifacts, formatted.Line, formatted.Speaker), runOptions, cancellationToken); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { throw; } catch (Exception exception) { logger.LogError( exception, "Offline transcript line workflow failed for speaker {Speaker}; keeping original transcript line", formatted.Speaker); return formatted.Line; } } private async Task TransitionMeetingAsync( MeetingSessionArtifacts artifacts, MeetingAssistantOptions runOptions, AssistantContextState from, AssistantContextState to, CancellationToken cancellationToken) { await meetingArtifactStore.UpdateAssistantContextStateAsync(artifacts, to, cancellationToken); await workflowEngine.RunAsync( MeetingWorkflowEvent.StateTransition(artifacts, from, to), runOptions, cancellationToken); } private static async Task WriteRecordingToPipelineAsync( string audioPath, ISpeechRecognitionPipeline pipeline, CancellationToken cancellationToken) { using var reader = new WaveFileReader(audioPath); var bytesPerSecond = reader.WaveFormat.AverageBytesPerSecond; var chunkSize = Math.Max( reader.WaveFormat.BlockAlign, bytesPerSecond * MaxChunkDurationMilliseconds / 1000); chunkSize -= chunkSize % reader.WaveFormat.BlockAlign; var buffer = new byte[chunkSize]; while (true) { cancellationToken.ThrowIfCancellationRequested(); var read = await reader.ReadAsync(buffer.AsMemory(0, buffer.Length), cancellationToken); if (read == 0) { break; } await pipeline.WriteAsync( new AudioChunk( buffer[..read], reader.WaveFormat.SampleRate, reader.WaveFormat.Channels), cancellationToken); } } }