Skip to content

AI dubbing

videopython.ai.dubbing — translate speech, clone the voice, and re-time the dub onto the source. Whisper for transcription, an Ollama model for translation, Chatterbox for TTS, Demucs for source separation. Task recipes are in Dub a video into another language.

VideoDubber

Entry points:

Method Input → output Notes
dub_file(input_path, output_path, ...) path → file Never loads frames; video is stream-copied
dub(video, ...) Video → DubbingResult
dub_and_replace(video, ...) Video → Video Convenience over dub
revoice(video, text, ...) / revoice_and_replace(...) Video → result / Video New words, original voice

dub, dub_and_replace and dub_file all accept a pre-computed transcription. Speaker labels on it drive per-speaker voice cloning; the diarize-on-supplied path needs word-level timings, so SRT-loaded transcriptions (one synthetic word per block) are rejected.

dub_file copies subtitle streams through automatically and gain-matches the dub to the source with BS.1770 integrated loudness (pyloudnorm; falls back to peak match under 400 ms, post-gain peaks clamped to 0.99). keep_original_audio=True retains the source audio as a secondary track.

VideoDubber

Dubs videos into different languages using the local pipeline.

Accepts either a :class:DubbingConfig or the same knobs as flat kwargs (device, low_memory, whisper_model, translator_model, etc.) -- the flat path builds a DubbingConfig internally. See :class:DubbingConfig for the full knob list and defaults.

Source code in src/videopython/ai/dubbing/dubber.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
class VideoDubber:
    """Dubs videos into different languages using the local pipeline.

    Accepts either a :class:`DubbingConfig` or the same knobs as flat kwargs
    (``device``, ``low_memory``, ``whisper_model``, ``translator_model``, etc.)
    -- the flat path builds a ``DubbingConfig`` internally. See
    :class:`DubbingConfig` for the full knob list and defaults.
    """

    def __init__(
        self,
        config: DubbingConfig | None = None,
        *,
        tts_backend: SpeechBackend | None = None,
        **kwargs: Any,
    ):
        self.config = DubbingConfig.from_args(config, **kwargs)
        # Optional injected speech backend. None -> the pipeline lazily builds
        # the local chatterbox-backed TextToSpeech (from the [ai] extra). Inject
        # a SpeechBackend to run synthesis out-of-process (e.g. a remote/Modal
        # function) without loading chatterbox here.
        self._tts_backend = tts_backend
        self._local_pipeline: Any = None
        logger.info(
            "VideoDubber initialized with %s",
            " ".join(f"{k}={v}" for k, v in self.config.init_log_fields().items()),
        )

    def _init_local_pipeline(self) -> None:
        from videopython.ai.dubbing.pipeline import LocalDubbingPipeline

        self._local_pipeline = LocalDubbingPipeline(config=self.config, tts_backend=self._tts_backend)

    def dub(
        self,
        video: Video,
        target_lang: str,
        source_lang: str | None = None,
        preserve_background: bool = True,
        voice_clone: bool = True,
        enable_diarization: bool = False,
        progress_callback: Callable[[str, float], None] | None = None,
        transcription: Any = None,
    ) -> DubbingResult:
        """Dub a video into a target language.

        Args:
            enable_diarization: Enable speaker diarization to clone each speaker's
                voice separately. With ``transcription=None``, runs alongside Whisper.
                With a supplied ``transcription`` that has no speakers, runs pyannote
                standalone and overlays speakers onto the supplied words. Ignored when
                the supplied transcription already has speaker labels.
            transcription: Optional pre-computed ``Transcription`` to skip the Whisper
                step. Speaker labels on the supplied transcription drive per-speaker
                voice cloning. If it has no speakers, pass ``enable_diarization=True``
                to add them via pyannote (requires word-level timings).
        """
        if self._local_pipeline is None:
            self._init_local_pipeline()

        return self._local_pipeline.process(
            source_audio=video.audio,
            target_lang=target_lang,
            source_lang=source_lang,
            preserve_background=preserve_background,
            voice_clone=voice_clone,
            enable_diarization=enable_diarization,
            progress_callback=progress_callback,
            transcription=transcription,
        )

    def dub_and_replace(
        self,
        video: Video,
        target_lang: str,
        source_lang: str | None = None,
        preserve_background: bool = True,
        voice_clone: bool = True,
        enable_diarization: bool = False,
        progress_callback: Callable[[str, float], None] | None = None,
        transcription: Any = None,
    ) -> Video:
        """Dub a video and return a new video with the dubbed audio.

        Args:
            transcription: Optional pre-computed ``Transcription`` to skip the Whisper
                step. Speaker labels on the supplied transcription drive per-speaker
                voice cloning. See ``dub()`` for the interaction with
                ``enable_diarization``.
        """
        result = self.dub(
            video=video,
            target_lang=target_lang,
            source_lang=source_lang,
            preserve_background=preserve_background,
            voice_clone=voice_clone,
            enable_diarization=enable_diarization,
            progress_callback=progress_callback,
            transcription=transcription,
        )
        return video.add_audio(result.dubbed_audio, overlay=False)

    def dub_file(
        self,
        input_path: str | Path,
        output_path: str | Path,
        target_lang: str,
        source_lang: str | None = None,
        preserve_background: bool = True,
        voice_clone: bool = True,
        enable_diarization: bool = False,
        progress_callback: Callable[[str, float], None] | None = None,
        transcription: Any = None,
        keep_original_audio: bool = False,
    ) -> DubbingResult:
        """Dub a video file in place on disk without loading video frames into memory.

        Extracts the audio track via ffmpeg, runs the dubbing pipeline on the
        audio only, then muxes the dubbed audio back into the source video
        using ffmpeg stream-copy (no video re-encode). Peak memory is bounded
        by model weights and the audio track — independent of video length and
        resolution.

        Use this instead of ``dub_and_replace`` when the source video is long
        or high-resolution and you don't need frame-level access in Python.

        Args:
            input_path: Path to the source video file.
            output_path: Path to write the dubbed video. Overwritten if it exists.
            target_lang: Target language code (e.g. ``"es"``, ``"fr"``).
            source_lang: Source language code, or ``None`` to auto-detect.
            preserve_background: Preserve background music/effects via source separation.
            voice_clone: Clone the source speaker's voice for the dubbed track.
            enable_diarization: Enable speaker diarization for per-speaker voice cloning.
                See ``dub()`` for the interaction with ``transcription``.
            progress_callback: Optional callback ``(stage: str, progress: float) -> None``.
            transcription: Optional pre-computed ``Transcription`` to skip the Whisper
                step. Speaker labels on the supplied transcription drive per-speaker
                voice cloning. If it has no speakers, pass ``enable_diarization=True``
                to add them via pyannote (requires word-level timings).
            keep_original_audio: If True, retain the source audio in the output
                as a secondary track behind the dubbed one (editorial A/B).

        Returns:
            ``DubbingResult`` with the dubbed audio, translated segments, and
            source transcription. The output video is written to ``output_path``.
        """
        from videopython.ai.dubbing.remux import replace_audio_stream_from_audio
        from videopython.audio import Audio

        input_path = Path(input_path)
        output_path = Path(output_path)

        if not input_path.exists():
            raise FileNotFoundError(f"Input video not found: {input_path}")

        logger.info("dub_file: loading audio from %s", input_path)
        source_audio = Audio.from_path(input_path)

        if self._local_pipeline is None:
            self._init_local_pipeline()

        result = self._local_pipeline.process(
            source_audio=source_audio,
            target_lang=target_lang,
            source_lang=source_lang,
            preserve_background=preserve_background,
            voice_clone=voice_clone,
            enable_diarization=enable_diarization,
            progress_callback=progress_callback,
            transcription=transcription,
        )

        # Stream the dubbed Audio directly into ffmpeg via stdin instead of
        # going through a temp WAV on disk. For a 2h dub the temp file would
        # be ~10 GB written-then-read; the streaming path drops both copies.
        replace_audio_stream_from_audio(
            video_path=input_path,
            audio=result.dubbed_audio,
            output_path=output_path,
            keep_original_audio=keep_original_audio,
        )

        return result

    def revoice(
        self,
        video: Video,
        text: str,
        preserve_background: bool = True,
        progress_callback: Callable[[str, float], None] | None = None,
    ) -> RevoiceResult:
        """Replace speech in a video with new text using voice cloning."""
        if self._local_pipeline is None:
            self._init_local_pipeline()

        return self._local_pipeline.revoice(
            source_audio=video.audio,
            text=text,
            preserve_background=preserve_background,
            progress_callback=progress_callback,
        )

    def revoice_and_replace(
        self,
        video: Video,
        text: str,
        preserve_background: bool = True,
        progress_callback: Callable[[str, float], None] | None = None,
    ) -> Video:
        """Revoice a video and return a new video with the revoiced audio."""
        result = self.revoice(
            video=video,
            text=text,
            preserve_background=preserve_background,
            progress_callback=progress_callback,
        )

        speech_duration = result.speech_duration
        video_duration = video.total_seconds

        if video_duration > speech_duration:
            output_video = video[: round(speech_duration * video.fps)]
        else:
            output_video = video

        return output_video.add_audio(result.revoiced_audio, overlay=False)

    @staticmethod
    def get_supported_languages() -> dict[str, str]:
        from videopython.ai.dubbing.translation import OllamaTranslator

        return OllamaTranslator.get_supported_languages()

dub 

dub(
    video: Video,
    target_lang: str,
    source_lang: str | None = None,
    preserve_background: bool = True,
    voice_clone: bool = True,
    enable_diarization: bool = False,
    progress_callback: Callable[[str, float], None]
    | None = None,
    transcription: Any = None,
) -> DubbingResult

Dub a video into a target language.

Parameters:

Name Type Description Default
enable_diarization bool

Enable speaker diarization to clone each speaker's voice separately. With transcription=None, runs alongside Whisper. With a supplied transcription that has no speakers, runs pyannote standalone and overlays speakers onto the supplied words. Ignored when the supplied transcription already has speaker labels.

 False
transcription Any

Optional pre-computed Transcription to skip the Whisper step. Speaker labels on the supplied transcription drive per-speaker voice cloning. If it has no speakers, pass enable_diarization=True to add them via pyannote (requires word-level timings).

 None
Source code in src/videopython/ai/dubbing/dubber.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def dub(
    self,
    video: Video,
    target_lang: str,
    source_lang: str | None = None,
    preserve_background: bool = True,
    voice_clone: bool = True,
    enable_diarization: bool = False,
    progress_callback: Callable[[str, float], None] | None = None,
    transcription: Any = None,
) -> DubbingResult:
    """Dub a video into a target language.

    Args:
        enable_diarization: Enable speaker diarization to clone each speaker's
            voice separately. With ``transcription=None``, runs alongside Whisper.
            With a supplied ``transcription`` that has no speakers, runs pyannote
            standalone and overlays speakers onto the supplied words. Ignored when
            the supplied transcription already has speaker labels.
        transcription: Optional pre-computed ``Transcription`` to skip the Whisper
            step. Speaker labels on the supplied transcription drive per-speaker
            voice cloning. If it has no speakers, pass ``enable_diarization=True``
            to add them via pyannote (requires word-level timings).
    """
    if self._local_pipeline is None:
        self._init_local_pipeline()

    return self._local_pipeline.process(
        source_audio=video.audio,
        target_lang=target_lang,
        source_lang=source_lang,
        preserve_background=preserve_background,
        voice_clone=voice_clone,
        enable_diarization=enable_diarization,
        progress_callback=progress_callback,
        transcription=transcription,
    )

dub_and_replace 

dub_and_replace(
    video: Video,
    target_lang: str,
    source_lang: str | None = None,
    preserve_background: bool = True,
    voice_clone: bool = True,
    enable_diarization: bool = False,
    progress_callback: Callable[[str, float], None]
    | None = None,
    transcription: Any = None,
) -> Video

Dub a video and return a new video with the dubbed audio.

Parameters:

Name Type Description Default
transcription Any

Optional pre-computed Transcription to skip the Whisper step. Speaker labels on the supplied transcription drive per-speaker voice cloning. See dub() for the interaction with enable_diarization.

 None
Source code in src/videopython/ai/dubbing/dubber.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def dub_and_replace(
    self,
    video: Video,
    target_lang: str,
    source_lang: str | None = None,
    preserve_background: bool = True,
    voice_clone: bool = True,
    enable_diarization: bool = False,
    progress_callback: Callable[[str, float], None] | None = None,
    transcription: Any = None,
) -> Video:
    """Dub a video and return a new video with the dubbed audio.

    Args:
        transcription: Optional pre-computed ``Transcription`` to skip the Whisper
            step. Speaker labels on the supplied transcription drive per-speaker
            voice cloning. See ``dub()`` for the interaction with
            ``enable_diarization``.
    """
    result = self.dub(
        video=video,
        target_lang=target_lang,
        source_lang=source_lang,
        preserve_background=preserve_background,
        voice_clone=voice_clone,
        enable_diarization=enable_diarization,
        progress_callback=progress_callback,
        transcription=transcription,
    )
    return video.add_audio(result.dubbed_audio, overlay=False)

dub_file 

dub_file(
    input_path: str | Path,
    output_path: str | Path,
    target_lang: str,
    source_lang: str | None = None,
    preserve_background: bool = True,
    voice_clone: bool = True,
    enable_diarization: bool = False,
    progress_callback: Callable[[str, float], None]
    | None = None,
    transcription: Any = None,
    keep_original_audio: bool = False,
) -> DubbingResult

Dub a video file in place on disk without loading video frames into memory.

Extracts the audio track via ffmpeg, runs the dubbing pipeline on the audio only, then muxes the dubbed audio back into the source video using ffmpeg stream-copy (no video re-encode). Peak memory is bounded by model weights and the audio track — independent of video length and resolution.

Use this instead of dub_and_replace when the source video is long or high-resolution and you don't need frame-level access in Python.

Parameters:

Name Type Description Default
input_path str | Path

Path to the source video file.

 required
output_path str | Path

Path to write the dubbed video. Overwritten if it exists.

 required
target_lang str

Target language code (e.g. "es", "fr").

 required
source_lang str | None

Source language code, or None to auto-detect.

 None
preserve_background bool

Preserve background music/effects via source separation.

 True
voice_clone bool

Clone the source speaker's voice for the dubbed track.

 True
enable_diarization bool

Enable speaker diarization for per-speaker voice cloning. See dub() for the interaction with transcription.

 False
progress_callback Callable[[str, float], None] | None

Optional callback (stage: str, progress: float) -> None.

 None
transcription Any

Optional pre-computed Transcription to skip the Whisper step. Speaker labels on the supplied transcription drive per-speaker voice cloning. If it has no speakers, pass enable_diarization=True to add them via pyannote (requires word-level timings).

 None
keep_original_audio bool

If True, retain the source audio in the output as a secondary track behind the dubbed one (editorial A/B).

 False

Returns:

Type Description
DubbingResult

DubbingResult with the dubbed audio, translated segments, and
DubbingResult

source transcription. The output video is written to output_path.
Source code in src/videopython/ai/dubbing/dubber.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def dub_file(
    self,
    input_path: str | Path,
    output_path: str | Path,
    target_lang: str,
    source_lang: str | None = None,
    preserve_background: bool = True,
    voice_clone: bool = True,
    enable_diarization: bool = False,
    progress_callback: Callable[[str, float], None] | None = None,
    transcription: Any = None,
    keep_original_audio: bool = False,
) -> DubbingResult:
    """Dub a video file in place on disk without loading video frames into memory.

    Extracts the audio track via ffmpeg, runs the dubbing pipeline on the
    audio only, then muxes the dubbed audio back into the source video
    using ffmpeg stream-copy (no video re-encode). Peak memory is bounded
    by model weights and the audio track — independent of video length and
    resolution.

    Use this instead of ``dub_and_replace`` when the source video is long
    or high-resolution and you don't need frame-level access in Python.

    Args:
        input_path: Path to the source video file.
        output_path: Path to write the dubbed video. Overwritten if it exists.
        target_lang: Target language code (e.g. ``"es"``, ``"fr"``).
        source_lang: Source language code, or ``None`` to auto-detect.
        preserve_background: Preserve background music/effects via source separation.
        voice_clone: Clone the source speaker's voice for the dubbed track.
        enable_diarization: Enable speaker diarization for per-speaker voice cloning.
            See ``dub()`` for the interaction with ``transcription``.
        progress_callback: Optional callback ``(stage: str, progress: float) -> None``.
        transcription: Optional pre-computed ``Transcription`` to skip the Whisper
            step. Speaker labels on the supplied transcription drive per-speaker
            voice cloning. If it has no speakers, pass ``enable_diarization=True``
            to add them via pyannote (requires word-level timings).
        keep_original_audio: If True, retain the source audio in the output
            as a secondary track behind the dubbed one (editorial A/B).

    Returns:
        ``DubbingResult`` with the dubbed audio, translated segments, and
        source transcription. The output video is written to ``output_path``.
    """
    from videopython.ai.dubbing.remux import replace_audio_stream_from_audio
    from videopython.audio import Audio

    input_path = Path(input_path)
    output_path = Path(output_path)

    if not input_path.exists():
        raise FileNotFoundError(f"Input video not found: {input_path}")

    logger.info("dub_file: loading audio from %s", input_path)
    source_audio = Audio.from_path(input_path)

    if self._local_pipeline is None:
        self._init_local_pipeline()

    result = self._local_pipeline.process(
        source_audio=source_audio,
        target_lang=target_lang,
        source_lang=source_lang,
        preserve_background=preserve_background,
        voice_clone=voice_clone,
        enable_diarization=enable_diarization,
        progress_callback=progress_callback,
        transcription=transcription,
    )

    # Stream the dubbed Audio directly into ffmpeg via stdin instead of
    # going through a temp WAV on disk. For a 2h dub the temp file would
    # be ~10 GB written-then-read; the streaming path drops both copies.
    replace_audio_stream_from_audio(
        video_path=input_path,
        audio=result.dubbed_audio,
        output_path=output_path,
        keep_original_audio=keep_original_audio,
    )

    return result

revoice 

revoice(
    video: Video,
    text: str,
    preserve_background: bool = True,
    progress_callback: Callable[[str, float], None]
    | None = None,
) -> RevoiceResult

Replace speech in a video with new text using voice cloning.

Source code in src/videopython/ai/dubbing/dubber.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def revoice(
    self,
    video: Video,
    text: str,
    preserve_background: bool = True,
    progress_callback: Callable[[str, float], None] | None = None,
) -> RevoiceResult:
    """Replace speech in a video with new text using voice cloning."""
    if self._local_pipeline is None:
        self._init_local_pipeline()

    return self._local_pipeline.revoice(
        source_audio=video.audio,
        text=text,
        preserve_background=preserve_background,
        progress_callback=progress_callback,
    )

revoice_and_replace 

revoice_and_replace(
    video: Video,
    text: str,
    preserve_background: bool = True,
    progress_callback: Callable[[str, float], None]
    | None = None,
) -> Video

Revoice a video and return a new video with the revoiced audio.

Source code in src/videopython/ai/dubbing/dubber.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def revoice_and_replace(
    self,
    video: Video,
    text: str,
    preserve_background: bool = True,
    progress_callback: Callable[[str, float], None] | None = None,
) -> Video:
    """Revoice a video and return a new video with the revoiced audio."""
    result = self.revoice(
        video=video,
        text=text,
        preserve_background=preserve_background,
        progress_callback=progress_callback,
    )

    speech_duration = result.speech_duration
    video_duration = video.total_seconds

    if video_duration > speech_duration:
        output_video = video[: round(speech_duration * video.fps)]
    else:
        output_video = video

    return output_video.add_audio(result.revoiced_audio, overlay=False)

DubbingConfig

Settings shared by VideoDubber and LocalDubbingPipeline. Pass config=DubbingConfig(...) or the same knobs as flat kwargs — the constructor builds a DubbingConfig either way.

DubbingConfig

Bases: BaseModel

Knobs shared by :class:VideoDubber and :class:LocalDubbingPipeline.

Accepted as either config=DubbingConfig(...) or flat kwargs on the two constructors; the flat path builds a DubbingConfig internally.

Attributes:

Name Type Description
device str | None

Execution device (cpu, cuda, mps, or None for auto).
low_memory bool

When True, each pipeline stage (Whisper, Demucs, translation, Chatterbox TTS) is unloaded from memory after it runs, so only one model is resident at a time. Trades per-run latency (~10-30s of extra model loads) for a much lower memory ceiling. Recommended for GPUs with <=12GB VRAM or hosts with <32GB RAM. Default False.
whisper_model WhisperModel

Whisper model size used for transcription. Larger models give better accuracy at the cost of VRAM and latency. One of tiny, base, small, medium, large, turbo. Default turbo.
condition_on_previous_text bool

Forwarded to AudioToText. Defaults to False (Whisper's own default is True). With conditioning on, a single hallucinated filler phrase cascades through the rest of the file. See AudioToText for the full rationale.
no_speech_threshold float

Forwarded to AudioToText. Whisper's no-speech probability cutoff; lower it to make that gate easier to trigger.
logprob_threshold float | None

Forwarded to AudioToText. Whisper's average log-probability gate.
vocabulary list[str] | None

Forwarded to AudioToText. Optional list of brand names, product names, or proper nouns to bias Whisper's first-window decoder via initial_prompt. Recovers near-mishears (e.g. Klarna -> "carna") on brand-monitoring inputs without new model deps.
strict_quality bool

When True, the pipeline raises :class:GarbageTranscriptError before Demucs/translation/TTS run if the transcript-quality heuristic returns "reject". When False (default), low-quality transcripts are logged at WARNING but processing continues. Either way the :class:TranscriptQuality is exposed on DubbingResult for inspection.
translator_model str | None

Ollama tag for the translation model (None uses the translator's default). translator_host sets the server URL.
Source code in src/videopython/ai/dubbing/config.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class DubbingConfig(BaseModel):
    """Knobs shared by :class:`VideoDubber` and :class:`LocalDubbingPipeline`.

    Accepted as either ``config=DubbingConfig(...)`` or flat kwargs on the
    two constructors; the flat path builds a ``DubbingConfig`` internally.

    Attributes:
        device: Execution device (``cpu``, ``cuda``, ``mps``, or ``None`` for auto).
        low_memory: When True, each pipeline stage (Whisper, Demucs, translation,
            Chatterbox TTS) is unloaded from memory after it runs, so only one
            model is resident at a time. Trades per-run latency (~10-30s of
            extra model loads) for a much lower memory ceiling. Recommended
            for GPUs with <=12GB VRAM or hosts with <32GB RAM. Default False.
        whisper_model: Whisper model size used for transcription. Larger
            models give better accuracy at the cost of VRAM and latency. One
            of ``tiny``, ``base``, ``small``, ``medium``, ``large``, ``turbo``.
            Default ``turbo``.
        condition_on_previous_text: Forwarded to ``AudioToText``. Defaults to
            ``False`` (Whisper's own default is ``True``). With conditioning
            on, a single hallucinated filler phrase cascades through the rest
            of the file. See ``AudioToText`` for the full rationale.
        no_speech_threshold: Forwarded to ``AudioToText``. Whisper's
            no-speech probability cutoff; lower it to make that gate easier to trigger.
        logprob_threshold: Forwarded to ``AudioToText``. Whisper's average
            log-probability gate.
        vocabulary: Forwarded to ``AudioToText``. Optional list of brand
            names, product names, or proper nouns to bias Whisper's
            first-window decoder via ``initial_prompt``. Recovers
            near-mishears (e.g. Klarna -> "carna") on brand-monitoring
            inputs without new model deps.
        strict_quality: When True, the pipeline raises
            :class:`GarbageTranscriptError` before Demucs/translation/TTS
            run if the transcript-quality heuristic returns ``"reject"``.
            When False (default), low-quality transcripts are logged at
            WARNING but processing continues. Either way the
            :class:`TranscriptQuality` is exposed on ``DubbingResult`` for
            inspection.
        translator_model: Ollama tag for the translation model (``None`` uses the
            translator's default). ``translator_host`` sets the server URL.
    """

    model_config = ConfigDict(frozen=True)

    device: str | None = None
    low_memory: bool = False
    whisper_model: WhisperModel = "turbo"
    condition_on_previous_text: bool = False
    no_speech_threshold: float = 0.6
    logprob_threshold: float | None = -1.0
    vocabulary: list[str] | None = None
    strict_quality: bool = False
    translator_model: str | None = None
    translator_host: str | None = None

    @classmethod
    def from_args(cls, config: DubbingConfig | None = None, /, **kwargs: Any) -> DubbingConfig:
        """Resolve either a ``config`` object or flat knob kwargs into a config.

        The shared accept-one-or-the-other guard for ``VideoDubber`` and
        ``LocalDubbingPipeline``, which both take ``config=DubbingConfig(...)``
        or the flat kwargs (not both).
        """
        if config is not None and kwargs:
            raise TypeError("Pass either `config=` or knob kwargs, not both")
        return config or cls(**kwargs)

    def init_log_fields(self) -> dict[str, object]:
        """Subset of fields surfaced in the init-log line.

        Hand-picked so log noise stays bounded as the config grows.
        """
        return {
            "device": self.device.lower() if isinstance(self.device, str) else "auto",
            "low_memory": self.low_memory,
            "whisper_model": self.whisper_model,
            "translator_model": self.translator_model,
        }

from_args classmethod 

from_args(
    config: DubbingConfig | None = None, /, **kwargs: Any
) -> DubbingConfig

Resolve either a config object or flat knob kwargs into a config.

The shared accept-one-or-the-other guard for VideoDubber and LocalDubbingPipeline, which both take config=DubbingConfig(...) or the flat kwargs (not both).

Source code in src/videopython/ai/dubbing/config.py 
66
67
68
69
70
71
72
73
74
75
76
@classmethod
def from_args(cls, config: DubbingConfig | None = None, /, **kwargs: Any) -> DubbingConfig:
    """Resolve either a ``config`` object or flat knob kwargs into a config.

    The shared accept-one-or-the-other guard for ``VideoDubber`` and
    ``LocalDubbingPipeline``, which both take ``config=DubbingConfig(...)``
    or the flat kwargs (not both).
    """
    if config is not None and kwargs:
        raise TypeError("Pass either `config=` or knob kwargs, not both")
    return config or cls(**kwargs)

init_log_fields 

init_log_fields() -> dict[str, object]

Subset of fields surfaced in the init-log line.

Hand-picked so log noise stays bounded as the config grows.

Source code in src/videopython/ai/dubbing/config.py 
78
79
80
81
82
83
84
85
86
87
88
def init_log_fields(self) -> dict[str, object]:
    """Subset of fields surfaced in the init-log line.

    Hand-picked so log noise stays bounded as the config grows.
    """
    return {
        "device": self.device.lower() if isinstance(self.device, str) else "auto",
        "low_memory": self.low_memory,
        "whisper_model": self.whisper_model,
        "translator_model": self.translator_model,
    }

Results

result = dubber.dub(video, target_lang="es")

result.num_segments, result.source_lang, result.target_lang
result.translation_failures            # original indices with missing/invalid translation parts
result.synthesis_failures              # original indices without generated speech

for segment in result.translated_segments:
    print(f"{segment.original_text!r} -> {segment.translated_text!r}")

for speaker, sample in result.voice_samples.items():
    print(f"{speaker}: {sample.metadata.duration_seconds:.1f}s sample")

DubbingResult

Bases: BaseModel

Result of a video dubbing operation.

Attributes:

Name Type Description
dubbed_audio Audio

The final dubbed audio track.
translated_segments list[TranslatedSegment]

List of translated segments with timing.
source_transcription Transcription

Original transcription of the source audio.
source_lang str

Detected or specified source language.
target_lang str

Target language for dubbing.
separated_audio SeparatedAudio | None

Separated audio components (if preserve_background=True).
voice_samples dict[str, Audio]

Dictionary mapping speaker IDs to voice sample Audio.
timing_summary TimingSummary | None

Aggregate stats over per-segment timing adjustments.
transcript_quality TranscriptQuality | None

Heuristic quality assessment of the transcription (None when the pipeline returned early on an empty transcription).
translation_failures list[int]

Indices of segments the translator could not translate (missing after its parse-retry pass); those segments are dubbed with empty text.
synthesis_failures list[int]

Original segment indices whose speech failed to generate or had less than 100 ms after fragment joining.
Source code in src/videopython/ai/dubbing/models.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
class DubbingResult(BaseModel):
    """Result of a video dubbing operation.

    Attributes:
        dubbed_audio: The final dubbed audio track.
        translated_segments: List of translated segments with timing.
        source_transcription: Original transcription of the source audio.
        source_lang: Detected or specified source language.
        target_lang: Target language for dubbing.
        separated_audio: Separated audio components (if preserve_background=True).
        voice_samples: Dictionary mapping speaker IDs to voice sample Audio.
        timing_summary: Aggregate stats over per-segment timing adjustments.
        transcript_quality: Heuristic quality assessment of the transcription
            (None when the pipeline returned early on an empty transcription).
        translation_failures: Indices of segments the translator could not
            translate (missing after its parse-retry pass); those segments are
            dubbed with empty text.
        synthesis_failures: Original segment indices whose speech failed to
            generate or had less than 100 ms after fragment joining.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    dubbed_audio: Audio
    translated_segments: list[TranslatedSegment]
    source_transcription: Transcription
    source_lang: str
    target_lang: str
    separated_audio: SeparatedAudio | None = None
    voice_samples: dict[str, Audio] = Field(default_factory=dict)
    timing_summary: TimingSummary | None = None
    transcript_quality: TranscriptQuality | None = None
    translation_failures: list[int] = Field(default_factory=list)
    # Original segment indices, including every member of a failed joined turn.
    synthesis_failures: list[int] = Field(default_factory=list)

    @property
    def num_segments(self) -> int:
        """Number of translated segments."""
        return len(self.translated_segments)

    @property
    def total_duration(self) -> float:
        """Total duration of the dubbed audio."""
        return self.dubbed_audio.metadata.duration_seconds

    def get_segments_by_speaker(self) -> dict[str, list[TranslatedSegment]]:
        """Group translated segments by speaker.

        Returns:
            Dictionary mapping speaker IDs to their segments.
        """
        segments_by_speaker: dict[str, list[TranslatedSegment]] = {}
        for segment in self.translated_segments:
            speaker = segment.speaker or "unknown"
            if speaker not in segments_by_speaker:
                segments_by_speaker[speaker] = []
            segments_by_speaker[speaker].append(segment)
        return segments_by_speaker

num_segments property 

num_segments: int

Number of translated segments.

total_duration property 

total_duration: float

Total duration of the dubbed audio.

get_segments_by_speaker 

get_segments_by_speaker() -> dict[
    str, list[TranslatedSegment]
]

Group translated segments by speaker.

Returns:

Type Description
dict[str, list[TranslatedSegment]]

Dictionary mapping speaker IDs to their segments.
Source code in src/videopython/ai/dubbing/models.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
def get_segments_by_speaker(self) -> dict[str, list[TranslatedSegment]]:
    """Group translated segments by speaker.

    Returns:
        Dictionary mapping speaker IDs to their segments.
    """
    segments_by_speaker: dict[str, list[TranslatedSegment]] = {}
    for segment in self.translated_segments:
        speaker = segment.speaker or "unknown"
        if speaker not in segments_by_speaker:
            segments_by_speaker[speaker] = []
        segments_by_speaker[speaker].append(segment)
    return segments_by_speaker

RevoiceResult

Bases: BaseModel

Result of a voice replacement operation.

Attributes:

Name Type Description
revoiced_audio Audio

The final audio with new speech.
text str

The text that was spoken.
separated_audio SeparatedAudio | None

Separated audio components (if preserve_background=True).
voice_sample Audio | None

Voice sample used for cloning.
original_duration float

Duration of the original audio.
speech_duration float

Duration of the generated speech.
Source code in src/videopython/ai/dubbing/models.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
class RevoiceResult(BaseModel):
    """Result of a voice replacement operation.

    Attributes:
        revoiced_audio: The final audio with new speech.
        text: The text that was spoken.
        separated_audio: Separated audio components (if preserve_background=True).
        voice_sample: Voice sample used for cloning.
        original_duration: Duration of the original audio.
        speech_duration: Duration of the generated speech.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    revoiced_audio: Audio
    text: str
    separated_audio: SeparatedAudio | None = None
    voice_sample: Audio | None = None
    original_duration: float = 0.0
    speech_duration: float = 0.0

    @property
    def total_duration(self) -> float:
        """Total duration of the revoiced audio."""
        return self.revoiced_audio.metadata.duration_seconds

total_duration property 

total_duration: float

Total duration of the revoiced audio.

TranslatedSegment

Bases: BaseModel

A segment of translated text with timing information.

Attributes:

Name Type Description
original_segment TranscriptionSegment

The original transcription segment.
translated_text str

The translated text.
source_lang str

Source language code (e.g., "en").
target_lang str

Target language code (e.g., "es").
speaker str | None

Speaker identifier if available.
start float

Start time in seconds.
end float

End time in seconds.
source_segment_index int | None

Original transcript index when this is a dubbing phrase.
Source code in src/videopython/ai/dubbing/models.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class TranslatedSegment(BaseModel):
    """A segment of translated text with timing information.

    Attributes:
        original_segment: The original transcription segment.
        translated_text: The translated text.
        source_lang: Source language code (e.g., "en").
        target_lang: Target language code (e.g., "es").
        speaker: Speaker identifier if available.
        start: Start time in seconds.
        end: End time in seconds.
        source_segment_index: Original transcript index when this is a dubbing phrase.
    """

    original_segment: TranscriptionSegment
    translated_text: str
    source_lang: str
    target_lang: str
    speaker: str | None = None
    start: float = 0.0
    end: float = 0.0
    source_segment_index: int | None = None

    @model_validator(mode="after")
    def _default_timing_from_segment(self) -> TranslatedSegment:
        # Cached records use zero start and end to request the source segment timing.
        if self.start == 0.0 and self.end == 0.0:
            self.start = self.original_segment.start
            self.end = self.original_segment.end
        if self.speaker is None:
            self.speaker = self.original_segment.speaker
        return self

    @property
    def original_text(self) -> str:
        """Get the original text from the segment."""
        return self.original_segment.text

    @property
    def duration(self) -> float:
        """Duration of the segment in seconds."""
        return self.end - self.start

original_text property 

original_text: str

Get the original text from the segment.

duration property 

duration: float

Duration of the segment in seconds.

SeparatedAudio

Bases: BaseModel

Audio separated into different components.

Attributes:

Name Type Description
vocals Audio

Isolated vocal/speech track.
background Audio

Combined background audio (music + effects).
music Audio | None

Isolated music track (if available).
effects Audio | None

Isolated sound effects track (if available).
original Audio

The original unseparated audio.
Source code in src/videopython/ai/dubbing/models.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class SeparatedAudio(BaseModel):
    """Audio separated into different components.

    Attributes:
        vocals: Isolated vocal/speech track.
        background: Combined background audio (music + effects).
        music: Isolated music track (if available).
        effects: Isolated sound effects track (if available).
        original: The original unseparated audio.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    vocals: Audio
    background: Audio
    original: Audio
    music: Audio | None = None
    effects: Audio | None = None

    @property
    def has_detailed_separation(self) -> bool:
        """Check if music and effects are separated."""
        return self.music is not None and self.effects is not None

has_detailed_separation property 

has_detailed_separation: bool

Check if music and effects are separated.

Expressiveness

Per-segment Chatterbox generate() knobs (exaggeration, cfg_weight, temperature). None on a field means "let Chatterbox use its default". The pipeline derives these from source vocals RMS relative to the whole-vocals baseline, so the dub tracks the source's loud/quiet shape instead of using flat defaults everywhere.

RMS ratio vs baseline exaggeration cfg_weight
< 0.7× (calm) 0.3 0.7
0.7×–1.3× (normal) Chatterbox default Chatterbox default
> 1.3× (dramatic) 0.85 0.35

Expressiveness

Bases: BaseModel

Chatterbox generate() knobs derived from source-segment prosody.

None on any field means "let Chatterbox use its own default" -- avoids pinning the dub against future Chatterbox default changes.

Attributes:

Name Type Description
exaggeration float | None

Emotional intensity. Chatterbox default 0.5; 0.7+ produces dramatic output.
cfg_weight float | None

Classifier-free guidance weight. Chatterbox default 0.5; lower values (~0.3) slow pacing.
temperature float | None

Sampling temperature. Chatterbox default 0.8.
Source code in src/videopython/ai/dubbing/models.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class Expressiveness(BaseModel):
    """Chatterbox ``generate()`` knobs derived from source-segment prosody.

    ``None`` on any field means "let Chatterbox use its own default" --
    avoids pinning the dub against future Chatterbox default changes.

    Attributes:
        exaggeration: Emotional intensity. Chatterbox default ``0.5``;
            ``0.7+`` produces dramatic output.
        cfg_weight: Classifier-free guidance weight. Chatterbox default
            ``0.5``; lower values (~``0.3``) slow pacing.
        temperature: Sampling temperature. Chatterbox default ``0.8``.
    """

    model_config = ConfigDict(frozen=True)

    exaggeration: float | None = None
    cfg_weight: float | None = None
    temperature: float | None = None

    def as_kwargs(self) -> dict[str, float]:
        """Knobs as a dict, dropping ``None`` entries.

        Suitable for ``**``-expansion into Chatterbox.
        """
        return {
            name: value
            for name, value in (
                ("exaggeration", self.exaggeration),
                ("cfg_weight", self.cfg_weight),
                ("temperature", self.temperature),
            )
            if value is not None
        }

as_kwargs 

as_kwargs() -> dict[str, float]

Knobs as a dict, dropping None entries.

Suitable for **-expansion into Chatterbox.

Source code in src/videopython/ai/dubbing/models.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def as_kwargs(self) -> dict[str, float]:
    """Knobs as a dict, dropping ``None`` entries.

    Suitable for ``**``-expansion into Chatterbox.
    """
    return {
        name: value
        for name, value in (
            ("exaggeration", self.exaggeration),
            ("cfg_weight", self.cfg_weight),
            ("temperature", self.temperature),
        )
        if value is not None
    }

TimingSummary

Aggregate stats over the per-segment timing adjustments. excessive_speed_count counts turns exceeding the preferred maximum speed; max_speed_factor records the fastest adjustment. The pipeline borrows following silence before speeding up and preserves complete speech instead of clipping its tail. Small tempo-filter duration errors are corrected by resampling the entire output, which can slightly shift pitch. clean_count includes speed factors within 0.01 of 1.0; stretched_count includes the remaining adjustments. See Update timing consumers when migrating callers or saved results.

TimingSummary

Bases: BaseModel

Summarize speed changes and count adjustments above the preferred maximum.

Source code in src/videopython/ai/dubbing/models.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class TimingSummary(BaseModel):
    """Summarize speed changes and count adjustments above the preferred maximum."""

    total_segments: int
    clean_count: int
    stretched_count: int
    mean_speed_factor: float
    excessive_speed_count: int
    max_speed_factor: float

    @classmethod
    def from_adjustments(cls, adjustments: list[TimingAdjustment]) -> TimingSummary:
        """Aggregate a list of TimingAdjustments into a TimingSummary."""
        total = len(adjustments)
        if total == 0:
            return cls(
                total_segments=0,
                clean_count=0,
                stretched_count=0,
                mean_speed_factor=1.0,
                excessive_speed_count=0,
                max_speed_factor=1.0,
            )

        clean = sum(abs(adj.speed_factor - 1.0) <= CLEAN_SPEED_TOLERANCE for adj in adjustments)

        return cls(
            total_segments=total,
            clean_count=clean,
            stretched_count=total - clean,
            mean_speed_factor=sum(adj.speed_factor for adj in adjustments) / total,
            excessive_speed_count=sum(adj.excessive_speed for adj in adjustments),
            max_speed_factor=max(adj.speed_factor for adj in adjustments),
        )

from_adjustments classmethod 

from_adjustments(
    adjustments: list[TimingAdjustment],
) -> TimingSummary

Aggregate a list of TimingAdjustments into a TimingSummary.

Source code in src/videopython/ai/dubbing/models.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
@classmethod
def from_adjustments(cls, adjustments: list[TimingAdjustment]) -> TimingSummary:
    """Aggregate a list of TimingAdjustments into a TimingSummary."""
    total = len(adjustments)
    if total == 0:
        return cls(
            total_segments=0,
            clean_count=0,
            stretched_count=0,
            mean_speed_factor=1.0,
            excessive_speed_count=0,
            max_speed_factor=1.0,
        )

    clean = sum(abs(adj.speed_factor - 1.0) <= CLEAN_SPEED_TOLERANCE for adj in adjustments)

    return cls(
        total_segments=total,
        clean_count=clean,
        stretched_count=total - clean,
        mean_speed_factor=sum(adj.speed_factor for adj in adjustments) / total,
        excessive_speed_count=sum(adj.excessive_speed for adj in adjustments),
        max_speed_factor=max(adj.speed_factor for adj in adjustments),
    )

TranscriptQuality

Heuristic assessment over the Whisper transcription, surfaced on every DubbingResult and driving the optional strict_quality reject path. Flags: dominant phrase covering ≥70% of segment characters, median avg_logprob < -1.5, or speech under 5% of a clip longer than 30 s. recommendation is "reject" when dominance fires together with another flag, "warn" for any single flag, "ok" otherwise.

TranscriptQuality

Bases: BaseModel

Quality assessment of a Whisper transcription.

Attributes:

Name Type Description
recommendation Recommendation

"ok" (continue), "warn" (continue, log), or "reject" (caller should refuse to dub if strict_quality).
dominant_phrase str | None

The repeating phrase that triggered the dominance flag, or None when the flag didn't fire.
dominant_phrase_fraction float

Character-count share of the most common normalized segment phrase. 0.0 when no segments.
median_avg_logprob float | None

Median of avg_logprob across segments that carry it; None when no segment had a logprob (e.g. SRT-loaded).
speech_fraction float

Sum of segment durations divided by the audio's wall-clock duration.
flags list[str]

Human-readable list of which checks fired.
Source code in src/videopython/ai/dubbing/quality.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class TranscriptQuality(BaseModel):
    """Quality assessment of a Whisper transcription.

    Attributes:
        recommendation: ``"ok"`` (continue), ``"warn"`` (continue, log), or
            ``"reject"`` (caller should refuse to dub if strict_quality).
        dominant_phrase: The repeating phrase that triggered the dominance
            flag, or None when the flag didn't fire.
        dominant_phrase_fraction: Character-count share of the most common
            normalized segment phrase. 0.0 when no segments.
        median_avg_logprob: Median of ``avg_logprob`` across segments that
            carry it; None when no segment had a logprob (e.g. SRT-loaded).
        speech_fraction: Sum of segment durations divided by the audio's
            wall-clock duration.
        flags: Human-readable list of which checks fired.
    """

    recommendation: Recommendation
    dominant_phrase: str | None
    dominant_phrase_fraction: float
    median_avg_logprob: float | None
    speech_fraction: float
    flags: list[str] = Field(default_factory=list)

GarbageTranscriptError

Bases: AiError, RuntimeError

Raised by the dubbing pipeline when strict_quality=True and the transcript heuristic returns recommendation="reject".

The triggering :class:TranscriptQuality is attached as quality so callers can introspect the flags without re-running the pipeline.

Source code in src/videopython/ai/dubbing/quality.py 
79
80
81
82
83
84
85
86
87
88
89
class GarbageTranscriptError(AiError, RuntimeError):
    """Raised by the dubbing pipeline when ``strict_quality=True`` and the
    transcript heuristic returns ``recommendation="reject"``.

    The triggering :class:`TranscriptQuality` is attached as ``quality`` so
    callers can introspect the flags without re-running the pipeline.
    """

    def __init__(self, message: str, quality: TranscriptQuality):
        super().__init__(message)
        self.quality = quality

Supported languages

VideoDubber.get_supported_languages()
# {'en': 'English', 'es': 'Spanish', 'fr': 'French', ...}

The returned map names languages known to the translator. It is not a tested language matrix for the complete dubbing pipeline. Translation attempts other codes; actual translation and synthesis support depends on the selected models.

OllamaTranslator

Import from videopython.ai.dubbing.translation. The default model is qwen3.6:27b; VideoDubber(translator_model=..., translator_host=...) forwards a model tag and host. Vision is not required.

max_tokens defaults to 4096 and must be at least 140. n_ctx defaults to 8192 and must be at least max_tokens + 1020. The options keys num_predict and num_ctx override these values. Invalid effective budgets raise at construction. These are allocation estimates, not tokenizer guarantees.

keep_alive defaults to five minutes between requests. None uses the server policy. unload() requests release on the Ollama server. A failed release logs a warning and clears the local client without discarding completed translations; server memory can remain allocated. VideoDubber(low_memory=True) requests release after translation and before speech synthesis.

OllamaTranslator

Bases: ManagedPredictor

Dub translation via a local Ollama text model.

The model must support Ollama's structured-output format; ollama pull <model> first. Long text is split into bounded requests. n_ctx reserves room for the prompt, source text and max_tokens output budget. options can override these as num_ctx and num_predict; effective budgets are validated at construction.

Source code in src/videopython/ai/dubbing/translation.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
class OllamaTranslator(ManagedPredictor):
    """Dub translation via a local Ollama text model.

    The model must support Ollama's structured-output ``format``; ``ollama pull
    <model>`` first. Long text is split into bounded requests. ``n_ctx`` reserves
    room for the prompt, source text and ``max_tokens`` output budget. ``options``
    can override these as ``num_ctx`` and ``num_predict``; effective budgets are
    validated at construction.
    """

    def __init__(
        self,
        model: str = DEFAULT_TRANSLATION_MODEL,
        *,
        host: str | None = None,
        n_ctx: int = 8192,
        max_tokens: int = 4096,
        temperature: float = 0.1,
        options: dict[str, Any] | None = None,
        keep_alive: str | int | None = "5m",
    ) -> None:
        client_options = {"temperature": temperature, "num_ctx": n_ctx, "num_predict": max_tokens, **(options or {})}
        self.n_ctx = int(client_options["num_ctx"])
        self.max_tokens = int(client_options["num_predict"])
        # Reserve space for target-language expansion and the JSON envelope.
        self._part_chars = min(
            800,
            int((self.n_ctx - self.max_tokens - 1000) * _CHARS_PER_TOKEN),
            int((self.max_tokens - 100) * _CHARS_PER_TOKEN / 2),
        )
        if self._part_chars < 40:
            raise ValueError(
                f"Translation requires max_tokens (num_predict) >= 140 and n_ctx (num_ctx) "
                f">= max_tokens + 1020; got n_ctx={self.n_ctx}, max_tokens={self.max_tokens}"
            )
        # Keep the model resident between bounded requests; low-memory pipelines
        # explicitly unload it at the end of translation before loading TTS.
        self._client = OllamaStructuredClient(model=model, host=host, options=client_options, keep_alive=keep_alive)
        self._failures_last_call: list[int] = []

    def translate_segments(
        self,
        segments: list[TranscriptionSegment],
        target_lang: str,
        source_lang: str | None = None,
        progress_callback: Callable[[float], None] | None = None,
    ) -> list[TranslatedSegment]:
        """Translate bounded source parts independently, retrying invalid replies.

        A failed part leaves its entire parent empty in ``translation_failures``.
        """
        from videopython.ai.dubbing.models import TranslatedSegment

        effective_source = source_lang or "en"
        self._failures_last_call = []

        units: list[tuple[int, int, str]] = []
        for parent, segment in enumerate(segments):
            if _is_translatable_text(segment.text):
                for part, text in enumerate(split_text(segment.text, self._part_chars)):
                    units.append((parent, part, text))
        translated_parts: dict[int, list[str]] = {}
        failed: set[int] = set()
        for identity, (parent, part, text) in enumerate(units):
            before = units[identity - 1][2][-240:] if identity else ""
            after = units[identity + 1][2][:240] if identity + 1 < len(units) else ""
            entry: dict[str, Any] = {"i": identity, "parent": parent, "part": part, "text": text}
            source_chars = len(" ".join(segments[parent].text.split()))
            part_duration = max(0.0, segments[parent].end - segments[parent].start) * len(text) / source_chars
            entry["target_chars"] = max(1, round(part_duration * _SPEECH_CHARS_PER_SEC.get(target_lang, 12.0)))
            logprob = segments[parent].avg_logprob
            if logprob is not None and logprob < -1.0:
                entry["low_confidence"] = True
            prompt = (
                "Context only (do not translate): "
                + json.dumps({"before": before, "after": after}, ensure_ascii=False)
                + "\nTarget:\n"
                + json.dumps(entry, ensure_ascii=False)
            )
            translated = None
            schema = deepcopy(_TRANSLATION_SCHEMA)
            schema["properties"]["translations"]["items"]["properties"]["i"]["const"] = identity
            for _attempt in range(2):
                try:
                    data = self._client.generate_json(
                        system=_build_system_prompt(effective_source, target_lang),
                        text=prompt
                        + ("\nReturn exactly the requested identity and all target text." if _attempt else ""),
                        schema=schema,
                    )
                    parsed = _parse_translations(data)
                    if set(parsed) == {identity} and len(parsed[identity]) <= max(80, 6 * len(text)):
                        translated = parsed[identity]
                        break
                except OllamaError:
                    pass
            if translated is None:
                failed.add(parent)
            else:
                translated_parts.setdefault(parent, []).append(translated)
            if progress_callback is not None:
                progress_callback(0.95 * (identity + 1) / len(units))
        # Never publish an incomplete parent when just one of its parts failed.
        self._failures_last_call = sorted(failed)
        translation_for_orig = {
            parent: " ".join(parts) for parent, parts in translated_parts.items() if parent not in failed
        }
        translated_segments = [
            TranslatedSegment(
                original_segment=seg,
                translated_text=translation_for_orig.get(i, ""),
                source_lang=effective_source,
                target_lang=target_lang,
                speaker=seg.speaker,
                start=seg.start,
                end=seg.end,
            )
            for i, seg in enumerate(segments)
        ]
        if progress_callback is not None:
            progress_callback(1.0)
        return translated_segments

    @property
    def translation_failures(self) -> list[int]:
        """Indices (in the most recent ``segments`` input) where translation failed entirely."""
        return list(self._failures_last_call)

    def unload(self) -> None:
        self._client.unload()

    @staticmethod
    def get_supported_languages() -> dict[str, str]:
        return LANGUAGE_NAMES.copy()

translation_failures property 

translation_failures: list[int]

Indices (in the most recent segments input) where translation failed entirely.

translate_segments 

translate_segments(
    segments: list[TranscriptionSegment],
    target_lang: str,
    source_lang: str | None = None,
    progress_callback: Callable[[float], None]
    | None = None,
) -> list[TranslatedSegment]

Translate bounded source parts independently, retrying invalid replies.

A failed part leaves its entire parent empty in translation_failures.

Source code in src/videopython/ai/dubbing/translation.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def translate_segments(
    self,
    segments: list[TranscriptionSegment],
    target_lang: str,
    source_lang: str | None = None,
    progress_callback: Callable[[float], None] | None = None,
) -> list[TranslatedSegment]:
    """Translate bounded source parts independently, retrying invalid replies.

    A failed part leaves its entire parent empty in ``translation_failures``.
    """
    from videopython.ai.dubbing.models import TranslatedSegment

    effective_source = source_lang or "en"
    self._failures_last_call = []

    units: list[tuple[int, int, str]] = []
    for parent, segment in enumerate(segments):
        if _is_translatable_text(segment.text):
            for part, text in enumerate(split_text(segment.text, self._part_chars)):
                units.append((parent, part, text))
    translated_parts: dict[int, list[str]] = {}
    failed: set[int] = set()
    for identity, (parent, part, text) in enumerate(units):
        before = units[identity - 1][2][-240:] if identity else ""
        after = units[identity + 1][2][:240] if identity + 1 < len(units) else ""
        entry: dict[str, Any] = {"i": identity, "parent": parent, "part": part, "text": text}
        source_chars = len(" ".join(segments[parent].text.split()))
        part_duration = max(0.0, segments[parent].end - segments[parent].start) * len(text) / source_chars
        entry["target_chars"] = max(1, round(part_duration * _SPEECH_CHARS_PER_SEC.get(target_lang, 12.0)))
        logprob = segments[parent].avg_logprob
        if logprob is not None and logprob < -1.0:
            entry["low_confidence"] = True
        prompt = (
            "Context only (do not translate): "
            + json.dumps({"before": before, "after": after}, ensure_ascii=False)
            + "\nTarget:\n"
            + json.dumps(entry, ensure_ascii=False)
        )
        translated = None
        schema = deepcopy(_TRANSLATION_SCHEMA)
        schema["properties"]["translations"]["items"]["properties"]["i"]["const"] = identity
        for _attempt in range(2):
            try:
                data = self._client.generate_json(
                    system=_build_system_prompt(effective_source, target_lang),
                    text=prompt
                    + ("\nReturn exactly the requested identity and all target text." if _attempt else ""),
                    schema=schema,
                )
                parsed = _parse_translations(data)
                if set(parsed) == {identity} and len(parsed[identity]) <= max(80, 6 * len(text)):
                    translated = parsed[identity]
                    break
            except OllamaError:
                pass
        if translated is None:
            failed.add(parent)
        else:
            translated_parts.setdefault(parent, []).append(translated)
        if progress_callback is not None:
            progress_callback(0.95 * (identity + 1) / len(units))
    # Never publish an incomplete parent when just one of its parts failed.
    self._failures_last_call = sorted(failed)
    translation_for_orig = {
        parent: " ".join(parts) for parent, parts in translated_parts.items() if parent not in failed
    }
    translated_segments = [
        TranslatedSegment(
            original_segment=seg,
            translated_text=translation_for_orig.get(i, ""),
            source_lang=effective_source,
            target_lang=target_lang,
            speaker=seg.speaker,
            start=seg.start,
            end=seg.end,
        )
        for i, seg in enumerate(segments)
    ]
    if progress_callback is not None:
        progress_callback(1.0)
    return translated_segments

Phrase and synthesis limits

Tiny adjacent fragments can join within one speaker when the gap is at most 150 ms. A group contains at most four turns spanning at most ten seconds. The longer fragment supplies the expression profile. Isolated groups shorter than 100 ms become synthesis failures. Original transcript entries remain separate.

Local synthesis retries invalid token outputs up to three attempts. Duration checks can flag the backend's output limit but cannot verify spoken-word coverage. For the relationship between turns, phrases, and source indices, see The dubbing pipeline.