mkv2cast.converter

Core conversion logic for mkv2cast.

Contains: - Codec detection and decision logic - Backend selection (VAAPI, QSV, CPU) - FFmpeg command building - File conversion functions - Progress callback support for library usage - Batch processing with multi-threading

Functions

build_transcode_cmd(inp, decision, backend, ...)

Build ffmpeg transcoding command.

calculate_eta(current_time_ms, dur_ms, ...)

Calculate ETA in seconds based on progress.

check_disk_space(output_dir, tmp_dir, ...)

Return error message if disk guard would be violated, else None.

convert_batch(input_paths[, cfg, ...])

Convert multiple files in parallel using multi-threading.

convert_file(input_path[, cfg, backend, ...])

Convert a single MKV file.

decide_for(path[, cfg])

Analyze a file and decide what transcoding is needed.

enforce_output_quota(output_path, ...)

Return error message if output exceeds quota, else None.

ffprobe_json(path)

Run ffprobe and return JSON output.

file_size(path)

Get file size in bytes.

get_output_tag(decision)

Get the output filename tag based on decision.

have_encoder(name)

Check if ffmpeg has the specified encoder.

is_audio_description(title)

Check if audio track is an audio description track.

parse_bitdepth_from_pix(pix)

Parse bit depth from pixel format string.

parse_ffmpeg_progress(line, dur_ms)

Parse FFmpeg progress line and return progress metrics.

pick_backend([cfg])

Select the best available encoding backend.

probe_duration_ms(path[, debug])

Get video duration in milliseconds.

run_quiet(cmd[, timeout])

Run a command quietly, return True if successful.

select_audio_track(streams[, cfg])

Select the best audio track based on user preferences.

select_subtitle_track(streams, audio_lang[, cfg])

Select the best subtitle track based on user preferences.

test_amf()

Test if AMD AMF encoding works.

test_nvenc()

Test if NVIDIA NVENC encoding works.

test_qsv([vaapi_device])

Test if QSV encoding works.

test_vaapi([vaapi_device])

Test if VAAPI encoding works.

video_args_for(backend[, cfg])

Get ffmpeg video encoding arguments for the specified backend.

Classes

Decision(need_v, need_a, aidx, add_silence, ...)

Decision about what transcoding is needed for a file.
mkv2cast.converter.run_quiet(cmd: List[str], timeout: float = 10.0) → bool[source]

Run a command quietly, return True if successful.

mkv2cast.converter.ffprobe_json(path: Path) → Dict[str, Any][source]

Run ffprobe and return JSON output.

mkv2cast.converter.probe_duration_ms(path: Path, debug: bool = False) → int[source]

Get video duration in milliseconds.

mkv2cast.converter.file_size(path: Path) → int[source]

Get file size in bytes.

mkv2cast.converter.check_disk_space(output_dir: Path, tmp_dir: Path | None, estimated_bytes: int, cfg: Config) → str | None[source]

Return error message if disk guard would be violated, else None.

mkv2cast.converter.enforce_output_quota(output_path: Path, input_size: int, cfg: Config) → str | None[source]

Return error message if output exceeds quota, else None.

mkv2cast.converter.have_encoder(name: str) → bool[source]

Check if ffmpeg has the specified encoder.

mkv2cast.converter.test_qsv(vaapi_device: str = '/dev/dri/renderD128') → bool[source]

Test if QSV encoding works.

mkv2cast.converter.test_vaapi(vaapi_device: str = '/dev/dri/renderD128') → bool[source]

Test if VAAPI encoding works.

mkv2cast.converter.test_nvenc() → bool[source]

Test if NVIDIA NVENC encoding works.

mkv2cast.converter.test_amf() → bool[source]

Test if AMD AMF encoding works.

mkv2cast.converter.pick_backend(cfg: Config | None = None) → str[source]

Select the best available encoding backend.

Parameters:

cfg – Config instance (uses global CFG if not provided).

Returns:

“nvenc”, “qsv”, “vaapi”, or “cpu”.

Return type:

Backend name

mkv2cast.converter.video_args_for(backend: str, cfg: Config | None = None) → List[str][source]

Get ffmpeg video encoding arguments for the specified backend.

class mkv2cast.converter.Decision(need_v: bool, need_a: bool, aidx: int, add_silence: bool, reason_v: str, vcodec: str, vpix: str, vbit: int, vhdr: bool, vprof: str, vlevel: int, acodec: str, ach: int, alang: str, format_name: str, sidx: int = -1, slang: str = '', sforced: bool = False)[source]

Bases: object

Decision about what transcoding is needed for a file.

need_v: bool
need_a: bool
aidx: int
add_silence: bool
reason_v: str
vcodec: str
vpix: str
vbit: int
vhdr: bool
vprof: str
vlevel: int
acodec: str
ach: int
alang: str
format_name: str
sidx: int = -1
slang: str = ''
sforced: bool = False
mkv2cast.converter.parse_bitdepth_from_pix(pix: str) → int[source]

Parse bit depth from pixel format string.

mkv2cast.converter.is_audio_description(title: str) → bool[source]

Check if audio track is an audio description track.

mkv2cast.converter.select_audio_track(streams: List[dict], cfg: Config | None = None) → Tuple[dict | None, str][source]

Select the best audio track based on user preferences.

Priority: 1. Explicit track index (–audio-track) 2. Language priority list (–audio-lang) 3. Default French preference (fre, fra, fr) 4. First audio track

Parameters:

  • streams – List of stream dictionaries from ffprobe.

  • cfg – Config instance.

Returns:

Tuple of (selected_stream, selected_language).

mkv2cast.converter.select_subtitle_track(streams: List[dict], audio_lang: str, cfg: Config | None = None) → Tuple[dict, bool] | None[source]

Select the best subtitle track based on user preferences.

Priority: 1. Explicit track index (–subtitle-track) 2. Forced subtitles in audio language (if –prefer-forced-subs) 3. Language priority list (–subtitle-lang) 4. No subtitles selected

Parameters:

  • streams – List of stream dictionaries from ffprobe.

  • audio_lang – The language of the selected audio track.

  • cfg – Config instance.

Returns:

Tuple of (selected_stream, is_forced) or None if no subtitle selected.

mkv2cast.converter.decide_for(path: Path, cfg: Config | None = None) → Decision[source]

Analyze a file and decide what transcoding is needed.

Parameters:

  • path – Path to the MKV file.

  • cfg – Config instance (uses global CFG if not provided).

Returns:

Decision dataclass with transcoding requirements.

mkv2cast.converter.build_transcode_cmd(inp: Path, decision: Decision, backend: str, tmp_out: Path, log_path: Path | None = None, cfg: Config | None = None) → Tuple[List[str], str][source]

Build ffmpeg transcoding command.

Parameters:

  • inp – Input file path.

  • decision – Decision dataclass with transcoding requirements.

  • backend – Encoding backend to use.

  • tmp_out – Temporary output path.

  • log_path – Optional path to write command log.

  • cfg – Config instance.

Returns:

Tuple of (command_args, stage_name).

mkv2cast.converter.parse_ffmpeg_progress(line: str, dur_ms: int) → Dict[str, Any][source]

Parse FFmpeg progress line and return progress metrics.

Parameters:

  • line – A line from FFmpeg stderr output.

  • dur_ms – Total duration in milliseconds.

Returns:

  • progress_percent: float (0-100)

  • fps: float

  • speed: str (e.g., “2.5x”)

  • bitrate: str (e.g., “2500kbits/s”)

  • current_time_ms: int

  • frame: int

  • size_bytes: int

Return type:

Dict with progress metrics

mkv2cast.converter.calculate_eta(current_time_ms: int, dur_ms: int, speed_str: str, start_time: float) → float[source]

Calculate ETA in seconds based on progress.

Parameters:

  • current_time_ms – Current position in milliseconds.

  • dur_ms – Total duration in milliseconds.

  • speed_str – Speed string like “2.5x”.

  • start_time – Start time (time.time()).

Returns:

Estimated time remaining in seconds.

mkv2cast.converter.get_output_tag(decision: Decision) → str[source]

Get the output filename tag based on decision.

mkv2cast.converter.convert_file(input_path: Path, cfg: Config | None = None, backend: str | None = None, output_dir: Path | None = None, log_path: Path | None = None, progress_callback: Callable[[Path, Dict[str, Any]], None] | None = None) → Tuple[bool, Path | None, str][source]

Convert a single MKV file.

Parameters:

  • input_path – Path to input MKV file.

  • cfg – Config instance (uses global CFG if not provided).

  • backend – Backend to use (auto-detected if not provided).

  • output_dir – Output directory (same as input if not provided).

  • log_path – Path for conversion log.

  • progress_callback – Optional callback function called with progress updates. The callback receives (filepath, progress_dict) where progress_dict contains: - stage: “checking” | “encoding” | “done” | “skipped” | “failed” - progress_percent: float (0-100) - fps: float - eta_seconds: float - bitrate: str - speed: str - current_time_ms: int - duration_ms: int - error: Optional[str]

Returns:

Tuple of (success, output_path, message).

Example

>>> def on_progress(filepath, progress):
...     print(f"{filepath.name}: {progress['stage']} - {progress['progress_percent']:.1f}%")
>>> success, output, msg = convert_file(Path("movie.mkv"), progress_callback=on_progress)
mkv2cast.converter.convert_batch(input_paths: List[Path], cfg: Config | None = None, progress_callback: Callable[[Path, Dict[str, Any]], None] | None = None, output_dir: Path | None = None, backend: str | None = None) → Dict[Path, Tuple[bool, Path | None, str]][source]

Convert multiple files in parallel using multi-threading.

This function processes multiple files concurrently, respecting the configured number of workers. Each file’s progress is reported via the optional callback.

Parameters:

  • input_paths – List of input file paths to convert.

  • cfg – Config instance (uses global CFG if not provided). The number of parallel workers is determined by cfg.encode_workers.

  • progress_callback – Optional callback function called with progress updates. The callback receives (filepath, progress_dict) for each file. The callback should be thread-safe if processing multiple files.

  • output_dir – Output directory for all files (same as input if not provided).

  • backend – Backend to use (auto-detected if not provided).

Returns:

Dict mapping input_path -> (success, output_path, message).

Example

>>> from mkv2cast import convert_batch, Config
>>> from pathlib import Path
>>>
>>> config = Config.for_library(hw="vaapi", encode_workers=2)
>>>
>>> def on_progress(filepath, progress):
...     print(f"{filepath.name}: {progress['progress_percent']:.1f}%")
>>>
>>> files = [Path("movie1.mkv"), Path("movie2.mkv")]
>>> results = convert_batch(files, cfg=config, progress_callback=on_progress)
>>>
>>> for filepath, (success, output, msg) in results.items():
...     print(f"{filepath.name}: {'OK' if success else 'FAIL'} - {msg}")