Converters Module

Video encoding converters supporting hardware and software modes.

Overview

The converters module provides H.264 to H.265 conversion using:

• Hardware Converter: Apple VideoToolbox (fast, efficient)
• Software Converter: libx265 (high quality, flexible)

Base Converter

base

Base converter interface for video encoding.

This module defines the abstract base class for all video converters, establishing a consistent interface for hardware and software encoders.

SDS Reference: SDS-V01-001 SRS Reference: SRS-201 (H.265 Encoding)

Example

from video_converter.converters.base import BaseConverter from video_converter.core import ConversionRequest, ConversionMode

class MyConverter(BaseConverter): ... def build_command(self, request: ConversionRequest) -> list[str]: ... return ["ffmpeg", "-i", str(request.input_path), ...] ... converter = MyConverter() request = ConversionRequest( ... input_path=Path("input.mov"), ... output_path=Path("output.mp4"), ... ) result = await converter.convert(request)

ConversionError

Bases: Exception

Exception raised when video conversion fails.

EncoderNotAvailableError

Bases: ConversionError

Exception raised when the required encoder is not available.

BaseConverter

BaseConverter(mode)

Bases: ABC

Abstract base class for video converters.

This class defines the interface that all video converters must implement. It provides common functionality for command execution and progress tracking.

Attributes:

Name	Type	Description
mode		The conversion mode (hardware or software).

Initialize the converter.

Parameters:

Name	Type	Description	Default
mode	ConversionMode	The conversion mode to use.	required

Source code in src/video_converter/converters/base.py

def __init__(self, mode: ConversionMode) -> None:
    """Initialize the converter.

    Args:
        mode: The conversion mode to use.
    """
    self.mode = mode
    self._command_runner = CommandRunner()
    self._ffprobe_runner = FFprobeRunner(self._command_runner)
    self._cancelled = False
    self._current_process: asyncio.subprocess.Process | None = None

encoder_name abstractmethod property

encoder_name

Get the encoder name for FFmpeg.

Returns:

Type	Description
str	The encoder name (e.g., "hevc_videotoolbox", "libx265").

is_available abstractmethod

is_available()

Check if this encoder is available on the system.

Returns:

Type	Description
bool	True if the encoder is available, False otherwise.

Source code in src/video_converter/converters/base.py

@abstractmethod
def is_available(self) -> bool:
    """Check if this encoder is available on the system.

    Returns:
        True if the encoder is available, False otherwise.
    """
    ...

build_command abstractmethod

build_command(request)

Build the FFmpeg command for conversion.

Parameters:

Name	Type	Description	Default
request	ConversionRequest	The conversion request with input/output paths and settings.	required

Returns:

Type	Description
list[str]	List of command arguments for FFmpeg.

Source code in src/video_converter/converters/base.py

@abstractmethod
def build_command(self, request: ConversionRequest) -> list[str]:
    """Build the FFmpeg command for conversion.

    Args:
        request: The conversion request with input/output paths and settings.

    Returns:
        List of command arguments for FFmpeg.
    """
    ...

convert async

convert(request, on_progress=None, on_progress_info=None)

Convert a video file.

Parameters:

Name	Type	Description	Default
request	ConversionRequest	The conversion request.	required
on_progress	Callable[[float], None] | None	Optional callback for simple progress updates (0.0-1.0).	None
on_progress_info	Callable[[ProgressInfo], None] | None	Optional callback for detailed progress updates with ProgressInfo containing ETA and other metrics.	None

Returns:

Type	Description
ConversionResult	ConversionResult with success status and statistics.

Raises:

Type	Description
ConversionError	If conversion fails.
EncoderNotAvailableError	If the encoder is not available.

Note

If both callbacks are provided, both will be called. on_progress_info provides more detailed information including eta_seconds, eta_formatted, and current encoding speed.

Source code in src/video_converter/converters/base.py

async def convert(
    self,
    request: ConversionRequest,
    on_progress: Callable[[float], None] | None = None,
    on_progress_info: Callable[[ProgressInfo], None] | None = None,
) -> ConversionResult:
    """Convert a video file.

    Args:
        request: The conversion request.
        on_progress: Optional callback for simple progress updates (0.0-1.0).
        on_progress_info: Optional callback for detailed progress updates
                         with ProgressInfo containing ETA and other metrics.

    Returns:
        ConversionResult with success status and statistics.

    Raises:
        ConversionError: If conversion fails.
        EncoderNotAvailableError: If the encoder is not available.

    Note:
        If both callbacks are provided, both will be called.
        on_progress_info provides more detailed information including
        eta_seconds, eta_formatted, and current encoding speed.
    """
    self._cancelled = False
    started_at = datetime.now()
    start_time = time.perf_counter()

    # Validate encoder availability
    if not self.is_available():
        return ConversionResult(
            success=False,
            request=request,
            error_message=f"Encoder '{self.encoder_name}' is not available",
            started_at=started_at,
            completed_at=datetime.now(),
        )

    # Get original file size
    try:
        original_size = request.input_path.stat().st_size
    except OSError as e:
        return ConversionResult(
            success=False,
            request=request,
            error_message=f"Cannot read input file: {e}",
            started_at=started_at,
            completed_at=datetime.now(),
        )

    # Get video duration for progress calculation
    video_duration = self._get_video_duration(request.input_path)
    progress_parser = ProgressParser(total_duration=video_duration)

    # Ensure output directory exists
    request.output_path.parent.mkdir(parents=True, exist_ok=True)

    # Build and execute command
    command = self.build_command(request)
    logger.info(f"Starting conversion: {request.input_path.name}")
    logger.debug(f"Command: {' '.join(command)}")

    stderr_output = []
    last_speed = 0.0

    try:
        # Create subprocess with stderr streaming
        self._current_process = await asyncio.create_subprocess_exec(
            *command,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        # Read stderr line by line for progress updates
        while True:
            if self._current_process.stderr is None:
                break

            line = await self._current_process.stderr.readline()
            if not line:
                break

            line_str = line.decode("utf-8", errors="replace")
            stderr_output.append(line_str)

            # Parse progress from FFmpeg output
            if progress_info := progress_parser.parse_line(line_str):
                last_speed = progress_info.speed
                # Call detailed progress callback with ProgressInfo
                if on_progress_info:
                    try:
                        on_progress_info(progress_info)
                    except Exception:
                        pass  # Swallow callback errors
                # Call simple progress callback with ratio (0.0-1.0)
                if on_progress and video_duration > 0:
                    on_progress(progress_info.percentage / 100.0)

        # Wait for process to complete
        await self._current_process.wait()

        if self._cancelled:
            # Clean up partial output
            if request.output_path.exists():
                request.output_path.unlink()
            return ConversionResult(
                success=False,
                request=request,
                original_size=original_size,
                error_message="Conversion cancelled",
                started_at=started_at,
                completed_at=datetime.now(),
            )

        if self._current_process.returncode != 0:
            error_msg = "".join(stderr_output).strip()
            # Clean up partial output
            if request.output_path.exists():
                request.output_path.unlink()
            return ConversionResult(
                success=False,
                request=request,
                original_size=original_size,
                error_message=f"FFmpeg failed: {error_msg[-500:]}",
                started_at=started_at,
                completed_at=datetime.now(),
            )

        # Get converted file size
        if not request.output_path.exists():
            return ConversionResult(
                success=False,
                request=request,
                original_size=original_size,
                error_message="Output file was not created",
                started_at=started_at,
                completed_at=datetime.now(),
            )

        converted_size = request.output_path.stat().st_size
        duration = time.perf_counter() - start_time

        # Calculate speed ratio from FFmpeg's reported speed or actual time
        speed_ratio = last_speed
        if speed_ratio == 0.0 and video_duration > 0 and duration > 0:
            speed_ratio = video_duration / duration

        # Call final progress callback
        if on_progress:
            on_progress(1.0)

        logger.info(
            f"Conversion complete: {request.input_path.name} "
            f"({original_size / 1024 / 1024:.1f}MB -> "
            f"{converted_size / 1024 / 1024:.1f}MB, "
            f"{(1 - converted_size / original_size) * 100:.1f}% reduction, "
            f"{speed_ratio:.1f}x speed)"
        )

        return ConversionResult(
            success=True,
            request=request,
            original_size=original_size,
            converted_size=converted_size,
            duration_seconds=duration,
            speed_ratio=speed_ratio,
            started_at=started_at,
            completed_at=datetime.now(),
        )

    except CommandNotFoundError as e:
        return ConversionResult(
            success=False,
            request=request,
            original_size=original_size,
            error_message=f"FFmpeg not found: {e}",
            started_at=started_at,
            completed_at=datetime.now(),
        )
    except Exception as e:
        logger.exception(f"Conversion error: {e}")
        # Clean up partial output
        if request.output_path.exists():
            request.output_path.unlink()
        return ConversionResult(
            success=False,
            request=request,
            original_size=original_size,
            error_message=str(e),
            started_at=started_at,
            completed_at=datetime.now(),
        )
    finally:
        self._current_process = None

cancel

cancel()

Cancel the current conversion.

Source code in src/video_converter/converters/base.py

def cancel(self) -> None:
    """Cancel the current conversion."""
    self._cancelled = True
    if self._current_process:
        try:
            self._current_process.terminate()
        except ProcessLookupError:
            pass  # Process already terminated

convert_sync

convert_sync(request, on_progress=None, on_progress_info=None)

Synchronous wrapper for convert.

Parameters:

Name	Type	Description	Default
request	ConversionRequest	The conversion request.	required
on_progress	Callable[[float], None] | None	Optional callback for simple progress updates (0.0-1.0).	None
on_progress_info	Callable[[ProgressInfo], None] | None	Optional callback for detailed progress updates with ProgressInfo containing ETA and other metrics.	None

Returns:

Type	Description
ConversionResult	ConversionResult with success status and statistics.

Source code in src/video_converter/converters/base.py

def convert_sync(
    self,
    request: ConversionRequest,
    on_progress: Callable[[float], None] | None = None,
    on_progress_info: Callable[[ProgressInfo], None] | None = None,
) -> ConversionResult:
    """Synchronous wrapper for convert.

    Args:
        request: The conversion request.
        on_progress: Optional callback for simple progress updates (0.0-1.0).
        on_progress_info: Optional callback for detailed progress updates
                         with ProgressInfo containing ETA and other metrics.

    Returns:
        ConversionResult with success status and statistics.
    """
    return asyncio.run(self.convert(request, on_progress, on_progress_info))

Hardware Converter

hardware

Hardware video converter using VideoToolbox.

This module implements hardware-accelerated H.265 encoding using macOS VideoToolbox through FFmpeg's hevc_videotoolbox encoder.

SDS Reference: SDS-V01-002 SRS Reference: SRS-203 (Hardware Encoding)

Example

from video_converter.converters.hardware import HardwareConverter from video_converter.core import ConversionRequest

converter = HardwareConverter() if converter.is_available(): ... request = ConversionRequest( ... input_path=Path("input.mov"), ... output_path=Path("output.mp4"), ... ) ... result = await converter.convert(request) ... print(f"Success: {result.success}")

HardwareConverter

HardwareConverter()

Bases: BaseConverter

Hardware-accelerated H.265 converter using VideoToolbox.

This converter uses Apple's VideoToolbox framework through FFmpeg for fast, hardware-accelerated H.265 encoding on macOS.

Attributes:

Name	Type	Description
mode		Always HARDWARE for this converter.

Initialize the hardware converter.

Source code in src/video_converter/converters/hardware.py

def __init__(self) -> None:
    """Initialize the hardware converter."""
    super().__init__(ConversionMode.HARDWARE)
    self._encoder_available: bool | None = None

encoder_name property

encoder_name

Get the encoder name.

Returns:

Type	Description
str	The VideoToolbox HEVC encoder name.

is_available

is_available()

Check if VideoToolbox encoder is available.

Returns:

Type	Description
bool	True if hevc_videotoolbox is available, False otherwise.

Source code in src/video_converter/converters/hardware.py

def is_available(self) -> bool:
    """Check if VideoToolbox encoder is available.

    Returns:
        True if hevc_videotoolbox is available, False otherwise.
    """
    if self._encoder_available is not None:
        return self._encoder_available

    try:
        result = subprocess.run(
            ["ffmpeg", "-hide_banner", "-encoders"],
            capture_output=True,
            text=True,
            timeout=10,
        )
        self._encoder_available = "hevc_videotoolbox" in result.stdout
        if self._encoder_available:
            logger.debug("VideoToolbox HEVC encoder is available")
        else:
            logger.warning("VideoToolbox HEVC encoder is not available")
        return self._encoder_available
    except (subprocess.SubprocessError, FileNotFoundError):
        self._encoder_available = False
        return False

build_command

build_command(request)

Build FFmpeg command for VideoToolbox encoding.

Parameters:

Name	Type	Description	Default
request	ConversionRequest	The conversion request with settings.	required

Returns:

Type	Description
list[str]	FFmpeg command arguments.

Source code in src/video_converter/converters/hardware.py

def build_command(self, request: ConversionRequest) -> list[str]:
    """Build FFmpeg command for VideoToolbox encoding.

    Args:
        request: The conversion request with settings.

    Returns:
        FFmpeg command arguments.
    """
    # Clamp quality to valid range
    quality = max(
        self.MIN_QUALITY,
        min(self.MAX_QUALITY, request.quality),
    )

    command = [
        "ffmpeg",
        "-hide_banner",
        "-y",  # Overwrite output
        "-i",
        str(request.input_path),
        # Video encoding
        "-c:v",
        self.encoder_name,
        "-q:v",
        str(quality),
        "-tag:v",
        "hvc1",  # Compatibility tag for Apple devices
        # Audio handling
        "-c:a",
        request.audio_mode,
        # Metadata handling
        "-map_metadata",
        "0",  # Copy all metadata
        "-movflags",
        "+faststart+use_metadata_tags",  # Enable streaming and preserve metadata tags
        # Output
        str(request.output_path),
    ]

    return command

create_hardware_converter

create_hardware_converter()

Factory function to create a hardware converter if available.

Returns:

Type	Description
HardwareConverter | None	HardwareConverter instance if available, None otherwise.

Source code in src/video_converter/converters/hardware.py

def create_hardware_converter() -> HardwareConverter | None:
    """Factory function to create a hardware converter if available.

    Returns:
        HardwareConverter instance if available, None otherwise.
    """
    converter = HardwareConverter()
    if converter.is_available():
        return converter
    return None

Software Converter

software

Software video converter using libx265.

This module implements software-based H.265 encoding using the libx265 encoder through FFmpeg.

SDS Reference: SDS-V01-003 SRS Reference: SRS-202 (Software Encoding)

Example

from video_converter.converters.software import SoftwareConverter from video_converter.core import ConversionRequest

converter = SoftwareConverter() if converter.is_available(): ... request = ConversionRequest( ... input_path=Path("input.mov"), ... output_path=Path("output.mp4"), ... preset="medium", ... crf=22, ... ) ... result = await converter.convert(request) ... print(f"Success: {result.success}")

SoftwareConverter

SoftwareConverter()

Bases: BaseConverter

Software-based H.265 converter using libx265.

This converter uses the libx265 encoder for CPU-based H.265 encoding. It's slower than hardware encoding but more widely compatible and offers more fine-grained control over encoding parameters.

Supports both 8-bit and 10-bit encoding for HDR content.

Attributes:

Name	Type	Description
mode		Always SOFTWARE for this converter.

Initialize the software converter.

Source code in src/video_converter/converters/software.py

def __init__(self) -> None:
    """Initialize the software converter."""
    super().__init__(ConversionMode.SOFTWARE)
    self._encoder_available: bool | None = None

encoder_name property

encoder_name

Get the encoder name.

Returns:

Type	Description
str	The libx265 encoder name.

is_available

is_available()

Check if libx265 encoder is available.

Returns:

Type	Description
bool	True if libx265 is available, False otherwise.

Source code in src/video_converter/converters/software.py

def is_available(self) -> bool:
    """Check if libx265 encoder is available.

    Returns:
        True if libx265 is available, False otherwise.
    """
    if self._encoder_available is not None:
        return self._encoder_available

    try:
        result = subprocess.run(
            ["ffmpeg", "-hide_banner", "-encoders"],
            capture_output=True,
            text=True,
            timeout=10,
        )
        self._encoder_available = "libx265" in result.stdout
        if self._encoder_available:
            logger.debug("libx265 encoder is available")
        else:
            logger.warning("libx265 encoder is not available")
        return self._encoder_available
    except (subprocess.SubprocessError, FileNotFoundError):
        self._encoder_available = False
        return False

build_command

build_command(request)

Build FFmpeg command for libx265 encoding.

Parameters:

Name	Type	Description	Default
request	ConversionRequest	The conversion request with settings.	required

Returns:

Type	Description
list[str]	FFmpeg command arguments.

Source code in src/video_converter/converters/software.py

def build_command(self, request: ConversionRequest) -> list[str]:
    """Build FFmpeg command for libx265 encoding.

    Args:
        request: The conversion request with settings.

    Returns:
        FFmpeg command arguments.
    """
    # Clamp CRF to valid range
    crf = max(self.MIN_CRF, min(self.MAX_CRF, request.crf))

    # Validate preset
    preset = request.preset if request.preset in self.VALID_PRESETS else self.DEFAULT_PRESET

    # Validate bit depth
    bit_depth = (
        request.bit_depth
        if request.bit_depth in self.VALID_BIT_DEPTHS
        else self.DEFAULT_BIT_DEPTH
    )

    # Select encoder based on bit depth
    # libx265 handles both 8-bit and 10-bit natively
    encoder = self.encoder_name

    command = [
        "ffmpeg",
        "-hide_banner",
        "-y",  # Overwrite output
        "-i",
        str(request.input_path),
        # Video encoding
        "-c:v",
        encoder,
        "-crf",
        str(crf),
        "-preset",
        preset,
        "-tag:v",
        "hvc1",  # Compatibility tag for Apple devices
    ]

    # Add 10-bit encoding options
    if bit_depth == 10:
        command.extend(["-pix_fmt", "yuv420p10le"])

        # Add HDR parameters if enabled
        if request.hdr:
            command.extend(["-x265-params", self.HDR_X265_PARAMS])

    # Audio handling
    command.extend(["-c:a", request.audio_mode])

    # Metadata handling
    command.extend(
        [
            "-map_metadata",
            "0",  # Copy all metadata
            "-movflags",
            "+faststart+use_metadata_tags",  # Enable streaming and preserve metadata tags
        ]
    )

    # Output
    command.append(str(request.output_path))

    return command

create_software_converter

create_software_converter()

Factory function to create a software converter if available.

Returns:

Type	Description
SoftwareConverter | None	SoftwareConverter instance if available, None otherwise.

Source code in src/video_converter/converters/software.py

def create_software_converter() -> SoftwareConverter | None:
    """Factory function to create a software converter if available.

    Returns:
        SoftwareConverter instance if available, None otherwise.
    """
    converter = SoftwareConverter()
    if converter.is_available():
        return converter
    return None

Factory

factory

Converter factory for creating appropriate video converters.

This module provides a factory pattern implementation for selecting and instantiating the appropriate video converter based on system capabilities and user preferences.

SDS Reference: SDS-V01-004 SRS Reference: SRS-204 (Encoder Selection)

Example

from video_converter.converters.factory import ConverterFactory from video_converter.core import ConversionMode

factory = ConverterFactory()

Get best available converter

converter = factory.get_converter()

Get specific converter type

hw_converter = factory.get_converter(ConversionMode.HARDWARE)

Check what's available

available = factory.get_available_converters() print(f"Available: {[c.encoder_name for c in available]}")

ConverterFactory

ConverterFactory()

Factory for creating video converters.

This factory manages converter instances and provides methods for selecting the most appropriate converter based on availability and user preferences.

Initialize the converter factory.

Source code in src/video_converter/converters/factory.py

def __init__(self) -> None:
    """Initialize the converter factory."""
    self._hardware_converter: HardwareConverter | None = None
    self._software_converter: SoftwareConverter | None = None
    self._initialized = False

get_converter

get_converter(mode=None, fallback=True)

Get a converter for the specified mode.

Parameters:

Name	Type	Description	Default
mode	ConversionMode | None	Preferred conversion mode. If None, selects best available.	None
fallback	bool	Whether to fall back to alternative if preferred unavailable.	True

Returns:

Type	Description
BaseConverter	A converter instance.

Raises:

Type	Description
EncoderNotAvailableError	If no suitable encoder is available.

Source code in src/video_converter/converters/factory.py

def get_converter(
    self,
    mode: ConversionMode | None = None,
    fallback: bool = True,
) -> BaseConverter:
    """Get a converter for the specified mode.

    Args:
        mode: Preferred conversion mode. If None, selects best available.
        fallback: Whether to fall back to alternative if preferred unavailable.

    Returns:
        A converter instance.

    Raises:
        EncoderNotAvailableError: If no suitable encoder is available.
    """
    self._ensure_initialized()

    # Auto-select mode if not specified
    if mode is None:
        return self._get_best_available()

    if mode == ConversionMode.HARDWARE:
        if self._hardware_converter and self._hardware_converter.is_available():
            return self._hardware_converter
        if fallback:
            logger.warning("Hardware encoder not available, falling back to software")
            if self._software_converter and self._software_converter.is_available():
                return self._software_converter
        raise EncoderNotAvailableError("Hardware encoder (VideoToolbox) is not available")

    if mode == ConversionMode.SOFTWARE:
        if self._software_converter and self._software_converter.is_available():
            return self._software_converter
        if fallback:
            logger.warning("Software encoder not available, falling back to hardware")
            if self._hardware_converter and self._hardware_converter.is_available():
                return self._hardware_converter
        raise EncoderNotAvailableError("Software encoder (libx265) is not available")

    raise ValueError(f"Unknown conversion mode: {mode}")

get_available_converters

get_available_converters()

Get list of all available converters.

Returns:

Type	Description
list[BaseConverter]	List of available converter instances.

Source code in src/video_converter/converters/factory.py

def get_available_converters(self) -> list[BaseConverter]:
    """Get list of all available converters.

    Returns:
        List of available converter instances.
    """
    self._ensure_initialized()

    available: list[BaseConverter] = []

    if self._hardware_converter and self._hardware_converter.is_available():
        available.append(self._hardware_converter)

    if self._software_converter and self._software_converter.is_available():
        available.append(self._software_converter)

    return available

is_hardware_available

is_hardware_available()

Check if hardware encoding is available.

Returns:

Type	Description
bool	True if VideoToolbox encoder is available.

Source code in src/video_converter/converters/factory.py

def is_hardware_available(self) -> bool:
    """Check if hardware encoding is available.

    Returns:
        True if VideoToolbox encoder is available.
    """
    self._ensure_initialized()
    return bool(self._hardware_converter and self._hardware_converter.is_available())

is_software_available

is_software_available()

Check if software encoding is available.

Returns:

Type	Description
bool	True if libx265 encoder is available.

Source code in src/video_converter/converters/factory.py

def is_software_available(self) -> bool:
    """Check if software encoding is available.

    Returns:
        True if libx265 encoder is available.
    """
    self._ensure_initialized()
    return bool(self._software_converter and self._software_converter.is_available())

get_converter

get_converter(mode=None, fallback=True)

Get a converter using the default factory.

Parameters:

Name	Type	Description	Default
mode	ConversionMode | None	Preferred conversion mode. If None, selects best available.	None
fallback	bool	Whether to fall back to alternative if preferred unavailable.	True

Returns:

Type	Description
BaseConverter	A converter instance.

Source code in src/video_converter/converters/factory.py

def get_converter(
    mode: ConversionMode | None = None,
    fallback: bool = True,
) -> BaseConverter:
    """Get a converter using the default factory.

    Args:
        mode: Preferred conversion mode. If None, selects best available.
        fallback: Whether to fall back to alternative if preferred unavailable.

    Returns:
        A converter instance.
    """
    global _default_factory
    if _default_factory is None:
        _default_factory = ConverterFactory()
    return _default_factory.get_converter(mode, fallback)

Progress

progress

Real-time conversion progress monitoring module.

This module provides real-time progress monitoring for FFmpeg video conversions, including accurate progress tracking, ETA calculation, and callback-based updates.

SDS Reference: SDS-V01-005 SRS Reference: SRS-205 (Real-time Progress Monitoring)

Example

from video_converter.converters.progress import ProgressMonitor, ProgressInfo

def on_progress(info: ProgressInfo) -> None: ... print(f"Progress: {info.percentage:.1f}% | ETA: {info.eta_formatted}")

monitor = ProgressMonitor(total_duration=180.5, callback=on_progress) monitor.on_output("frame= 720 fps=180 time=00:00:24.00 speed=6.0x")

ProgressInfo dataclass

ProgressInfo(frame=0, fps=0.0, current_time=0.0, total_time=0.0, current_size=0, bitrate=0.0, speed=0.0, quality=0.0)

Progress information for video conversion.

This dataclass contains all progress metrics extracted from FFmpeg output, including calculated properties for percentage and ETA.

Attributes:

Name	Type	Description
frame	int	Current frame number being processed.
fps	float	Encoding frames per second.
current_time	float	Current position in seconds.
total_time	float	Total video duration in seconds.
current_size	int	Current output file size in bytes.
bitrate	float	Current encoding bitrate in kbps.
speed	float	Encoding speed multiplier (e.g., 6.0 means 6x realtime).
quality	float	FFmpeg quality parameter (q value).

percentage property

percentage

Calculate completion percentage.

Returns:

Type	Description
float	Completion percentage from 0.0 to 100.0.

eta_seconds property

eta_seconds

Calculate estimated time remaining in seconds.

Returns:

Type	Description
float	Estimated seconds remaining, or infinity if speed is zero.

eta_formatted property

eta_formatted

Get human-readable ETA string.

Returns:

Type	Description
str	Formatted ETA string like "1h 30m", "5m 30s", or "calculating...".

size_formatted property

size_formatted

Get human-readable current size string.

Returns:

Type	Description
str	Formatted size string like "1.5 GB", "256 MB", or "0 B".

ProgressParser

ProgressParser(total_duration=0.0)

Parser for FFmpeg stderr progress output.

This parser extracts real-time progress information from FFmpeg's stderr, converting it into ProgressInfo objects for further processing.

Attributes:

Name	Type	Description
total_duration		Total video duration in seconds.

Initialize the progress parser.

Parameters:

Name	Type	Description	Default
total_duration	float	Total video duration in seconds. Used for percentage and ETA calculations.	0.0

Source code in src/video_converter/converters/progress.py

def __init__(self, total_duration: float = 0.0) -> None:
    """Initialize the progress parser.

    Args:
        total_duration: Total video duration in seconds.
                       Used for percentage and ETA calculations.
    """
    self.total_duration = max(0.0, total_duration)
    self._last_info: ProgressInfo | None = None

last_info property

last_info

Get the most recently parsed progress info.

Returns:

Type	Description
ProgressInfo | None	The last ProgressInfo parsed, or None if nothing parsed yet.

parse_line

parse_line(line)

Parse a single line of FFmpeg stderr output.

Parameters:

Name	Type	Description	Default
line	str	A line from FFmpeg's stderr output.	required

Returns:

Type	Description
ProgressInfo | None	ProgressInfo object if progress data was found, None otherwise.

Source code in src/video_converter/converters/progress.py

def parse_line(self, line: str) -> ProgressInfo | None:
    """Parse a single line of FFmpeg stderr output.

    Args:
        line: A line from FFmpeg's stderr output.

    Returns:
        ProgressInfo object if progress data was found, None otherwise.
    """
    # Quick check for progress-related content
    if "frame=" not in line or "time=" not in line:
        return None

    info = ProgressInfo(total_time=self.total_duration)

    # Parse frame count
    if match := self._FRAME_PATTERN.search(line):
        info.frame = int(match.group(1))

    # Parse FPS
    if match := self._FPS_PATTERN.search(line):
        info.fps = float(match.group(1))

    # Parse quality
    if match := self._QUALITY_PATTERN.search(line):
        info.quality = float(match.group(1))

    # Parse current output size (convert from kB to bytes)
    if match := self._SIZE_PATTERN.search(line):
        info.current_size = int(match.group(1)) * 1024

    # Parse current time position
    if match := self._TIME_PATTERN.search(line):
        hours = int(match.group(1))
        minutes = int(match.group(2))
        seconds = int(match.group(3))
        centiseconds = int(match.group(4))
        info.current_time = hours * 3600 + minutes * 60 + seconds + centiseconds / 100

    # Parse bitrate
    if match := self._BITRATE_PATTERN.search(line):
        info.bitrate = float(match.group(1))

    # Parse encoding speed
    if match := self._SPEED_PATTERN.search(line):
        info.speed = float(match.group(1))

    self._last_info = info
    return info

ProgressMonitor dataclass

ProgressMonitor(total_duration, callback, min_interval=0.1)

Monitor and report FFmpeg conversion progress.

This class combines parsing and callback functionality to provide a complete progress monitoring solution. It handles debouncing, minimum update intervals, and error handling.

Attributes:

Name	Type	Description
total_duration	float	Total video duration in seconds.
callback	Callable[[ProgressInfo], None]	Function to call with progress updates.
min_interval	float	Minimum seconds between callback invocations.

__post_init__

__post_init__()

Initialize the parser after dataclass initialization.

Source code in src/video_converter/converters/progress.py

def __post_init__(self) -> None:
    """Initialize the parser after dataclass initialization."""
    self._parser = ProgressParser(self.total_duration)

on_output

on_output(line)

Process a line of FFmpeg output.

This method parses the line and invokes the callback if new progress information is available and enough time has passed since the last callback.

Parameters:

Name	Type	Description	Default
line	str	A line from FFmpeg's stderr output.	required

Source code in src/video_converter/converters/progress.py

def on_output(self, line: str) -> None:
    """Process a line of FFmpeg output.

    This method parses the line and invokes the callback if new
    progress information is available and enough time has passed
    since the last callback.

    Args:
        line: A line from FFmpeg's stderr output.
    """
    import time

    info = self._parser.parse_line(line)
    if info is None:
        return

    current_time = time.monotonic()
    if current_time - self._last_callback_time >= self.min_interval:
        self._last_callback_time = current_time
        try:
            self.callback(info)
        except Exception as e:
            # Log callback exceptions to aid debugging while preventing conversion interruption
            logger.debug("Progress callback failed: %s", e)

get_current_progress

get_current_progress()

Get the current progress without triggering a callback.

Returns:

Type	Description
ProgressInfo | None	The most recent ProgressInfo, or None if not available.

Source code in src/video_converter/converters/progress.py

def get_current_progress(self) -> ProgressInfo | None:
    """Get the current progress without triggering a callback.

    Returns:
        The most recent ProgressInfo, or None if not available.
    """
    return self._parser.last_info

force_callback

force_callback()

Force a callback with current progress, ignoring interval.

Useful for final progress update at 100% completion.

Source code in src/video_converter/converters/progress.py

def force_callback(self) -> None:
    """Force a callback with current progress, ignoring interval.

    Useful for final progress update at 100% completion.
    """
    if info := self._parser.last_info:
        try:
            self.callback(info)
        except Exception as e:
            logger.debug("Force callback failed: %s", e)

create_simple_callback

create_simple_callback(show_bar=True, bar_width=30)

Create a simple console progress callback.

Parameters:

Name	Type	Description	Default
show_bar	bool	Whether to show a visual progress bar.	True
bar_width	int	Width of the progress bar in characters.	30

Returns:

Type	Description
Callable[[ProgressInfo], None]	A callback function suitable for ProgressMonitor.

Example

callback = create_simple_callback() monitor = ProgressMonitor(duration, callback)

Source code in src/video_converter/converters/progress.py

def create_simple_callback(
    show_bar: bool = True, bar_width: int = 30
) -> Callable[[ProgressInfo], None]:
    """Create a simple console progress callback.

    Args:
        show_bar: Whether to show a visual progress bar.
        bar_width: Width of the progress bar in characters.

    Returns:
        A callback function suitable for ProgressMonitor.

    Example:
        >>> callback = create_simple_callback()
        >>> monitor = ProgressMonitor(duration, callback)
    """

    def callback(info: ProgressInfo) -> None:
        if show_bar:
            filled = int(info.percentage / 100 * bar_width)
            bar = "\u2588" * filled + "\u2591" * (bar_width - filled)
            print(
                f"\r[{bar}] {info.percentage:.1f}% | "
                f"{info.size_formatted} | "
                f"ETA: {info.eta_formatted} | "
                f"{info.speed:.1f}x",
                end="",
                flush=True,
            )
        else:
            print(
                f"\rProgress: {info.percentage:.1f}% | ETA: {info.eta_formatted}",
                end="",
                flush=True,
            )

    return callback