Utils Module¶

Utility functions and helper classes.

Command Runner¶

command_runner ¶

Command execution utilities for external tools.

This module provides a wrapper for executing external commands like FFprobe, FFmpeg, and ExifTool with proper error handling and output capture.

SDS Reference: SDS-U01-001

CommandResult `dataclass` ¶

CommandResult(returncode, stdout, stderr)

Result of a command execution.

Attributes:

Name	Type	Description
`returncode`	`int`	Exit code of the command (0 = success).
`stdout`	`str`	Standard output from the command.
`stderr`	`str`	Standard error output from the command.
`success`	`bool`	Whether the command completed successfully.

success `property` ¶

success

Check if command completed successfully.

CommandNotFoundError ¶

CommandNotFoundError(command)

Bases: Exception

Raised when required external command is not found.

This exception provides helpful installation instructions for common tools.

Attributes:

Name	Type	Description
`command`		The command that was not found.

Source code in src/video_converter/utils/command_runner.py

def __init__(self, command: str) -> None:
    self.command = command
    hint = self.INSTALL_HINTS.get(command, "")
    msg = f"Command '{command}' not found."
    if hint:
        msg += f" {hint}"
    super().__init__(msg)

CommandTimeoutError ¶

CommandTimeoutError(command, timeout)

Bases: Exception

Raised when command execution exceeds timeout.

Attributes:

Name	Type	Description
`command`		The command that timed out.
`timeout`		The timeout value in seconds.

Source code in src/video_converter/utils/command_runner.py

def __init__(self, command: str, timeout: float) -> None:
    self.command = command
    self.timeout = timeout
    super().__init__(f"Command '{command}' timed out after {timeout:.1f} seconds.")

CommandExecutionError ¶

CommandExecutionError(command, returncode, stderr)

Bases: Exception

Raised when command execution fails.

Source code in src/video_converter/utils/command_runner.py

def __init__(self, command: str, returncode: int, stderr: str) -> None:
    self.command = command
    self.returncode = returncode
    self.stderr = stderr
    super().__init__(f"Command '{command}' failed with code {returncode}: {stderr}")

CommandRunner ¶

Wrapper for executing external commands.

This class provides both synchronous and asynchronous methods for running external commands with proper error handling.

Example

runner = CommandRunner() result = runner.run(["ffprobe", "-version"]) print(result.stdout)

check_command_exists `staticmethod` ¶

check_command_exists(command)

Check if a command exists in PATH.

Parameters:

Name	Type	Description	Default
`command`	`str`	The command name to check.	required

Returns:

Type	Description
`bool`	True if the command exists, False otherwise.

Source code in src/video_converter/utils/command_runner.py

@staticmethod
def check_command_exists(command: str) -> bool:
    """Check if a command exists in PATH.

    Args:
        command: The command name to check.

    Returns:
        True if the command exists, False otherwise.
    """
    return shutil.which(command) is not None

ensure_command_exists `staticmethod` ¶

ensure_command_exists(command)

Ensure a command exists, raising an error if not.

Parameters:

Name	Type	Description	Default
`command`	`str`	The command name to check.	required

Raises:

Type	Description
`CommandNotFoundError`	If the command is not found.

Source code in src/video_converter/utils/command_runner.py

@staticmethod
def ensure_command_exists(command: str) -> None:
    """Ensure a command exists, raising an error if not.

    Args:
        command: The command name to check.

    Raises:
        CommandNotFoundError: If the command is not found.
    """
    if not CommandRunner.check_command_exists(command):
        raise CommandNotFoundError(command)

run ¶

run(args, *, timeout=60.0, check=False, capture_output=True)

Run a command synchronously.

Parameters:

Name	Type	Description	Default
`args`	`list[str]`	Command and arguments to execute.	required
`timeout`	`float \| None`	Maximum time to wait for command (seconds).	`60.0`
`check`	`bool`	If True, raise exception on non-zero exit code.	`False`
`capture_output`	`bool`	If True, capture stdout and stderr.	`True`

Returns:

Type	Description
`CommandResult`	CommandResult containing the execution result.

Raises:

Type	Description
`CommandNotFoundError`	If the command is not found.
`CommandExecutionError`	If check=True and command fails.
`TimeoutExpired`	If command times out.

Source code in src/video_converter/utils/command_runner.py

def run(
    self,
    args: list[str],
    *,
    timeout: float | None = 60.0,
    check: bool = False,
    capture_output: bool = True,
) -> CommandResult:
    """Run a command synchronously.

    Args:
        args: Command and arguments to execute.
        timeout: Maximum time to wait for command (seconds).
        check: If True, raise exception on non-zero exit code.
        capture_output: If True, capture stdout and stderr.

    Returns:
        CommandResult containing the execution result.

    Raises:
        CommandNotFoundError: If the command is not found.
        CommandExecutionError: If check=True and command fails.
        subprocess.TimeoutExpired: If command times out.
    """
    command_name = args[0] if args else ""
    self.ensure_command_exists(command_name)

    try:
        result = subprocess.run(
            args,
            timeout=timeout,
            capture_output=capture_output,
            text=True,
        )

        cmd_result = CommandResult(
            returncode=result.returncode,
            stdout=result.stdout if capture_output else "",
            stderr=result.stderr if capture_output else "",
        )

        if check and not cmd_result.success:
            raise CommandExecutionError(command_name, cmd_result.returncode, cmd_result.stderr)

        return cmd_result

    except FileNotFoundError as e:
        raise CommandNotFoundError(command_name) from e

run_async `async` ¶

run_async(args, *, timeout=60.0, check=False)

Run a command asynchronously.

Parameters:

Name	Type	Description	Default
`args`	`list[str]`	Command and arguments to execute.	required
`timeout`	`float \| None`	Maximum time to wait for command (seconds).	`60.0`
`check`	`bool`	If True, raise exception on non-zero exit code.	`False`

Returns:

Type	Description
`CommandResult`	CommandResult containing the execution result.

Raises:

Type	Description
`CommandNotFoundError`	If the command is not found.
`CommandExecutionError`	If check=True and command fails.
`TimeoutError`	If command times out.

Source code in src/video_converter/utils/command_runner.py

async def run_async(
    self,
    args: list[str],
    *,
    timeout: float | None = 60.0,
    check: bool = False,
) -> CommandResult:
    """Run a command asynchronously.

    Args:
        args: Command and arguments to execute.
        timeout: Maximum time to wait for command (seconds).
        check: If True, raise exception on non-zero exit code.

    Returns:
        CommandResult containing the execution result.

    Raises:
        CommandNotFoundError: If the command is not found.
        CommandExecutionError: If check=True and command fails.
        asyncio.TimeoutError: If command times out.
    """
    command_name = args[0] if args else ""
    self.ensure_command_exists(command_name)

    try:
        process = await asyncio.create_subprocess_exec(
            *args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        stdout, stderr = await asyncio.wait_for(
            process.communicate(),
            timeout=timeout,
        )

        cmd_result = CommandResult(
            returncode=process.returncode or 0,
            stdout=stdout.decode("utf-8", errors="replace"),
            stderr=stderr.decode("utf-8", errors="replace"),
        )

        if check and not cmd_result.success:
            raise CommandExecutionError(command_name, cmd_result.returncode, cmd_result.stderr)

        return cmd_result

    except FileNotFoundError as e:
        raise CommandNotFoundError(command_name) from e

run_with_callback ¶

run_with_callback(args, on_output, *, timeout=600.0, check=False, read_stderr=True)

Run a command with real-time output streaming.

This method allows monitoring command output in real-time, which is useful for tracking progress of long-running operations like video encoding.

Parameters:

Name	Type	Description	Default
`args`	`list[str]`	Command and arguments to execute.	required
`on_output`	`Callable[[str], None]`	Callback function called for each output line.	required
`timeout`	`float \| None`	Maximum time to wait for command (seconds).	`600.0`
`check`	`bool`	If True, raise exception on non-zero exit code.	`False`
`read_stderr`	`bool`	If True, read from stderr (FFmpeg outputs to stderr).	`True`

Returns:

Type	Description
`CommandResult`	CommandResult containing the final execution result.

Raises:

Type	Description
`CommandNotFoundError`	If the command is not found.
`CommandExecutionError`	If check=True and command fails.
`CommandTimeoutError`	If command times out.

Example

def on_progress(line: str): ... if "frame=" in line: ... print(f"Progress: {line}") runner = CommandRunner() result = runner.run_with_callback( ... ["ffmpeg", "-i", "input.mp4", "output.mp4"], ... on_output=on_progress, ... read_stderr=True, ... )

Source code in src/video_converter/utils/command_runner.py

def run_with_callback(
    self,
    args: list[str],
    on_output: Callable[[str], None],
    *,
    timeout: float | None = 600.0,
    check: bool = False,
    read_stderr: bool = True,
) -> CommandResult:
    """Run a command with real-time output streaming.

    This method allows monitoring command output in real-time, which is
    useful for tracking progress of long-running operations like video
    encoding.

    Args:
        args: Command and arguments to execute.
        on_output: Callback function called for each output line.
        timeout: Maximum time to wait for command (seconds).
        check: If True, raise exception on non-zero exit code.
        read_stderr: If True, read from stderr (FFmpeg outputs to stderr).

    Returns:
        CommandResult containing the final execution result.

    Raises:
        CommandNotFoundError: If the command is not found.
        CommandExecutionError: If check=True and command fails.
        CommandTimeoutError: If command times out.

    Example:
        >>> def on_progress(line: str):
        ...     if "frame=" in line:
        ...         print(f"Progress: {line}")
        >>> runner = CommandRunner()
        >>> result = runner.run_with_callback(
        ...     ["ffmpeg", "-i", "input.mp4", "output.mp4"],
        ...     on_output=on_progress,
        ...     read_stderr=True,
        ... )
    """
    command_name = args[0] if args else ""
    self.ensure_command_exists(command_name)

    stdout_lines: list[str] = []
    stderr_lines: list[str] = []

    try:
        process = subprocess.Popen(
            args,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE if read_stderr else subprocess.DEVNULL,
            text=True,
            bufsize=1,
        )

        import selectors
        import time

        sel = selectors.DefaultSelector()
        if process.stdout:
            sel.register(process.stdout, selectors.EVENT_READ)
        if read_stderr and process.stderr:
            sel.register(process.stderr, selectors.EVENT_READ)

        start_time = time.monotonic()

        while True:
            # Check timeout
            if timeout is not None:
                elapsed = time.monotonic() - start_time
                if elapsed >= timeout:
                    process.kill()
                    process.wait()
                    raise CommandTimeoutError(command_name, timeout)

                remaining = timeout - elapsed
            else:
                remaining = None

            events = sel.select(timeout=min(remaining, 0.1) if remaining else 0.1)

            for key, _ in events:
                line = key.fileobj.readline()  # type: ignore[union-attr]
                if line:
                    line = line.rstrip("\n\r")
                    if key.fileobj == process.stdout:
                        stdout_lines.append(line)
                    else:
                        stderr_lines.append(line)
                    on_output(line)

            # Check if process has finished
            if process.poll() is not None:
                # Read remaining output
                if process.stdout:
                    for line in process.stdout:
                        line = line.rstrip("\n\r")
                        stdout_lines.append(line)
                        on_output(line)
                if read_stderr and process.stderr:
                    for line in process.stderr:
                        line = line.rstrip("\n\r")
                        stderr_lines.append(line)
                        on_output(line)
                break

        sel.close()

        cmd_result = CommandResult(
            returncode=process.returncode or 0,
            stdout="\n".join(stdout_lines),
            stderr="\n".join(stderr_lines),
        )

        if check and not cmd_result.success:
            raise CommandExecutionError(command_name, cmd_result.returncode, cmd_result.stderr)

        return cmd_result

    except FileNotFoundError as e:
        raise CommandNotFoundError(command_name) from e

FFprobeRunner ¶

FFprobeRunner(command_runner=None)

Specialized runner for FFprobe commands.

Provides convenient methods for common FFprobe operations with JSON output parsing.

Example

runner = FFprobeRunner()
info = runner.get_format_info(Path("video.mp4"))
print(info["format"]["duration"])

Initialize FFprobe runner.

Parameters:

Name	Type	Description	Default
`command_runner`	`CommandRunner \| None`	CommandRunner instance to use. If None, creates a new one.	`None`

Source code in src/video_converter/utils/command_runner.py

def __init__(self, command_runner: CommandRunner | None = None) -> None:
    """Initialize FFprobe runner.

    Args:
        command_runner: CommandRunner instance to use. If None, creates a new one.
    """
    self._runner = command_runner or CommandRunner()

probe ¶

probe(path, *, show_format=True, show_streams=True, timeout=30.0)

Probe a video file and return information as a dictionary.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required
`show_format`	`bool`	Include format information.	`True`
`show_streams`	`bool`	Include stream information.	`True`
`timeout`	`float`	Maximum time to wait (seconds).	`30.0`

Returns:

Type	Description
`dict[str, Any]`	Dictionary containing video information.

Raises:

Type	Description
`CommandNotFoundError`	If FFprobe is not installed.
`CommandExecutionError`	If probing fails.
`JSONDecodeError`	If output cannot be parsed.
`FileNotFoundError`	If the video file doesn't exist.

Source code in src/video_converter/utils/command_runner.py

def probe(
    self,
    path: Path,
    *,
    show_format: bool = True,
    show_streams: bool = True,
    timeout: float = 30.0,
) -> dict[str, Any]:
    """Probe a video file and return information as a dictionary.

    Args:
        path: Path to the video file.
        show_format: Include format information.
        show_streams: Include stream information.
        timeout: Maximum time to wait (seconds).

    Returns:
        Dictionary containing video information.

    Raises:
        CommandNotFoundError: If FFprobe is not installed.
        CommandExecutionError: If probing fails.
        json.JSONDecodeError: If output cannot be parsed.
        FileNotFoundError: If the video file doesn't exist.
    """
    if not path.exists():
        raise FileNotFoundError(f"Video file not found: {path}")

    args = self._build_json_args(
        path,
        show_format=show_format,
        show_streams=show_streams,
    )

    result = self._runner.run(args, timeout=timeout, check=True)
    parsed: dict[str, Any] = json.loads(result.stdout)
    return parsed

probe_async `async` ¶

probe_async(path, *, show_format=True, show_streams=True, timeout=30.0)

Probe a video file asynchronously.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required
`show_format`	`bool`	Include format information.	`True`
`show_streams`	`bool`	Include stream information.	`True`
`timeout`	`float`	Maximum time to wait (seconds).	`30.0`

Returns:

Type	Description
`dict[str, Any]`	Dictionary containing video information.

Source code in src/video_converter/utils/command_runner.py

async def probe_async(
    self,
    path: Path,
    *,
    show_format: bool = True,
    show_streams: bool = True,
    timeout: float = 30.0,
) -> dict[str, Any]:
    """Probe a video file asynchronously.

    Args:
        path: Path to the video file.
        show_format: Include format information.
        show_streams: Include stream information.
        timeout: Maximum time to wait (seconds).

    Returns:
        Dictionary containing video information.
    """
    if not path.exists():
        raise FileNotFoundError(f"Video file not found: {path}")

    args = self._build_json_args(
        path,
        show_format=show_format,
        show_streams=show_streams,
    )

    result = await self._runner.run_async(args, timeout=timeout, check=True)
    parsed: dict[str, Any] = json.loads(result.stdout)
    return parsed

quick_check ¶

quick_check(path, timeout=10.0)

Quickly check if a video file is valid.

This performs a minimal check to see if the file can be opened and basic information can be read.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required
`timeout`	`float`	Maximum time to wait (seconds).	`10.0`

Returns:

Type	Description
`bool`	True if the file is valid, False otherwise.

Source code in src/video_converter/utils/command_runner.py

def quick_check(self, path: Path, timeout: float = 10.0) -> bool:
    """Quickly check if a video file is valid.

    This performs a minimal check to see if the file can be opened
    and basic information can be read.

    Args:
        path: Path to the video file.
        timeout: Maximum time to wait (seconds).

    Returns:
        True if the file is valid, False otherwise.
    """
    if not path.exists():
        return False

    args = [
        self.FFPROBE_CMD,
        "-v",
        "error",
        str(path),
    ]

    try:
        result = self._runner.run(args, timeout=timeout)
        return result.success and not result.stderr.strip()
    except (CommandNotFoundError, CommandExecutionError):
        return False

ExifToolRunner ¶

ExifToolRunner(command_runner=None)

Specialized runner for ExifTool commands.

Provides convenient methods for extracting and modifying metadata from media files using ExifTool.

Example

runner = ExifToolRunner() metadata = runner.read_metadata(Path("video.mp4")) print(metadata.get("CreateDate")) '2024:01:15 10:30:00'

Initialize ExifTool runner.

Parameters:

Name	Type	Description	Default
`command_runner`	`CommandRunner \| None`	CommandRunner instance to use. If None, creates a new one.	`None`

Source code in src/video_converter/utils/command_runner.py

def __init__(self, command_runner: CommandRunner | None = None) -> None:
    """Initialize ExifTool runner.

    Args:
        command_runner: CommandRunner instance to use. If None, creates a new one.
    """
    self._runner = command_runner or CommandRunner()

read_metadata ¶

read_metadata(path, *, timeout=30.0, tags=None)

Read metadata from a file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the file.	required
`timeout`	`float`	Maximum time to wait (seconds).	`30.0`
`tags`	`list[str] \| None`	Specific tags to read. If None, reads all tags.	`None`

Returns:

Type	Description
`dict[str, Any]`	Dictionary containing metadata key-value pairs.

Raises:

Type	Description
`CommandNotFoundError`	If ExifTool is not installed.
`CommandExecutionError`	If reading fails.
`FileNotFoundError`	If the file doesn't exist.

Source code in src/video_converter/utils/command_runner.py

def read_metadata(
    self,
    path: Path,
    *,
    timeout: float = 30.0,
    tags: list[str] | None = None,
) -> dict[str, Any]:
    """Read metadata from a file.

    Args:
        path: Path to the file.
        timeout: Maximum time to wait (seconds).
        tags: Specific tags to read. If None, reads all tags.

    Returns:
        Dictionary containing metadata key-value pairs.

    Raises:
        CommandNotFoundError: If ExifTool is not installed.
        CommandExecutionError: If reading fails.
        FileNotFoundError: If the file doesn't exist.
    """
    if not path.exists():
        raise FileNotFoundError(f"File not found: {path}")

    args = [self.EXIFTOOL_CMD, "-j", "-n"]

    if tags:
        args.extend(f"-{tag}" for tag in tags)

    args.append(str(path))

    result = self._runner.run(args, timeout=timeout, check=True)
    data: list[dict[str, Any]] = json.loads(result.stdout)

    # ExifTool returns a list with one dict per file
    return data[0] if data else {}

read_metadata_async `async` ¶

read_metadata_async(path, *, timeout=30.0, tags=None)

Read metadata from a file asynchronously.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the file.	required
`timeout`	`float`	Maximum time to wait (seconds).	`30.0`
`tags`	`list[str] \| None`	Specific tags to read. If None, reads all tags.	`None`

Returns:

Type	Description
`dict[str, Any]`	Dictionary containing metadata key-value pairs.

Source code in src/video_converter/utils/command_runner.py

async def read_metadata_async(
    self,
    path: Path,
    *,
    timeout: float = 30.0,
    tags: list[str] | None = None,
) -> dict[str, Any]:
    """Read metadata from a file asynchronously.

    Args:
        path: Path to the file.
        timeout: Maximum time to wait (seconds).
        tags: Specific tags to read. If None, reads all tags.

    Returns:
        Dictionary containing metadata key-value pairs.
    """
    if not path.exists():
        raise FileNotFoundError(f"File not found: {path}")

    args = [self.EXIFTOOL_CMD, "-j", "-n"]

    if tags:
        args.extend(f"-{tag}" for tag in tags)

    args.append(str(path))

    result = await self._runner.run_async(args, timeout=timeout, check=True)
    data: list[dict[str, Any]] = json.loads(result.stdout)

    return data[0] if data else {}

write_metadata ¶

write_metadata(path, metadata, *, timeout=30.0, overwrite_original=True)

Write metadata to a file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the file.	required
`metadata`	`dict[str, Any]`	Dictionary of tag-value pairs to write.	required
`timeout`	`float`	Maximum time to wait (seconds).	`30.0`
`overwrite_original`	`bool`	If True, modify file in place without backup.	`True`

Returns:

Type	Description
`bool`	True if metadata was written successfully.

Raises:

Type	Description
`CommandNotFoundError`	If ExifTool is not installed.
`CommandExecutionError`	If writing fails.
`FileNotFoundError`	If the file doesn't exist.

Source code in src/video_converter/utils/command_runner.py

def write_metadata(
    self,
    path: Path,
    metadata: dict[str, Any],
    *,
    timeout: float = 30.0,
    overwrite_original: bool = True,
) -> bool:
    """Write metadata to a file.

    Args:
        path: Path to the file.
        metadata: Dictionary of tag-value pairs to write.
        timeout: Maximum time to wait (seconds).
        overwrite_original: If True, modify file in place without backup.

    Returns:
        True if metadata was written successfully.

    Raises:
        CommandNotFoundError: If ExifTool is not installed.
        CommandExecutionError: If writing fails.
        FileNotFoundError: If the file doesn't exist.
    """
    if not path.exists():
        raise FileNotFoundError(f"File not found: {path}")

    args = [self.EXIFTOOL_CMD]

    if overwrite_original:
        args.append("-overwrite_original")

    for tag, value in metadata.items():
        args.append(f"-{tag}={value}")

    args.append(str(path))

    result = self._runner.run(args, timeout=timeout, check=True)
    return result.success

copy_metadata ¶

copy_metadata(source, dest, *, timeout=30.0, overwrite_original=True, all_tags=True)

Copy metadata from source file to destination file.

Parameters:

Name	Type	Description	Default
`source`	`Path`	Source file to copy metadata from.	required
`dest`	`Path`	Destination file to copy metadata to.	required
`timeout`	`float`	Maximum time to wait (seconds).	`30.0`
`overwrite_original`	`bool`	If True, modify dest in place without backup.	`True`
`all_tags`	`bool`	If True, copy all tags. Otherwise, only common tags.	`True`

Returns:

Type	Description
`bool`	True if metadata was copied successfully.

Raises:

Type	Description
`CommandNotFoundError`	If ExifTool is not installed.
`FileNotFoundError`	If either file doesn't exist.

Source code in src/video_converter/utils/command_runner.py

def copy_metadata(
    self,
    source: Path,
    dest: Path,
    *,
    timeout: float = 30.0,
    overwrite_original: bool = True,
    all_tags: bool = True,
) -> bool:
    """Copy metadata from source file to destination file.

    Args:
        source: Source file to copy metadata from.
        dest: Destination file to copy metadata to.
        timeout: Maximum time to wait (seconds).
        overwrite_original: If True, modify dest in place without backup.
        all_tags: If True, copy all tags. Otherwise, only common tags.

    Returns:
        True if metadata was copied successfully.

    Raises:
        CommandNotFoundError: If ExifTool is not installed.
        FileNotFoundError: If either file doesn't exist.
    """
    if not source.exists():
        raise FileNotFoundError(f"Source file not found: {source}")
    if not dest.exists():
        raise FileNotFoundError(f"Destination file not found: {dest}")

    args = [self.EXIFTOOL_CMD]

    if overwrite_original:
        args.append("-overwrite_original")

    if all_tags:
        args.append("-TagsFromFile")
        args.append(str(source))
        args.append("-all:all")
    else:
        args.append("-TagsFromFile")
        args.append(str(source))

    args.append(str(dest))

    result = self._runner.run(args, timeout=timeout, check=True)
    return result.success

quick_check ¶

quick_check(path, timeout=10.0)

Quickly check if ExifTool can read a file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the file.	required
`timeout`	`float`	Maximum time to wait (seconds).	`10.0`

Returns:

Type	Description
`bool`	True if the file is readable, False otherwise.

Source code in src/video_converter/utils/command_runner.py

def quick_check(self, path: Path, timeout: float = 10.0) -> bool:
    """Quickly check if ExifTool can read a file.

    Args:
        path: Path to the file.
        timeout: Maximum time to wait (seconds).

    Returns:
        True if the file is readable, False otherwise.
    """
    if not path.exists():
        return False

    args = [self.EXIFTOOL_CMD, "-ver", str(path)]

    try:
        result = self._runner.run(args, timeout=timeout)
        return result.success
    except (CommandNotFoundError, CommandExecutionError):
        return False

run_command ¶

run_command(args, *, timeout=60.0, check=False, on_output=None)

Run an external command.

This is a convenience function that creates a CommandRunner instance and executes the command.

Parameters:

Name	Type	Description	Default
`args`	`list[str]`	Command and arguments to execute.	required
`timeout`	`float \| None`	Maximum time to wait for command (seconds).	`60.0`
`check`	`bool`	If True, raise exception on non-zero exit code.	`False`
`on_output`	`Callable[[str], None] \| None`	Optional callback for real-time output streaming.	`None`

Returns:

Type	Description
`CommandResult`	CommandResult containing the execution result.

Example

result = run_command(["ffmpeg", "-version"]) print(result.stdout)

Source code in src/video_converter/utils/command_runner.py

def run_command(
    args: list[str],
    *,
    timeout: float | None = 60.0,
    check: bool = False,
    on_output: Callable[[str], None] | None = None,
) -> CommandResult:
    """Run an external command.

    This is a convenience function that creates a CommandRunner instance
    and executes the command.

    Args:
        args: Command and arguments to execute.
        timeout: Maximum time to wait for command (seconds).
        check: If True, raise exception on non-zero exit code.
        on_output: Optional callback for real-time output streaming.

    Returns:
        CommandResult containing the execution result.

    Example:
        >>> result = run_command(["ffmpeg", "-version"])
        >>> print(result.stdout)
    """
    runner = CommandRunner()
    if on_output is not None:
        return runner.run_with_callback(args, on_output, timeout=timeout, check=check)
    return runner.run(args, timeout=timeout, check=check)

run_ffprobe ¶

run_ffprobe(path, *, timeout=30.0)

Run FFprobe on a video file and return parsed JSON output.

This is a convenience function for quick video analysis.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to the video file.	required
`timeout`	`float`	Maximum time to wait (seconds).	`30.0`

Returns:

Type	Description
`dict[str, Any]`	Dictionary containing video information.

Example

info = run_ffprobe("video.mp4")
print(info["streams"][0]["codec_name"])
# Output: 'h264'

Source code in src/video_converter/utils/command_runner.py

def run_ffprobe(
    path: str | Path,
    *,
    timeout: float = 30.0,
) -> dict[str, Any]:
    """Run FFprobe on a video file and return parsed JSON output.

    This is a convenience function for quick video analysis.

    Args:
        path: Path to the video file.
        timeout: Maximum time to wait (seconds).

    Returns:
        Dictionary containing video information.

    Example:
        ```python
        info = run_ffprobe("video.mp4")
        print(info["streams"][0]["codec_name"])
        # Output: 'h264'
        ```
    """
    runner = FFprobeRunner()
    return runner.probe(Path(path), timeout=timeout)

run_exiftool ¶

run_exiftool(path, *, timeout=30.0, tags=None)

Run ExifTool on a file and return metadata.

This is a convenience function for quick metadata extraction.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to the file.	required
`timeout`	`float`	Maximum time to wait (seconds).	`30.0`
`tags`	`list[str] \| None`	Specific tags to read. If None, reads all tags.	`None`

Returns:

Type	Description
`dict[str, Any]`	Dictionary containing metadata key-value pairs.

Example

metadata = run_exiftool("video.mp4") print(metadata.get("CreateDate")) '2024:01:15 10:30:00'

Source code in src/video_converter/utils/command_runner.py

def run_exiftool(
    path: str | Path,
    *,
    timeout: float = 30.0,
    tags: list[str] | None = None,
) -> dict[str, Any]:
    """Run ExifTool on a file and return metadata.

    This is a convenience function for quick metadata extraction.

    Args:
        path: Path to the file.
        timeout: Maximum time to wait (seconds).
        tags: Specific tags to read. If None, reads all tags.

    Returns:
        Dictionary containing metadata key-value pairs.

    Example:
        >>> metadata = run_exiftool("video.mp4")
        >>> print(metadata.get("CreateDate"))
        '2024:01:15 10:30:00'
    """
    runner = ExifToolRunner()
    return runner.read_metadata(Path(path), timeout=timeout, tags=tags)

Progress Parser¶

progress_parser ¶

FFmpeg progress parsing utilities.

This module provides utilities for parsing FFmpeg stderr output to extract progress information during video conversion.

SDS Reference: SDS-U01-002 SRS Reference: SRS-205 (Real-time Progress Monitoring)

Example

parser = FFmpegProgressParser(total_duration=120.0) progress = parser.parse_line( ... "frame= 720 fps=180 q=32.0 size= 15360kB " ... "time=00:00:24.00 bitrate=5242.9kbits/s speed=6.0x" ... ) print(f"Progress: {progress.percentage:.1f}%") Progress: 20.0%

FFmpegProgress `dataclass` ¶

FFmpegProgress(frame=0, fps=0.0, quality=0.0, size_kb=0, time_seconds=0.0, bitrate_kbps=0.0, speed=0.0, percentage=0.0)

Progress information extracted from FFmpeg output.

Attributes:

Name	Type	Description
`frame`	`int`	Current frame number.
`fps`	`float`	Frames per second being processed.
`quality`	`float`	Quality value (q parameter).
`size_kb`	`int`	Current output size in kilobytes.
`time_seconds`	`float`	Current position in seconds.
`bitrate_kbps`	`float`	Current bitrate in kbits/s.
`speed`	`float`	Encoding speed multiplier (e.g., 6.0 means 6x realtime).
`percentage`	`float`	Progress percentage (0.0-100.0).

FFmpegProgressParser ¶

FFmpegProgressParser(total_duration=0.0)

Parser for FFmpeg stderr progress output.

This parser extracts progress information from FFmpeg's stderr output, which includes frame count, time position, speed, and other metrics.

Attributes:

Name	Type	Description
`total_duration`		Total duration of the video in seconds.

Initialize the progress parser.

Parameters:

Name	Type	Description	Default
`total_duration`	`float`	Total duration of the video in seconds. Used to calculate percentage progress.	`0.0`

Source code in src/video_converter/utils/progress_parser.py

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

    Args:
        total_duration: Total duration of the video in seconds.
                       Used to calculate percentage progress.
    """
    self.total_duration = max(0.0, total_duration)
    self._last_progress = FFmpegProgress()

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
`FFmpegProgress \| None`	FFmpegProgress object if progress info was found, None otherwise.

Source code in src/video_converter/utils/progress_parser.py

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

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

    Returns:
        FFmpegProgress object if progress info was found, None otherwise.
    """
    if "frame=" not in line or "time=" not in line:
        return None

    progress = FFmpegProgress()

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

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

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

    # Parse size
    if match := self._SIZE_PATTERN.search(line):
        progress.size_kb = int(match.group(1))

    # Parse time
    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))
        progress.time_seconds = hours * 3600 + minutes * 60 + seconds + centiseconds / 100

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

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

    # Calculate percentage
    if self.total_duration > 0 and progress.time_seconds > 0:
        progress.percentage = min(100.0, (progress.time_seconds / self.total_duration) * 100)

    self._last_progress = progress
    return progress

get_last_progress ¶

get_last_progress()

Get the last parsed progress.

Returns:

Type	Description
`FFmpegProgress`	The most recently parsed progress, or empty progress if none parsed.

Source code in src/video_converter/utils/progress_parser.py

def get_last_progress(self) -> FFmpegProgress:
    """Get the last parsed progress.

    Returns:
        The most recently parsed progress, or empty progress if none parsed.
    """
    return self._last_progress

parse_time_to_seconds `staticmethod` ¶

parse_time_to_seconds(time_str)

Parse FFmpeg time string to seconds.

Parameters:

Name	Type	Description	Default
`time_str`	`str`	Time string in format HH:MM:SS.cs or similar.	required

Returns:

Type	Description
`float`	Time in seconds.

Source code in src/video_converter/utils/progress_parser.py

@staticmethod
def parse_time_to_seconds(time_str: str) -> float:
    """Parse FFmpeg time string to seconds.

    Args:
        time_str: Time string in format HH:MM:SS.cs or similar.

    Returns:
        Time in seconds.
    """
    pattern = re.compile(r"(\d+):(\d+):(\d+)\.?(\d*)")
    if match := pattern.match(time_str):
        hours = int(match.group(1))
        minutes = int(match.group(2))
        seconds = int(match.group(3))
        centiseconds = int(match.group(4)) if match.group(4) else 0
        return hours * 3600 + minutes * 60 + seconds + centiseconds / 100
    return 0.0

File Utils¶

file_utils ¶

File utilities for path and size management.

This module provides utility functions for file operations including path manipulation, size calculations, temporary file management, and disk space checks.

SDS Reference: SDS-U01-002

Example

from video_converter.utils.file_utils import ( ... expand_path, ... format_size, ... get_temp_path, ... check_disk_space, ... ) path = expand_path("~/Videos/Converted") print(path) /Users/username/Videos/Converted print(format_size(1536000000)) 1.43 GB

InsufficientSpaceError ¶

InsufficientSpaceError(path, required, available)

Bases: Exception

Raised when there is not enough disk space for an operation.

Attributes:

Name	Type	Description
`path`		The path where space was checked.
`required`		Required space in bytes.
`available`		Available space in bytes.

Source code in src/video_converter/utils/file_utils.py

def __init__(self, path: Path, required: int, available: int) -> None:
    self.path = path
    self.required = required
    self.available = available
    super().__init__(
        f"Insufficient disk space at {path}: "
        f"required {format_size(required)}, available {format_size(available)}"
    )

AtomicWriteError ¶

AtomicWriteError(path, reason)

Bases: Exception

Raised when atomic write operation fails.

Attributes:

Name	Type	Description
`path`		The target path for the write operation.
`reason`		Description of why the write failed.

Source code in src/video_converter/utils/file_utils.py

def __init__(self, path: Path, reason: str) -> None:
    self.path = path
    self.reason = reason
    super().__init__(f"Atomic write to {path} failed: {reason}")

expand_path ¶

expand_path(path)

Expand user paths and resolve to absolute path.

Expands ~ to user's home directory and resolves relative paths to absolute paths.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path string or Path object to expand.	required

Returns:

Type	Description
`Path`	Fully expanded and resolved absolute Path.

Example

expand_path("~/Videos/Converted") PosixPath('/Users/username/Videos/Converted') expand_path("./relative/path") PosixPath('/current/working/dir/relative/path')

Source code in src/video_converter/utils/file_utils.py

def expand_path(path: str | Path) -> Path:
    """Expand user paths and resolve to absolute path.

    Expands ~ to user's home directory and resolves relative paths to
    absolute paths.

    Args:
        path: Path string or Path object to expand.

    Returns:
        Fully expanded and resolved absolute Path.

    Example:
        >>> expand_path("~/Videos/Converted")
        PosixPath('/Users/username/Videos/Converted')
        >>> expand_path("./relative/path")
        PosixPath('/current/working/dir/relative/path')
    """
    return Path(path).expanduser().resolve()

format_size ¶

format_size(size_bytes, precision=2)

Format bytes into human-readable size string.

Converts raw byte count to a human-readable format with appropriate unit suffix (B, KB, MB, GB, TB, PB).

Parameters:

Name	Type	Description	Default
`size_bytes`	`int \| float`	Size in bytes.	required
`precision`	`int`	Number of decimal places. Default is 2.	`2`

Returns:

Type	Description
`str`	Human-readable size string (e.g., "1.43 GB").

Example

format_size(1023) '1023 B' format_size(1024) '1.00 KB' format_size(1536000000) '1.43 GB'

Source code in src/video_converter/utils/file_utils.py

def format_size(size_bytes: int | float, precision: int = 2) -> str:
    """Format bytes into human-readable size string.

    Converts raw byte count to a human-readable format with appropriate
    unit suffix (B, KB, MB, GB, TB, PB).

    Args:
        size_bytes: Size in bytes.
        precision: Number of decimal places. Default is 2.

    Returns:
        Human-readable size string (e.g., "1.43 GB").

    Example:
        >>> format_size(1023)
        '1023 B'
        >>> format_size(1024)
        '1.00 KB'
        >>> format_size(1536000000)
        '1.43 GB'
    """
    if size_bytes < 0:
        return f"-{format_size(-size_bytes, precision)}"

    if size_bytes < BYTES_PER_UNIT:
        return f"{int(size_bytes)} B"

    size = float(size_bytes)
    for unit in SIZE_UNITS[1:]:
        size /= BYTES_PER_UNIT
        if size < BYTES_PER_UNIT:
            return f"{size:.{precision}f} {unit}"

    return f"{size:.{precision}f} {SIZE_UNITS[-1]}"

parse_size ¶

parse_size(size_str)

Parse human-readable size string to bytes.

Converts a human-readable size string back to bytes. Supports B, KB, MB, GB, TB, PB units (case-insensitive).

Parameters:

Name	Type	Description	Default
`size_str`	`str`	Size string to parse (e.g., "1.5 GB", "500MB").	required

Returns:

Type	Description
`int`	Size in bytes.

Raises:

Type	Description
`ValueError`	If the size string is invalid.

Example

parse_size("1.5 GB") 1610612736 parse_size("500MB") 524288000

Source code in src/video_converter/utils/file_utils.py

def parse_size(size_str: str) -> int:
    """Parse human-readable size string to bytes.

    Converts a human-readable size string back to bytes. Supports
    B, KB, MB, GB, TB, PB units (case-insensitive).

    Args:
        size_str: Size string to parse (e.g., "1.5 GB", "500MB").

    Returns:
        Size in bytes.

    Raises:
        ValueError: If the size string is invalid.

    Example:
        >>> parse_size("1.5 GB")
        1610612736
        >>> parse_size("500MB")
        524288000
    """
    size_str = size_str.strip().upper()

    # Handle plain bytes
    if size_str.isdigit():
        return int(size_str)

    # Find the numeric part and unit
    for i, char in enumerate(size_str):
        if char.isalpha():
            num_part = size_str[:i].strip()
            unit_part = size_str[i:].strip()
            break
    else:
        raise ValueError(f"Invalid size string: {size_str}")

    try:
        num = float(num_part)
    except ValueError:
        raise ValueError(f"Invalid numeric value in size string: {size_str}") from None

    # Normalize unit (remove trailing 'B' if present, handle 'K' -> 'KB')
    unit_part = unit_part.rstrip("B").strip()
    if not unit_part:
        unit_part = "B"
    elif len(unit_part) == 1 and unit_part != "B":
        unit_part = unit_part + "B"
    else:
        unit_part = unit_part + "B"

    try:
        unit_index = SIZE_UNITS.index(unit_part)
    except ValueError:
        raise ValueError(f"Unknown size unit: {unit_part}") from None

    return int(num * (BYTES_PER_UNIT**unit_index))

get_temp_dir ¶

get_temp_dir()

Get the temporary directory for video converter operations.

Creates a dedicated subdirectory in the system temp directory for video converter temporary files.

Returns:

Type	Description
`Path`	Path to the temporary directory.

Source code in src/video_converter/utils/file_utils.py

def get_temp_dir() -> Path:
    """Get the temporary directory for video converter operations.

    Creates a dedicated subdirectory in the system temp directory for
    video converter temporary files.

    Returns:
        Path to the temporary directory.
    """
    temp_base = Path(tempfile.gettempdir()) / "video_converter"
    temp_base.mkdir(parents=True, exist_ok=True)
    return temp_base

get_temp_path ¶

get_temp_path(filename, suffix=None, unique=True)

Generate a unique temporary file path.

Creates a temporary file path in the video converter temp directory. The path is automatically registered for cleanup on exit.

Parameters:

Name	Type	Description	Default
`filename`	`str \| Path`	Original filename to base the temp name on.	required
`suffix`	`str \| None`	Optional suffix to append (e.g., ".mp4").	`None`
`unique`	`bool`	If True, adds a unique identifier to prevent conflicts.	`True`

Returns:

Type	Description
`Path`	Path to the temporary file location.

Example

get_temp_path("vacation.mp4") PosixPath('/tmp/video_converter/vacation_a1b2c3d4.mp4') get_temp_path("video", suffix=".hevc") PosixPath('/tmp/video_converter/video_e5f6g7h8.hevc')

Source code in src/video_converter/utils/file_utils.py

def get_temp_path(
    filename: str | Path,
    suffix: str | None = None,
    unique: bool = True,
) -> Path:
    """Generate a unique temporary file path.

    Creates a temporary file path in the video converter temp directory.
    The path is automatically registered for cleanup on exit.

    Args:
        filename: Original filename to base the temp name on.
        suffix: Optional suffix to append (e.g., ".mp4").
        unique: If True, adds a unique identifier to prevent conflicts.

    Returns:
        Path to the temporary file location.

    Example:
        >>> get_temp_path("vacation.mp4")
        PosixPath('/tmp/video_converter/vacation_a1b2c3d4.mp4')
        >>> get_temp_path("video", suffix=".hevc")
        PosixPath('/tmp/video_converter/video_e5f6g7h8.hevc')
    """
    temp_dir = get_temp_dir()
    original = Path(filename)

    if suffix is None:
        suffix = original.suffix

    stem = original.stem
    if unique:
        unique_id = uuid.uuid4().hex[:8]
        stem = f"{stem}_{unique_id}"

    temp_path = temp_dir / f"{stem}{suffix}"

    # Register for cleanup
    _temp_files.add(temp_path)

    logger.debug("Generated temp path: %s", temp_path)
    return temp_path

create_temp_directory ¶

create_temp_directory(prefix=None)

Create a temporary directory for video converter operations.

Creates a unique temporary directory that is automatically registered for cleanup on exit.

Parameters:

Name	Type	Description	Default
`prefix`	`str \| None`	Optional prefix for the directory name.	`None`

Returns:

Type	Description
`Path`	Path to the created temporary directory.

Example

temp_dir = create_temp_directory("conversion") print(temp_dir) /tmp/video_converter/conversion_a1b2c3d4

Source code in src/video_converter/utils/file_utils.py

def create_temp_directory(prefix: str | None = None) -> Path:
    """Create a temporary directory for video converter operations.

    Creates a unique temporary directory that is automatically registered
    for cleanup on exit.

    Args:
        prefix: Optional prefix for the directory name.

    Returns:
        Path to the created temporary directory.

    Example:
        >>> temp_dir = create_temp_directory("conversion")
        >>> print(temp_dir)
        /tmp/video_converter/conversion_a1b2c3d4
    """
    temp_base = get_temp_dir()
    prefix = prefix or TEMP_DIR_PREFIX

    unique_id = uuid.uuid4().hex[:8]
    temp_dir = temp_base / f"{prefix}{unique_id}"
    temp_dir.mkdir(parents=True, exist_ok=True)

    # Register for cleanup
    _temp_dirs.add(temp_dir)

    logger.debug("Created temp directory: %s", temp_dir)
    return temp_dir

check_disk_space ¶

check_disk_space(path)

Check available disk space at the given path.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to check disk space for.	required

Returns:

Type	Description
`int`	Available disk space in bytes.

Example

available = check_disk_space("/Users/username") print(format_size(available)) '128.5 GB'

Source code in src/video_converter/utils/file_utils.py

def check_disk_space(path: str | Path) -> int:
    """Check available disk space at the given path.

    Args:
        path: Path to check disk space for.

    Returns:
        Available disk space in bytes.

    Example:
        >>> available = check_disk_space("/Users/username")
        >>> print(format_size(available))
        '128.5 GB'
    """
    path = expand_path(path)

    # Find an existing parent directory
    check_path = path
    while not check_path.exists():
        check_path = check_path.parent
        if check_path == check_path.parent:  # Reached root
            break

    usage = shutil.disk_usage(check_path)
    return usage.free

ensure_disk_space ¶

ensure_disk_space(path, required_bytes, multiplier=2.0)

Ensure sufficient disk space is available.

Checks if there is enough disk space at the given path. For video conversion, it's recommended to have at least 2x the input file size.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to check disk space for.	required
`required_bytes`	`int`	Minimum required space in bytes.	required
`multiplier`	`float`	Safety multiplier for required space. Default is 2.0.	`2.0`

Raises:

Type	Description
`InsufficientSpaceError`	If there is not enough disk space.

Example

ensure_disk_space("/output", input_size, multiplier=2.0)

Source code in src/video_converter/utils/file_utils.py

def ensure_disk_space(
    path: str | Path,
    required_bytes: int,
    multiplier: float = 2.0,
) -> None:
    """Ensure sufficient disk space is available.

    Checks if there is enough disk space at the given path. For video
    conversion, it's recommended to have at least 2x the input file size.

    Args:
        path: Path to check disk space for.
        required_bytes: Minimum required space in bytes.
        multiplier: Safety multiplier for required space. Default is 2.0.

    Raises:
        InsufficientSpaceError: If there is not enough disk space.

    Example:
        >>> ensure_disk_space("/output", input_size, multiplier=2.0)
    """
    path = expand_path(path)
    required = int(required_bytes * multiplier)
    available = check_disk_space(path)

    if available < required:
        raise InsufficientSpaceError(path, required, available)

    logger.debug(
        "Disk space check passed: required=%s, available=%s",
        format_size(required),
        format_size(available),
    )

safe_move ¶

safe_move(src, dst, overwrite=False)

Safely move a file to a new location.

Moves a file with proper error handling. If the destination already exists, behavior depends on the overwrite flag.

Parameters:

Name	Type	Description	Default
`src`	`str \| Path`	Source file path.	required
`dst`	`str \| Path`	Destination file path.	required
`overwrite`	`bool`	If True, overwrite existing destination. Default is False.	`False`

Returns:

Type	Description
`Path`	Path to the destination file.

Raises:

Type	Description
`FileNotFoundError`	If source file doesn't exist.
`FileExistsError`	If destination exists and overwrite is False.
`OSError`	If move operation fails.

Example

safe_move("/tmp/video.mp4", "/output/video.mp4") PosixPath('/output/video.mp4')

Source code in src/video_converter/utils/file_utils.py

def safe_move(src: str | Path, dst: str | Path, overwrite: bool = False) -> Path:
    """Safely move a file to a new location.

    Moves a file with proper error handling. If the destination already
    exists, behavior depends on the overwrite flag.

    Args:
        src: Source file path.
        dst: Destination file path.
        overwrite: If True, overwrite existing destination. Default is False.

    Returns:
        Path to the destination file.

    Raises:
        FileNotFoundError: If source file doesn't exist.
        FileExistsError: If destination exists and overwrite is False.
        OSError: If move operation fails.

    Example:
        >>> safe_move("/tmp/video.mp4", "/output/video.mp4")
        PosixPath('/output/video.mp4')
    """
    src_path = expand_path(src)
    dst_path = expand_path(dst)

    if not src_path.exists():
        raise FileNotFoundError(f"Source file not found: {src_path}")

    if dst_path.exists() and not overwrite:
        raise FileExistsError(f"Destination already exists: {dst_path}")

    # Ensure destination directory exists
    dst_path.parent.mkdir(parents=True, exist_ok=True)

    # Remove destination if overwriting
    if dst_path.exists() and overwrite:
        dst_path.unlink()

    shutil.move(str(src_path), str(dst_path))
    logger.debug("Moved file: %s -> %s", src_path, dst_path)

    return dst_path

safe_copy ¶

safe_copy(src, dst, overwrite=False)

Safely copy a file to a new location.

Copies a file with proper error handling, preserving metadata.

Parameters:

Name	Type	Description	Default
`src`	`str \| Path`	Source file path.	required
`dst`	`str \| Path`	Destination file path.	required
`overwrite`	`bool`	If True, overwrite existing destination. Default is False.	`False`

Returns:

Type	Description
`Path`	Path to the destination file.

Raises:

Type	Description
`FileNotFoundError`	If source file doesn't exist.
`FileExistsError`	If destination exists and overwrite is False.
`OSError`	If copy operation fails.

Example

safe_copy("/input/video.mp4", "/backup/video.mp4") PosixPath('/backup/video.mp4')

Source code in src/video_converter/utils/file_utils.py

def safe_copy(src: str | Path, dst: str | Path, overwrite: bool = False) -> Path:
    """Safely copy a file to a new location.

    Copies a file with proper error handling, preserving metadata.

    Args:
        src: Source file path.
        dst: Destination file path.
        overwrite: If True, overwrite existing destination. Default is False.

    Returns:
        Path to the destination file.

    Raises:
        FileNotFoundError: If source file doesn't exist.
        FileExistsError: If destination exists and overwrite is False.
        OSError: If copy operation fails.

    Example:
        >>> safe_copy("/input/video.mp4", "/backup/video.mp4")
        PosixPath('/backup/video.mp4')
    """
    src_path = expand_path(src)
    dst_path = expand_path(dst)

    if not src_path.exists():
        raise FileNotFoundError(f"Source file not found: {src_path}")

    if dst_path.exists() and not overwrite:
        raise FileExistsError(f"Destination already exists: {dst_path}")

    # Ensure destination directory exists
    dst_path.parent.mkdir(parents=True, exist_ok=True)

    # Remove destination if overwriting
    if dst_path.exists() and overwrite:
        dst_path.unlink()

    shutil.copy2(str(src_path), str(dst_path))
    logger.debug("Copied file: %s -> %s", src_path, dst_path)

    return dst_path

safe_delete ¶

safe_delete(path, missing_ok=True)

Safely delete a file.

Deletes a file with proper error handling.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to the file to delete.	required
`missing_ok`	`bool`	If True, don't raise error if file doesn't exist.	`True`

Returns:

Type	Description
`bool`	True if file was deleted, False if it didn't exist.

Raises:

Type	Description
`FileNotFoundError`	If file doesn't exist and missing_ok is False.
`OSError`	If deletion fails.

Example

safe_delete("/tmp/video.mp4") True

Source code in src/video_converter/utils/file_utils.py

def safe_delete(path: str | Path, missing_ok: bool = True) -> bool:
    """Safely delete a file.

    Deletes a file with proper error handling.

    Args:
        path: Path to the file to delete.
        missing_ok: If True, don't raise error if file doesn't exist.

    Returns:
        True if file was deleted, False if it didn't exist.

    Raises:
        FileNotFoundError: If file doesn't exist and missing_ok is False.
        OSError: If deletion fails.

    Example:
        >>> safe_delete("/tmp/video.mp4")
        True
    """
    file_path = expand_path(path)

    if not file_path.exists():
        if missing_ok:
            return False
        raise FileNotFoundError(f"File not found: {file_path}")

    file_path.unlink()
    logger.debug("Deleted file: %s", file_path)

    return True

atomic_write ¶

atomic_write(path, mode='wb')

Context manager for atomic file writes.

Writes to a temporary file first, then atomically moves it to the target location. This prevents file corruption if the write is interrupted.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Target file path.	required
`mode`	`str`	File open mode. Default is "wb" (write binary).	`'wb'`

Yields:

Type	Description
`Path`	Path to the temporary file to write to.

Raises:

Type	Description
`AtomicWriteError`	If the atomic move fails.

Example

with atomic_write("/output/video.mp4") as temp_path: ... # Write to temp_path ... shutil.copy("input.mp4", temp_path)

File is now atomically moved to /output/video.mp4¶

Source code in src/video_converter/utils/file_utils.py

@contextmanager
def atomic_write(
    path: str | Path,
    mode: str = "wb",
) -> Generator[Path, None, None]:
    """Context manager for atomic file writes.

    Writes to a temporary file first, then atomically moves it to the
    target location. This prevents file corruption if the write is
    interrupted.

    Args:
        path: Target file path.
        mode: File open mode. Default is "wb" (write binary).

    Yields:
        Path to the temporary file to write to.

    Raises:
        AtomicWriteError: If the atomic move fails.

    Example:
        >>> with atomic_write("/output/video.mp4") as temp_path:
        ...     # Write to temp_path
        ...     shutil.copy("input.mp4", temp_path)
        >>> # File is now atomically moved to /output/video.mp4
    """
    target_path = expand_path(path)
    temp_path = get_temp_path(target_path.name)

    try:
        yield temp_path

        # If we get here, write was successful - move to target
        if temp_path.exists():
            # Ensure target directory exists
            target_path.parent.mkdir(parents=True, exist_ok=True)

            # Atomic move
            shutil.move(str(temp_path), str(target_path))
            logger.debug("Atomic write completed: %s", target_path)

    except Exception as e:
        # Clean up temp file on failure
        if temp_path.exists():
            try:
                temp_path.unlink()
            except OSError:
                pass
        raise AtomicWriteError(target_path, str(e)) from e

    finally:
        # Remove from tracking since we handled it
        _temp_files.discard(temp_path)

get_file_size ¶

get_file_size(path)

Get the size of a file in bytes.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to the file.	required

Returns:

Type	Description
`int`	File size in bytes.

Raises:

Type	Description
`FileNotFoundError`	If file doesn't exist.

Example

size = get_file_size("/path/to/video.mp4") print(format_size(size)) '1.5 GB'

Source code in src/video_converter/utils/file_utils.py

def get_file_size(path: str | Path) -> int:
    """Get the size of a file in bytes.

    Args:
        path: Path to the file.

    Returns:
        File size in bytes.

    Raises:
        FileNotFoundError: If file doesn't exist.

    Example:
        >>> size = get_file_size("/path/to/video.mp4")
        >>> print(format_size(size))
        '1.5 GB'
    """
    file_path = expand_path(path)

    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {file_path}")

    return file_path.stat().st_size

get_directory_size ¶

get_directory_size(path)

Get the total size of all files in a directory.

Recursively calculates the total size of all files in a directory.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to the directory.	required

Returns:

Type	Description
`int`	Total size in bytes.

Raises:

Type	Description
`NotADirectoryError`	If path is not a directory.

Example

size = get_directory_size("/path/to/videos") print(format_size(size)) '15.3 GB'

Source code in src/video_converter/utils/file_utils.py

def get_directory_size(path: str | Path) -> int:
    """Get the total size of all files in a directory.

    Recursively calculates the total size of all files in a directory.

    Args:
        path: Path to the directory.

    Returns:
        Total size in bytes.

    Raises:
        NotADirectoryError: If path is not a directory.

    Example:
        >>> size = get_directory_size("/path/to/videos")
        >>> print(format_size(size))
        '15.3 GB'
    """
    dir_path = expand_path(path)

    if not dir_path.is_dir():
        raise NotADirectoryError(f"Not a directory: {dir_path}")

    total_size = 0
    for file_path in dir_path.rglob("*"):
        if file_path.is_file():
            total_size += file_path.stat().st_size

    return total_size

cleanup_temp_files ¶

cleanup_temp_files()

Clean up all registered temporary files and directories.

Removes all temporary files and directories that were created using get_temp_path() and create_temp_directory().

Returns:

Type	Description
`int`	Number of items cleaned up.

Source code in src/video_converter/utils/file_utils.py

def cleanup_temp_files() -> int:
    """Clean up all registered temporary files and directories.

    Removes all temporary files and directories that were created using
    get_temp_path() and create_temp_directory().

    Returns:
        Number of items cleaned up.
    """
    cleaned = 0

    # Clean up files
    for temp_file in list(_temp_files):
        try:
            if temp_file.exists():
                temp_file.unlink()
                cleaned += 1
                logger.debug("Cleaned up temp file: %s", temp_file)
        except OSError as e:
            logger.warning("Failed to clean up temp file %s: %s", temp_file, e)
        finally:
            _temp_files.discard(temp_file)

    # Clean up directories
    for temp_dir in list(_temp_dirs):
        try:
            if temp_dir.exists():
                shutil.rmtree(temp_dir)
                cleaned += 1
                logger.debug("Cleaned up temp directory: %s", temp_dir)
        except OSError as e:
            logger.warning("Failed to clean up temp directory %s: %s", temp_dir, e)
        finally:
            _temp_dirs.discard(temp_dir)

    if cleaned > 0:
        logger.info("Cleaned up %d temporary items", cleaned)

    return cleaned

ensure_directory ¶

ensure_directory(path)

Ensure a directory exists, creating it if necessary.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to the directory.	required

Returns:

Type	Description
`Path`	Path to the directory.

Example

ensure_directory("~/Videos/Converted") PosixPath('/Users/username/Videos/Converted')

Source code in src/video_converter/utils/file_utils.py

def ensure_directory(path: str | Path) -> Path:
    """Ensure a directory exists, creating it if necessary.

    Args:
        path: Path to the directory.

    Returns:
        Path to the directory.

    Example:
        >>> ensure_directory("~/Videos/Converted")
        PosixPath('/Users/username/Videos/Converted')
    """
    dir_path = expand_path(path)
    dir_path.mkdir(parents=True, exist_ok=True)
    return dir_path

is_video_file ¶

is_video_file(path)

Check if a path points to a video file based on extension.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to check.	required

Returns:

Type	Description
`bool`	True if the file has a video extension.

Example

is_video_file("/path/to/video.mp4") True is_video_file("/path/to/image.jpg") False

Source code in src/video_converter/utils/file_utils.py

def is_video_file(path: str | Path) -> bool:
    """Check if a path points to a video file based on extension.

    Args:
        path: Path to check.

    Returns:
        True if the file has a video extension.

    Example:
        >>> is_video_file("/path/to/video.mp4")
        True
        >>> is_video_file("/path/to/image.jpg")
        False
    """
    video_extensions = {".mp4", ".mov", ".m4v", ".mkv", ".avi", ".wmv", ".webm"}
    file_path = Path(path)
    return file_path.suffix.lower() in video_extensions

generate_output_path ¶

generate_output_path(input_path, output_dir=None, suffix='_hevc', extension=None)

Generate an output path for a converted video.

Parameters:

Name	Type	Description	Default
`input_path`	`str \| Path`	Path to the input video.	required
`output_dir`	`str \| Path \| None`	Directory for output. If None, uses input directory.	`None`
`suffix`	`str`	Suffix to add to filename. Default is "_hevc".	`'_hevc'`
`extension`	`str \| None`	New file extension. If None, keeps original.	`None`

Returns:

Type	Description
`Path`	Generated output path.

Example

generate_output_path("/input/video.mp4") PosixPath('/input/video_hevc.mp4') generate_output_path("/input/video.mp4", "/output", suffix="") PosixPath('/output/video.mp4')

Source code in src/video_converter/utils/file_utils.py

def generate_output_path(
    input_path: str | Path,
    output_dir: str | Path | None = None,
    suffix: str = "_hevc",
    extension: str | None = None,
) -> Path:
    """Generate an output path for a converted video.

    Args:
        input_path: Path to the input video.
        output_dir: Directory for output. If None, uses input directory.
        suffix: Suffix to add to filename. Default is "_hevc".
        extension: New file extension. If None, keeps original.

    Returns:
        Generated output path.

    Example:
        >>> generate_output_path("/input/video.mp4")
        PosixPath('/input/video_hevc.mp4')
        >>> generate_output_path("/input/video.mp4", "/output", suffix="")
        PosixPath('/output/video.mp4')
    """
    input_path = expand_path(input_path)

    if output_dir is None:
        output_dir = input_path.parent
    else:
        output_dir = expand_path(output_dir)

    stem = input_path.stem
    ext = extension if extension else input_path.suffix

    if not ext.startswith("."):
        ext = f".{ext}"

    return output_dir / f"{stem}{suffix}{ext}"

Dependency Checker¶

dependency_checker ¶

Dependency checker for required external tools.

This module validates that all required external tools are installed and provides installation instructions for missing dependencies.

SDS Reference: SDS-U01-002

DependencyStatus ¶

Bases: Enum

Status of a dependency check.

DependencyInfo `dataclass` ¶

DependencyInfo(name, required_version, current_version, status, install_instruction, description='')

Information about a single dependency.

Attributes:

Name	Type	Description
`name`	`str`	Name of the dependency.
`required_version`	`str \| None`	Minimum required version (None if any version is OK).
`current_version`	`str \| None`	Currently installed version (None if not found).
`status`	`DependencyStatus`	Status of the dependency check.
`install_instruction`	`str`	How to install this dependency.
`description`	`str`	Brief description of what this dependency is for.

is_satisfied `property` ¶

is_satisfied

Check if dependency is satisfied.

DependencyCheckResult `dataclass` ¶

DependencyCheckResult(dependencies=list())

Result of checking all dependencies.

Attributes:

Name	Type	Description
`dependencies`	`list[DependencyInfo]`	List of all checked dependencies.
`all_satisfied`	`bool`	Whether all dependencies are satisfied.

all_satisfied `property` ¶

all_satisfied

Check if all dependencies are satisfied.

missing `property` ¶

missing

Get list of missing or unsatisfied dependencies.

satisfied `property` ¶

satisfied

Get list of satisfied dependencies.

DependencyChecker ¶

DependencyChecker(command_runner=None)

Check for required external dependencies.

This class validates that all required tools are installed and provides installation instructions for missing dependencies.

Example

checker = DependencyChecker() result = checker.check_all() if not result.all_satisfied: ... for dep in result.missing: ... print(f"Missing: {dep.name}") ... print(f"Install: {dep.install_instruction}")

Initialize dependency checker.

Parameters:

Name	Type	Description	Default
`command_runner`	`CommandRunner \| None`	CommandRunner instance to use. If None, creates a new one.	`None`

Source code in src/video_converter/utils/dependency_checker.py

def __init__(self, command_runner: CommandRunner | None = None) -> None:
    """Initialize dependency checker.

    Args:
        command_runner: CommandRunner instance to use. If None, creates a new one.
    """
    self._runner = command_runner or CommandRunner()

check_all ¶

check_all()

Check all dependencies.

Returns:

Type	Description
`DependencyCheckResult`	DependencyCheckResult containing status of all dependencies.

Source code in src/video_converter/utils/dependency_checker.py

def check_all(self) -> DependencyCheckResult:
    """Check all dependencies.

    Returns:
        DependencyCheckResult containing status of all dependencies.
    """
    result = DependencyCheckResult()

    # Check each dependency
    result.dependencies.append(self.check_macos())
    result.dependencies.append(self.check_python())
    result.dependencies.append(self.check_ffmpeg())
    result.dependencies.append(self.check_videotoolbox())
    result.dependencies.append(self.check_exiftool())
    result.dependencies.append(self.check_osxphotos())

    return result

check_macos ¶

check_macos()

Check macOS version.

Returns:

Type	Description
`DependencyInfo`	DependencyInfo for macOS.

Source code in src/video_converter/utils/dependency_checker.py

def check_macos(self) -> DependencyInfo:
    """Check macOS version.

    Returns:
        DependencyInfo for macOS.
    """
    name = "macOS"
    install_instruction = "Upgrade macOS to version 12.0 (Monterey) or later"
    description = "Operating system"

    try:
        # Get macOS version using sw_vers
        result = self._runner.run(["sw_vers", "-productVersion"], timeout=5.0)
        if not result.success:
            return DependencyInfo(
                name=name,
                required_version=self.MIN_MACOS_VERSION,
                current_version=None,
                status=DependencyStatus.CHECK_FAILED,
                install_instruction=install_instruction,
                description=description,
            )

        version = result.stdout.strip()
        logger.debug("Detected macOS version: %s", version)

        # Get macOS name
        name_result = self._runner.run(["sw_vers", "-productName"], timeout=5.0)
        product_name = name_result.stdout.strip() if name_result.success else "macOS"

        if compare_versions(version, self.MIN_MACOS_VERSION) >= 0:
            return DependencyInfo(
                name=f"{product_name} {version}",
                required_version=self.MIN_MACOS_VERSION,
                current_version=version,
                status=DependencyStatus.SATISFIED,
                install_instruction=install_instruction,
                description=description,
            )
        else:
            return DependencyInfo(
                name=name,
                required_version=self.MIN_MACOS_VERSION,
                current_version=version,
                status=DependencyStatus.VERSION_TOO_LOW,
                install_instruction=install_instruction,
                description=description,
            )

    except Exception as e:
        logger.warning("Failed to check macOS version: %s", e)
        return DependencyInfo(
            name=name,
            required_version=self.MIN_MACOS_VERSION,
            current_version=None,
            status=DependencyStatus.CHECK_FAILED,
            install_instruction=install_instruction,
            description=description,
        )

check_python ¶

check_python()

Check Python version.

Returns:

Type	Description
`DependencyInfo`	DependencyInfo for Python.

Source code in src/video_converter/utils/dependency_checker.py

def check_python(self) -> DependencyInfo:
    """Check Python version.

    Returns:
        DependencyInfo for Python.
    """
    name = "Python"
    install_instruction = "Install Python 3.10+ from python.org or brew install python@3.10"
    description = "Python interpreter"

    version = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
    version_short = f"{sys.version_info.major}.{sys.version_info.minor}"

    if compare_versions(version_short, self.MIN_PYTHON_VERSION) >= 0:
        return DependencyInfo(
            name=f"Python {version}",
            required_version=self.MIN_PYTHON_VERSION,
            current_version=version,
            status=DependencyStatus.SATISFIED,
            install_instruction=install_instruction,
            description=description,
        )
    else:
        return DependencyInfo(
            name=name,
            required_version=self.MIN_PYTHON_VERSION,
            current_version=version,
            status=DependencyStatus.VERSION_TOO_LOW,
            install_instruction=install_instruction,
            description=description,
        )

check_ffmpeg ¶

check_ffmpeg()

Check FFmpeg installation and version.

Returns:

Type	Description
`DependencyInfo`	DependencyInfo for FFmpeg.

Source code in src/video_converter/utils/dependency_checker.py

def check_ffmpeg(self) -> DependencyInfo:
    """Check FFmpeg installation and version.

    Returns:
        DependencyInfo for FFmpeg.
    """
    name = "FFmpeg"
    install_instruction = "brew install ffmpeg"
    description = "Video processing tool"

    try:
        if not self._runner.check_command_exists("ffmpeg"):
            return DependencyInfo(
                name=name,
                required_version=self.MIN_FFMPEG_VERSION,
                current_version=None,
                status=DependencyStatus.MISSING,
                install_instruction=install_instruction,
                description=description,
            )

        # Get FFmpeg version
        result = self._runner.run(["ffmpeg", "-version"], timeout=5.0)
        if not result.success:
            return DependencyInfo(
                name=name,
                required_version=self.MIN_FFMPEG_VERSION,
                current_version=None,
                status=DependencyStatus.CHECK_FAILED,
                install_instruction=install_instruction,
                description=description,
            )

        # Parse version from output: "ffmpeg version 6.1 Copyright..."
        version_match = re.search(r"ffmpeg version (\d+\.\d+(?:\.\d+)?)", result.stdout)
        if not version_match:
            # Try alternative format: "ffmpeg version N-xxxxx-..."
            version_match = re.search(r"ffmpeg version (\d+)", result.stdout)

        if version_match:
            version = version_match.group(1)
            logger.debug("Detected FFmpeg version: %s", version)

            if compare_versions(version, self.MIN_FFMPEG_VERSION) >= 0:
                return DependencyInfo(
                    name=f"FFmpeg {version}",
                    required_version=self.MIN_FFMPEG_VERSION,
                    current_version=version,
                    status=DependencyStatus.SATISFIED,
                    install_instruction=install_instruction,
                    description=description,
                )
            else:
                return DependencyInfo(
                    name=name,
                    required_version=self.MIN_FFMPEG_VERSION,
                    current_version=version,
                    status=DependencyStatus.VERSION_TOO_LOW,
                    install_instruction="brew upgrade ffmpeg",
                    description=description,
                )
        else:
            return DependencyInfo(
                name=name,
                required_version=self.MIN_FFMPEG_VERSION,
                current_version="unknown",
                status=DependencyStatus.SATISFIED,  # Assume OK if we can't parse
                install_instruction=install_instruction,
                description=description,
            )

    except Exception as e:
        logger.warning("Failed to check FFmpeg: %s", e)
        return DependencyInfo(
            name=name,
            required_version=self.MIN_FFMPEG_VERSION,
            current_version=None,
            status=DependencyStatus.CHECK_FAILED,
            install_instruction=install_instruction,
            description=description,
        )

check_videotoolbox ¶

check_videotoolbox()

Check VideoToolbox/hevc_videotoolbox encoder availability.

Returns:

Type	Description
`DependencyInfo`	DependencyInfo for VideoToolbox encoder.

Source code in src/video_converter/utils/dependency_checker.py

def check_videotoolbox(self) -> DependencyInfo:
    """Check VideoToolbox/hevc_videotoolbox encoder availability.

    Returns:
        DependencyInfo for VideoToolbox encoder.
    """
    name = "hevc_videotoolbox"
    install_instruction = "brew reinstall ffmpeg (ensure VideoToolbox is enabled)"
    description = "Hardware H.265 encoder"

    try:
        if not self._runner.check_command_exists("ffmpeg"):
            return DependencyInfo(
                name=name,
                required_version=None,
                current_version=None,
                status=DependencyStatus.MISSING,
                install_instruction="Install FFmpeg first: brew install ffmpeg",
                description=description,
            )

        # Check if hevc_videotoolbox encoder is available
        result = self._runner.run(["ffmpeg", "-encoders"], timeout=5.0)
        if not result.success:
            return DependencyInfo(
                name=name,
                required_version=None,
                current_version=None,
                status=DependencyStatus.CHECK_FAILED,
                install_instruction=install_instruction,
                description=description,
            )

        if "hevc_videotoolbox" in result.stdout:
            return DependencyInfo(
                name=name,
                required_version=None,
                current_version="available",
                status=DependencyStatus.SATISFIED,
                install_instruction=install_instruction,
                description=description,
            )
        else:
            return DependencyInfo(
                name=name,
                required_version=None,
                current_version=None,
                status=DependencyStatus.MISSING,
                install_instruction=install_instruction,
                description=description,
            )

    except Exception as e:
        logger.warning("Failed to check VideoToolbox: %s", e)
        return DependencyInfo(
            name=name,
            required_version=None,
            current_version=None,
            status=DependencyStatus.CHECK_FAILED,
            install_instruction=install_instruction,
            description=description,
        )

check_exiftool ¶

check_exiftool()

Check ExifTool installation and version.

Returns:

Type	Description
`DependencyInfo`	DependencyInfo for ExifTool.

Source code in src/video_converter/utils/dependency_checker.py

def check_exiftool(self) -> DependencyInfo:
    """Check ExifTool installation and version.

    Returns:
        DependencyInfo for ExifTool.
    """
    name = "ExifTool"
    install_instruction = "brew install exiftool"
    description = "Metadata extraction tool"

    try:
        if not self._runner.check_command_exists("exiftool"):
            return DependencyInfo(
                name=name,
                required_version=self.MIN_EXIFTOOL_VERSION,
                current_version=None,
                status=DependencyStatus.MISSING,
                install_instruction=install_instruction,
                description=description,
            )

        # Get ExifTool version
        result = self._runner.run(["exiftool", "-ver"], timeout=5.0)
        if not result.success:
            return DependencyInfo(
                name=name,
                required_version=self.MIN_EXIFTOOL_VERSION,
                current_version=None,
                status=DependencyStatus.CHECK_FAILED,
                install_instruction=install_instruction,
                description=description,
            )

        version = result.stdout.strip()
        logger.debug("Detected ExifTool version: %s", version)

        if compare_versions(version, self.MIN_EXIFTOOL_VERSION) >= 0:
            return DependencyInfo(
                name=f"ExifTool {version}",
                required_version=self.MIN_EXIFTOOL_VERSION,
                current_version=version,
                status=DependencyStatus.SATISFIED,
                install_instruction=install_instruction,
                description=description,
            )
        else:
            return DependencyInfo(
                name=name,
                required_version=self.MIN_EXIFTOOL_VERSION,
                current_version=version,
                status=DependencyStatus.VERSION_TOO_LOW,
                install_instruction="brew upgrade exiftool",
                description=description,
            )

    except Exception as e:
        logger.warning("Failed to check ExifTool: %s", e)
        return DependencyInfo(
            name=name,
            required_version=self.MIN_EXIFTOOL_VERSION,
            current_version=None,
            status=DependencyStatus.CHECK_FAILED,
            install_instruction=install_instruction,
            description=description,
        )

check_osxphotos ¶

check_osxphotos()

Check osxphotos package installation.

Returns:

Type	Description
`DependencyInfo`	DependencyInfo for osxphotos.

Source code in src/video_converter/utils/dependency_checker.py

def check_osxphotos(self) -> DependencyInfo:
    """Check osxphotos package installation.

    Returns:
        DependencyInfo for osxphotos.
    """
    name = "osxphotos"
    install_instruction = "pip install osxphotos>=0.70"
    description = "macOS Photos library access"

    try:
        import importlib.metadata

        version = importlib.metadata.version("osxphotos")
        logger.debug("Detected osxphotos version: %s", version)

        if compare_versions(version, self.MIN_OSXPHOTOS_VERSION) >= 0:
            return DependencyInfo(
                name=f"osxphotos {version}",
                required_version=self.MIN_OSXPHOTOS_VERSION,
                current_version=version,
                status=DependencyStatus.SATISFIED,
                install_instruction=install_instruction,
                description=description,
            )
        else:
            return DependencyInfo(
                name=name,
                required_version=self.MIN_OSXPHOTOS_VERSION,
                current_version=version,
                status=DependencyStatus.VERSION_TOO_LOW,
                install_instruction="pip install --upgrade osxphotos",
                description=description,
            )

    except importlib.metadata.PackageNotFoundError:
        return DependencyInfo(
            name=name,
            required_version=self.MIN_OSXPHOTOS_VERSION,
            current_version=None,
            status=DependencyStatus.MISSING,
            install_instruction=install_instruction,
            description=description,
        )
    except Exception as e:
        logger.warning("Failed to check osxphotos: %s", e)
        return DependencyInfo(
            name=name,
            required_version=self.MIN_OSXPHOTOS_VERSION,
            current_version=None,
            status=DependencyStatus.CHECK_FAILED,
            install_instruction=install_instruction,
            description=description,
        )

format_report ¶

format_report(result)

Format dependency check result as a readable report.

Parameters:

Name	Type	Description	Default
`result`	`DependencyCheckResult`	The dependency check result.	required

Returns:

Type	Description
`str`	Formatted string report.

Source code in src/video_converter/utils/dependency_checker.py

def format_report(self, result: DependencyCheckResult) -> str:
    """Format dependency check result as a readable report.

    Args:
        result: The dependency check result.

    Returns:
        Formatted string report.
    """
    lines = ["Checking dependencies...", ""]

    for dep in result.dependencies:
        if dep.is_satisfied:
            status_icon = "✓"
            lines.append(f"{status_icon} {dep.name}")
        else:
            status_icon = "✗"
            if dep.status == DependencyStatus.MISSING:
                lines.append(f"{status_icon} {dep.name} not found")
            elif dep.status == DependencyStatus.VERSION_TOO_LOW:
                lines.append(
                    f"{status_icon} {dep.name} {dep.current_version} "
                    f"(requires {dep.required_version}+)"
                )
            else:
                lines.append(f"{status_icon} {dep.name} (check failed)")

    lines.append("")

    if result.all_satisfied:
        lines.append("All dependencies satisfied!")
    else:
        lines.append("Missing or outdated dependencies:")
        for dep in result.missing:
            lines.append(f"  {dep.name}: {dep.install_instruction}")
        lines.append("")
        lines.append("Please install missing dependencies and try again.")

    return "\n".join(lines)

compare_versions ¶

compare_versions(version1, version2)

Compare two version strings.

Supports semantic versioning (e.g., "5.0", "5.1.2", "6.0-beta").

Parameters:

Name	Type	Description	Default
`version1`	`str`	First version string.	required
`version2`	`str`	Second version string.	required

Returns:

Type	Description
`int`	-1 if version1 < version2
`int`	0 if version1 == version2
`int`	1 if version1 > version2

Example

compare_versions("5.0", "5.1") -1 compare_versions("6.0", "5.1") 1 compare_versions("5.1", "5.1") 0

Source code in src/video_converter/utils/dependency_checker.py

def compare_versions(version1: str, version2: str) -> int:
    """Compare two version strings.

    Supports semantic versioning (e.g., "5.0", "5.1.2", "6.0-beta").

    Args:
        version1: First version string.
        version2: Second version string.

    Returns:
        -1 if version1 < version2
        0 if version1 == version2
        1 if version1 > version2

    Example:
        >>> compare_versions("5.0", "5.1")
        -1
        >>> compare_versions("6.0", "5.1")
        1
        >>> compare_versions("5.1", "5.1")
        0
    """

    def normalize(v: str) -> list[int]:
        """Normalize version string to list of integers."""
        # Remove any suffix like -beta, -rc1, etc.
        v = re.split(r"[-_]", v)[0]
        # Split by dots and convert to integers
        parts = []
        for part in v.split("."):
            # Extract leading digits
            match = re.match(r"(\d+)", part)
            if match:
                parts.append(int(match.group(1)))
        return parts

    v1_parts = normalize(version1)
    v2_parts = normalize(version2)

    # Pad shorter version with zeros
    max_len = max(len(v1_parts), len(v2_parts))
    v1_parts.extend([0] * (max_len - len(v1_parts)))
    v2_parts.extend([0] * (max_len - len(v2_parts)))

    for p1, p2 in zip(v1_parts, v2_parts, strict=True):
        if p1 < p2:
            return -1
        if p1 > p2:
            return 1

    return 0

Utils Module¶

Command Runner¶

command_runner ¶

CommandResult dataclass ¶

success property ¶

CommandNotFoundError ¶

CommandTimeoutError ¶

CommandExecutionError ¶

CommandRunner ¶

check_command_exists staticmethod ¶

ensure_command_exists staticmethod ¶

run ¶

run_async async ¶

run_with_callback ¶

FFprobeRunner ¶

probe ¶

probe_async async ¶

quick_check ¶

ExifToolRunner ¶

read_metadata ¶

read_metadata_async async ¶

write_metadata ¶

copy_metadata ¶

quick_check ¶

run_command ¶

run_ffprobe ¶

run_exiftool ¶

Progress Parser¶

progress_parser ¶

FFmpegProgress dataclass ¶

FFmpegProgressParser ¶

parse_line ¶

get_last_progress ¶

parse_time_to_seconds staticmethod ¶

File Utils¶

file_utils ¶

InsufficientSpaceError ¶

AtomicWriteError ¶

expand_path ¶

format_size ¶

parse_size ¶

get_temp_dir ¶

get_temp_path ¶

create_temp_directory ¶

check_disk_space ¶

ensure_disk_space ¶

safe_move ¶

safe_copy ¶

safe_delete ¶

atomic_write ¶

File is now atomically moved to /output/video.mp4¶

get_file_size ¶

get_directory_size ¶

cleanup_temp_files ¶

ensure_directory ¶

is_video_file ¶

generate_output_path ¶

Dependency Checker¶

dependency_checker ¶

DependencyStatus ¶

DependencyInfo dataclass ¶

is_satisfied property ¶

DependencyCheckResult dataclass ¶

all_satisfied property ¶

missing property ¶

satisfied property ¶

DependencyChecker ¶

check_all ¶

check_macos ¶

check_python ¶

check_ffmpeg ¶

check_videotoolbox ¶

check_exiftool ¶

check_osxphotos ¶

format_report ¶

compare_versions ¶

CommandResult `dataclass` ¶

success `property` ¶

check_command_exists `staticmethod` ¶

ensure_command_exists `staticmethod` ¶

run_async `async` ¶

probe_async `async` ¶

read_metadata_async `async` ¶

FFmpegProgress `dataclass` ¶

parse_time_to_seconds `staticmethod` ¶

DependencyInfo `dataclass` ¶

is_satisfied `property` ¶

DependencyCheckResult `dataclass` ¶

all_satisfied `property` ¶

missing `property` ¶

satisfied `property` ¶