native-whisperx 0.1.14

//! User-facing transcription workflow orchestration and output reporting.

use std::time::Instant;

mod execution;
mod multi_input;
mod progress;

use crate::config::{
    resolve_automatic_workflow_selection, AsrProvider, NativeWhisperxConfig, NativeWhisperxError,
    NativeWhisperxReport, NativeWorkflowSelectionReport, VadMethod,
};
use crate::config_mapping::{
    build_transcription_request, build_transcription_request_from_resolved_config,
    run_native_with_selected_vad_and_progress,
};
use crate::output::write_outputs_with_options;
use crate::report::{
    append_automatic_workflow_selection_diagnostics, append_native_alignment_diagnostics,
    append_native_diarization_diagnostics,
};

pub(crate) use execution::{run_with_phase_observer, run_with_progress_observer};
pub use multi_input::{
    run_many, run_many_reusing_native_provider, run_many_with_control, run_many_with_observer,
};
pub use progress::{
    CancellationHandle, FiniteCancellation, FiniteTranscriptionOutcome,
    MultiInputTranscriptionOutcome, NoopTranscriptionProgressObserver, TranscriptionProgressEvent,
    TranscriptionProgressObserver, TranscriptionProgressTask, UnfinishedTranscription,
};
pub(crate) use progress::{NativeProgressContext, ProgressTaskTracker};

pub fn run_live_asr_window(
    config: NativeWhisperxConfig,
) -> Result<crate::TranscriptionPipelineResponse, NativeWhisperxError> {
    validate_live_asr_window_config(&config)?;
    let request = build_transcription_request(&config)?;
    run_with_phase_observer(request, &config)
}

/// Runs one bounded native ASR Near-Live Window with operational progress.
///
/// Cancellation belongs to the surrounding live session so a cancellation
/// request cannot interrupt this bounded model operation. The session observes
/// the shared handle again after this function reaches its safe boundary.
pub fn run_live_asr_window_with_observer(
    config: NativeWhisperxConfig,
    session_id: &str,
    window_index: usize,
    observer: &mut dyn crate::LiveTranscriptionProgressObserver,
) -> Result<crate::TranscriptionPipelineResponse, NativeWhisperxError> {
    validate_live_asr_window_config(&config)?;
    let request = build_transcription_request(&config)?;
    let cancellation = CancellationHandle::new();
    let mut task_tracker = ProgressTaskTracker::default();
    let mut bridge = LiveProgressBridge {
        session_id,
        window_index,
        observer,
    };
    run_with_progress_observer(
        request,
        &config,
        Some(NativeProgressContext {
            observer: &mut bridge,
            file_index: window_index,
            task_tracker: &mut task_tracker,
            cancellation: &cancellation,
        }),
    )
}

fn validate_live_asr_window_config(
    config: &NativeWhisperxConfig,
) -> Result<(), NativeWhisperxError> {
    if config.asr.provider != AsrProvider::Native {
        return Err(NativeWhisperxError::InvalidConfig(
            "live-transcribe supports native ASR only".to_string(),
        ));
    }
    if config.translation.enabled {
        return Err(NativeWhisperxError::InvalidConfig(
            "live-transcribe does not support translation in the first live workflow".to_string(),
        ));
    }
    if config.alignment.enabled {
        return Err(NativeWhisperxError::InvalidConfig(
            "live-transcribe does not support alignment in the first live workflow".to_string(),
        ));
    }
    if config.diarization.enabled {
        return Err(NativeWhisperxError::InvalidConfig(
            "live-transcribe does not support diarization in the first live workflow".to_string(),
        ));
    }
    Ok(())
}

struct LiveProgressBridge<'a> {
    session_id: &'a str,
    window_index: usize,
    observer: &'a mut dyn crate::LiveTranscriptionProgressObserver,
}

impl TranscriptionProgressObserver for LiveProgressBridge<'_> {
    fn observe(&mut self, event: TranscriptionProgressEvent) {
        let event = match event {
            TranscriptionProgressEvent::ModelResolutionStart {
                provider, model_id, ..
            } => crate::LiveTranscriptionProgressEvent::ModelResolutionStart {
                session_id: self.session_id.to_string(),
                window_index: self.window_index,
                provider,
                model_id,
            },
            TranscriptionProgressEvent::ModelResolutionEnd {
                provider,
                model_id,
                source,
                ..
            } => crate::LiveTranscriptionProgressEvent::ModelResolutionEnd {
                session_id: self.session_id.to_string(),
                window_index: self.window_index,
                provider,
                model_id,
                source,
            },
            TranscriptionProgressEvent::ModelDownloadStart {
                provider, model_id, ..
            } => crate::LiveTranscriptionProgressEvent::ModelDownloadStart {
                session_id: self.session_id.to_string(),
                window_index: self.window_index,
                provider,
                model_id,
            },
            TranscriptionProgressEvent::ModelDownloadEnd {
                provider,
                model_id,
                duration_seconds,
                ..
            } => crate::LiveTranscriptionProgressEvent::ModelDownloadEnd {
                session_id: self.session_id.to_string(),
                window_index: self.window_index,
                provider,
                model_id,
                duration_seconds,
            },
            TranscriptionProgressEvent::ModelLoadStart {
                provider, model_id, ..
            } => crate::LiveTranscriptionProgressEvent::ModelLoadStart {
                session_id: self.session_id.to_string(),
                window_index: self.window_index,
                provider,
                model_id,
            },
            TranscriptionProgressEvent::ModelLoadEnd {
                provider,
                model_id,
                duration_seconds,
                ..
            } => crate::LiveTranscriptionProgressEvent::ModelLoadEnd {
                session_id: self.session_id.to_string(),
                window_index: self.window_index,
                provider,
                model_id,
                duration_seconds,
            },
            TranscriptionProgressEvent::ModelReuse {
                provider, model_id, ..
            } => crate::LiveTranscriptionProgressEvent::ModelReuse {
                session_id: self.session_id.to_string(),
                window_index: self.window_index,
                provider,
                model_id,
            },
            _ => return,
        };
        self.observer.observe(event);
    }
}

pub fn run(config: NativeWhisperxConfig) -> Result<NativeWhisperxReport, NativeWhisperxError> {
    let mut observer = NoopTranscriptionProgressObserver;
    run_with_observer(config, &mut observer)
}

pub fn run_with_observer(
    config: NativeWhisperxConfig,
    observer: &mut dyn TranscriptionProgressObserver,
) -> Result<NativeWhisperxReport, NativeWhisperxError> {
    let cancellation = CancellationHandle::new();
    run_one_with_observer(config, 0, 1, observer, true, &cancellation)
}

pub(crate) fn run_one_with_observer(
    config: NativeWhisperxConfig,
    file_index: usize,
    total_files: usize,
    observer: &mut dyn TranscriptionProgressObserver,
    emit_run_events: bool,
    cancellation: &CancellationHandle,
) -> Result<NativeWhisperxReport, NativeWhisperxError> {
    match run_one_with_control(
        config,
        file_index,
        total_files,
        observer,
        emit_run_events,
        cancellation,
    )? {
        FiniteTranscriptionOutcome::Completed(report) => Ok(*report),
        FiniteTranscriptionOutcome::Cancelled(_) => {
            unreachable!("the compatibility finite entry point uses an uncancelled handle")
        }
    }
}

/// Runs one finite Workflow Composition with progress and cooperative control.
pub fn run_with_control(
    config: NativeWhisperxConfig,
    observer: &mut dyn TranscriptionProgressObserver,
    cancellation: &CancellationHandle,
) -> Result<FiniteTranscriptionOutcome, NativeWhisperxError> {
    run_one_with_control(config, 0, 1, observer, true, cancellation)
}

pub(crate) fn run_one_with_control(
    config: NativeWhisperxConfig,
    file_index: usize,
    total_files: usize,
    observer: &mut dyn TranscriptionProgressObserver,
    emit_run_events: bool,
    cancellation: &CancellationHandle,
) -> Result<FiniteTranscriptionOutcome, NativeWhisperxError> {
    let run_started = Instant::now();
    let input = progress_input_path(&config);
    if emit_run_events {
        observer.observe(TranscriptionProgressEvent::RunStart { total_files });
    }
    observer.observe(TranscriptionProgressEvent::FileStart {
        file_index,
        total_files,
        input: input.clone(),
    });
    let mut task_tracker = ProgressTaskTracker::default();
    let result: Result<NativeWhisperxReport, NativeWhisperxError> = (|| {
        ensure_active(cancellation)?;
        let selection = resolve_automatic_workflow_selection(&config)?;
        let resolved_config = selection.config.clone();
        ensure_active(cancellation)?;
        let request = build_transcription_request_from_resolved_config(&resolved_config)?;
        ensure_active(cancellation)?;
        let mut response = if resolved_config.asr.provider == AsrProvider::Native
            && resolved_config.translation.enabled
        {
            crate::run_native_with_translation_with_progress(
                request,
                &resolved_config,
                Some(NativeProgressContext {
                    observer,
                    file_index,
                    task_tracker: &mut task_tracker,
                    cancellation,
                }),
            )?
        } else if resolved_config.asr.provider == AsrProvider::Native
            && matches!(
                resolved_config.vad.method,
                VadMethod::Silero | VadMethod::Pyannote
            )
        {
            run_native_with_selected_vad_and_progress(
                request,
                &resolved_config,
                Some(NativeProgressContext {
                    observer,
                    file_index,
                    task_tracker: &mut task_tracker,
                    cancellation,
                }),
            )?
        } else {
            run_with_progress_observer(
                request,
                &resolved_config,
                Some(NativeProgressContext {
                    observer,
                    file_index,
                    task_tracker: &mut task_tracker,
                    cancellation,
                }),
            )?
        };
        append_automatic_workflow_selection_diagnostics(&mut response, &selection);
        append_native_alignment_diagnostics(&mut response, &resolved_config);
        append_native_diarization_diagnostics(&mut response, &resolved_config);
        ensure_active(cancellation)?;
        crate::save_draft_speakers_from_response(&mut response, &resolved_config)?;
        ensure_active(cancellation)?;
        let (output_files, output_seconds) = write_outputs_with_control(
            &response,
            &resolved_config.output,
            resolved_config.alignment.return_char_alignments,
            file_index,
            observer,
            cancellation,
            &mut task_tracker,
        )?;
        response
            .diagnostics
            .push(format!("phaseOutputSeconds={:.6}", output_seconds));
        let total_seconds = run_started.elapsed().as_secs_f64();
        response
            .diagnostics
            .push(format!("phaseNativeTotalSeconds={:.6}", total_seconds));
        observer.observe(TranscriptionProgressEvent::FileEnd {
            file_index,
            total_files,
            input: input.clone(),
            duration_seconds: total_seconds,
        });
        if emit_run_events {
            observer.observe(TranscriptionProgressEvent::RunEnd {
                total_files,
                duration_seconds: total_seconds,
            });
        }
        Ok(NativeWhisperxReport {
            response,
            output_files,
            workflow_selection: NativeWorkflowSelectionReport::from_selection(&selection),
        })
    })();

    if result.is_err() && cancellation.is_cancelled() {
        let cancelled = FiniteCancellation::new(file_index, input.clone(), task_tracker.current());
        observer.observe(TranscriptionProgressEvent::Cancelled {
            file_index,
            input,
            task: cancelled.task(),
            duration_seconds: run_started.elapsed().as_secs_f64(),
        });
        return Ok(FiniteTranscriptionOutcome::Cancelled(cancelled));
    }

    match result {
        Ok(report) => Ok(FiniteTranscriptionOutcome::Completed(Box::new(report))),
        Err(error) => {
            observer.observe(TranscriptionProgressEvent::Failure {
                file_index,
                input,
                task: task_tracker.current(),
                duration_seconds: run_started.elapsed().as_secs_f64(),
                message: error.to_string(),
            });
            Err(error)
        }
    }
}

pub(crate) fn ensure_active(cancellation: &CancellationHandle) -> Result<(), NativeWhisperxError> {
    if cancellation.is_cancelled() {
        return Err(NativeWhisperxError::Transcription(
            "transcription cancelled at a safe workflow boundary".to_string(),
        ));
    }
    Ok(())
}

pub(crate) fn write_outputs_with_control(
    response: &crate::TranscriptionPipelineResponse,
    output: &crate::config::OutputConfig,
    return_char_alignments: bool,
    file_index: usize,
    observer: &mut dyn TranscriptionProgressObserver,
    cancellation: &CancellationHandle,
    task_tracker: &mut ProgressTaskTracker,
) -> Result<(Vec<crate::config::OutputFile>, f64), NativeWhisperxError> {
    let output_started = Instant::now();
    task_tracker.set_current(Some(TranscriptionProgressTask::Output));
    observer.observe(TranscriptionProgressEvent::TaskStart {
        file_index,
        task: TranscriptionProgressTask::Output,
    });
    ensure_active(cancellation)?;
    let output_files = write_outputs_with_options(response, output, return_char_alignments)?;
    let output_seconds = output_started.elapsed().as_secs_f64();
    observer.observe(TranscriptionProgressEvent::TaskEnd {
        file_index,
        task: TranscriptionProgressTask::Output,
        duration_seconds: output_seconds,
    });
    task_tracker.set_current(None);
    Ok((output_files, output_seconds))
}

pub(crate) fn progress_input_path(config: &NativeWhisperxConfig) -> std::path::PathBuf {
    match &config.input {
        crate::config::InputSource::Path { path } => path.clone(),
        crate::config::InputSource::Samples { source, .. } => source
            .as_ref()
            .map(std::path::PathBuf::from)
            .unwrap_or_else(|| std::path::PathBuf::from("<samples>")),
    }
}