Processors Module¶

Video processing components for analysis, metadata, and validation.

Codec Detector¶

codec_detector ¶

Video codec detection module using FFprobe.

This module provides codec detection functionality to analyze video files and determine their codec, resolution, and other properties. It is used to identify H.264 videos that need conversion to H.265/HEVC.

SDS Reference: SDS-P01-002 SRS Reference: SRS-201 (Codec Detection)

Example

from video_converter.processors.codec_detector import CodecDetector detector = CodecDetector() info = detector.analyze(Path("vacation.mp4")) print(f"Codec: {info.codec}") # "h264" print(f"Resolution: {info.resolution_label}") # "4K" print(f"Needs conversion: {info.is_h264}") # True

InvalidVideoError ¶

InvalidVideoError(path, reason='Not a valid video file')

Bases: Exception

Raised when a file is not a valid video.

This error indicates that the file cannot be recognized as a valid video format by FFprobe.

Attributes:

Name	Type	Description
`path`		Path to the invalid file.
`reason`		Explanation of why the file is invalid.

Source code in src/video_converter/processors/codec_detector.py

def __init__(self, path: Path, reason: str = "Not a valid video file") -> None:
    self.path = path
    self.reason = reason
    super().__init__(f"{path}: {reason}")

CorruptedVideoError ¶

CorruptedVideoError(path, reason='Video file is corrupted')

Bases: Exception

Raised when a video file is corrupted or incomplete.

This error indicates that while the file is recognized as a video, it cannot be properly read or is missing critical data.

Attributes:

Name	Type	Description
`path`		Path to the corrupted file.
`reason`		Explanation of the corruption.

Source code in src/video_converter/processors/codec_detector.py

def __init__(self, path: Path, reason: str = "Video file is corrupted") -> None:
    self.path = path
    self.reason = reason
    super().__init__(f"{path}: {reason}")

UnsupportedCodecError ¶

UnsupportedCodecError(path, codec)

Bases: Exception

Raised when an unknown or unsupported codec is encountered.

This error indicates that the video uses a codec that is not recognized or supported by the converter.

Attributes:

Name	Type	Description
`path`		Path to the video file.
`codec`		The unsupported codec name.

Source code in src/video_converter/processors/codec_detector.py

def __init__(self, path: Path, codec: str) -> None:
    self.path = path
    self.codec = codec
    super().__init__(f"{path}: Unsupported codec '{codec}'")

CodecInfo `dataclass` ¶

CodecInfo(path, codec, width, height, fps, duration, bitrate, size, audio_codec, container, creation_time=None, color_space=None, bit_depth=None, profile=None, level=None)

Video codec and property information.

This class contains detailed information about a video file's codec and properties, used to determine if conversion is needed.

Attributes:

Name	Type	Description
`path`	`Path`	Path to the video file.
`codec`	`str`	Video codec name (e.g., "h264", "hevc", "vp9").
`width`	`int`	Video width in pixels.
`height`	`int`	Video height in pixels.
`fps`	`float`	Frames per second.
`duration`	`float`	Duration in seconds.
`bitrate`	`int`	Video bitrate in bits per second.
`size`	`int`	File size in bytes.
`audio_codec`	`str \| None`	Audio codec name (e.g., "aac", "opus").
`container`	`str`	Container format (e.g., "mp4", "mov", "mkv").
`creation_time`	`datetime \| None`	Recording timestamp (if available).
`color_space`	`str \| None`	Video color space (if available).
`bit_depth`	`int \| None`	Video bit depth (if available).
`profile`	`str \| None`	Codec profile (e.g., "High", "Main").
`level`	`str \| None`	Codec level (e.g., "4.0", "5.1").

Example

info = CodecInfo( ... path=Path("video.mp4"), ... codec="h264", ... width=3840, ... height=2160, ... fps=29.97, ... duration=180.5, ... bitrate=50_000_000, ... size=1_125_000_000, ... audio_codec="aac", ... container="mp4" ... ) print(f"Resolution: {info.resolution_label}") # "4K" print(f"Needs conversion: {info.is_h264}") # True

is_h264 `property` ¶

is_h264

Check if video codec is H.264/AVC.

Returns:

Type	Description
`bool`	True if the codec is any variant of H.264.

is_hevc `property` ¶

is_hevc

Check if video codec is H.265/HEVC.

Returns:

Type	Description
`bool`	True if the codec is any variant of H.265/HEVC.

needs_conversion `property` ¶

needs_conversion

Check if video needs H.265 conversion.

Returns:

Type	Description
`bool`	True if the video is H.264 and would benefit from conversion.

resolution_label `property` ¶

resolution_label

Get human-readable resolution label.

Returns:

Type	Description
`str`	Resolution label like "4K", "1080p", "720p", etc.

aspect_ratio `property` ¶

aspect_ratio

Calculate aspect ratio.

Returns:

Type	Description
`str`	Aspect ratio string like "16:9", "4:3", etc.

bitrate_mbps `property` ¶

bitrate_mbps

Get bitrate in Mbps.

Returns:

Type	Description
`float`	Bitrate in megabits per second.

size_mb `property` ¶

size_mb

Get file size in MB.

Returns:

Type	Description
`float`	File size in megabytes.

size_gb `property` ¶

size_gb

Get file size in GB.

Returns:

Type	Description
`float`	File size in gigabytes.

str ¶

__str__()

Return human-readable summary.

Source code in src/video_converter/processors/codec_detector.py

def __str__(self) -> str:
    """Return human-readable summary."""
    return (
        f"{self.path.name}: {self.codec.upper()} {self.resolution_label} "
        f"@ {self.fps:.2f}fps, {self.duration:.1f}s, {self.size_mb:.1f}MB"
    )

CodecDetector ¶

CodecDetector(ffprobe_runner=None, timeout=30.0)

Video codec detector using FFprobe.

This class analyzes video files to extract codec information and determine if conversion to H.265/HEVC is needed.

SDS Reference: SDS-P01-002

Example

detector = CodecDetector() info = detector.analyze(Path("vacation.mp4")) if info.needs_conversion: ... print(f"Converting {info.path.name} from {info.codec} to HEVC") ...

Batch analysis¶

videos = list(Path("videos").glob("*.mp4")) h264_videos = [v for v in videos if detector.analyze(v).is_h264]

Attributes:

Name	Type	Description
`ffprobe`		FFprobe runner instance.
`timeout`		Default timeout for FFprobe commands.

Initialize CodecDetector.

Parameters:

Name	Type	Description	Default
`ffprobe_runner`	`FFprobeRunner \| None`	FFprobe runner to use. If None, creates a new one.	`None`
`timeout`	`float`	Default timeout for FFprobe commands in seconds.	`30.0`

Source code in src/video_converter/processors/codec_detector.py

def __init__(
    self,
    ffprobe_runner: FFprobeRunner | None = None,
    timeout: float = 30.0,
) -> None:
    """Initialize CodecDetector.

    Args:
        ffprobe_runner: FFprobe runner to use. If None, creates a new one.
        timeout: Default timeout for FFprobe commands in seconds.
    """
    self._ffprobe = ffprobe_runner or FFprobeRunner()
    self._timeout = timeout

analyze ¶

analyze(path)

Analyze a video file and return codec information.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required

Returns:

Type	Description
`CodecInfo`	CodecInfo containing video properties.

Raises:

Type	Description
`FileNotFoundError`	If the file doesn't exist.
`InvalidVideoError`	If the file is not a valid video.
`CorruptedVideoError`	If the video file is corrupted.
`CommandNotFoundError`	If FFprobe is not installed.

Source code in src/video_converter/processors/codec_detector.py

def analyze(self, path: Path) -> CodecInfo:
    """Analyze a video file and return codec information.

    Args:
        path: Path to the video file.

    Returns:
        CodecInfo containing video properties.

    Raises:
        FileNotFoundError: If the file doesn't exist.
        InvalidVideoError: If the file is not a valid video.
        CorruptedVideoError: If the video file is corrupted.
        CommandNotFoundError: If FFprobe is not installed.
    """
    if not path.exists():
        raise FileNotFoundError(f"Video file not found: {path}")

    try:
        probe_data = self._ffprobe.probe(
            path,
            show_format=True,
            show_streams=True,
            timeout=self._timeout,
        )
    except CommandExecutionError as e:
        raise InvalidVideoError(path, f"FFprobe failed: {e.stderr}") from e
    except CommandNotFoundError:
        raise

    return self._parse_probe_data(path, probe_data)

analyze_async `async` ¶

analyze_async(path)

Analyze a video file asynchronously.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required

Returns:

Type	Description
`CodecInfo`	CodecInfo containing video properties.

Raises:

Type	Description
`FileNotFoundError`	If the file doesn't exist.
`InvalidVideoError`	If the file is not a valid video.
`CorruptedVideoError`	If the video file is corrupted.
`CommandNotFoundError`	If FFprobe is not installed.

Source code in src/video_converter/processors/codec_detector.py

async def analyze_async(self, path: Path) -> CodecInfo:
    """Analyze a video file asynchronously.

    Args:
        path: Path to the video file.

    Returns:
        CodecInfo containing video properties.

    Raises:
        FileNotFoundError: If the file doesn't exist.
        InvalidVideoError: If the file is not a valid video.
        CorruptedVideoError: If the video file is corrupted.
        CommandNotFoundError: If FFprobe is not installed.
    """
    if not path.exists():
        raise FileNotFoundError(f"Video file not found: {path}")

    try:
        probe_data = await self._ffprobe.probe_async(
            path,
            show_format=True,
            show_streams=True,
            timeout=self._timeout,
        )
    except CommandExecutionError as e:
        raise InvalidVideoError(path, f"FFprobe failed: {e.stderr}") from e
    except CommandNotFoundError:
        raise

    return self._parse_probe_data(path, probe_data)

is_h264 ¶

is_h264(path)

Quick check if a video is H.264 encoded.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required

Returns:

Type	Description
`bool`	True if the video is H.264, False otherwise.

Source code in src/video_converter/processors/codec_detector.py

def is_h264(self, path: Path) -> bool:
    """Quick check if a video is H.264 encoded.

    Args:
        path: Path to the video file.

    Returns:
        True if the video is H.264, False otherwise.
    """
    try:
        info = self.analyze(path)
        return info.is_h264
    except (InvalidVideoError, CorruptedVideoError, FileNotFoundError):
        return False

is_hevc ¶

is_hevc(path)

Quick check if a video is H.265/HEVC encoded.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required

Returns:

Type	Description
`bool`	True if the video is HEVC, False otherwise.

Source code in src/video_converter/processors/codec_detector.py

def is_hevc(self, path: Path) -> bool:
    """Quick check if a video is H.265/HEVC encoded.

    Args:
        path: Path to the video file.

    Returns:
        True if the video is HEVC, False otherwise.
    """
    try:
        info = self.analyze(path)
        return info.is_hevc
    except (InvalidVideoError, CorruptedVideoError, FileNotFoundError):
        return False

needs_conversion ¶

needs_conversion(path)

Check if a video needs conversion to H.265.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required

Returns:

Type	Description
`bool`	True if the video is H.264 and should be converted.

Source code in src/video_converter/processors/codec_detector.py

def needs_conversion(self, path: Path) -> bool:
    """Check if a video needs conversion to H.265.

    Args:
        path: Path to the video file.

    Returns:
        True if the video is H.264 and should be converted.
    """
    return self.is_h264(path)

get_codec ¶

get_codec(path)

Get the video codec name.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required

Returns:

Type	Description
`str`	Codec name (e.g., "h264", "hevc").

Raises:

Type	Description
`InvalidVideoError`	If the file is not a valid video.

Source code in src/video_converter/processors/codec_detector.py

def get_codec(self, path: Path) -> str:
    """Get the video codec name.

    Args:
        path: Path to the video file.

    Returns:
        Codec name (e.g., "h264", "hevc").

    Raises:
        InvalidVideoError: If the file is not a valid video.
    """
    return self.analyze(path).codec

Metadata Processor¶

metadata ¶

Metadata processing module for video files.

This module provides ExifTool integration for extracting and applying metadata on video files. It handles GPS coordinates, creation dates, camera information, and other critical metadata tags.

SDS Reference: SDS-P01-002 SRS Reference: SRS-401 (Metadata Preservation)

Example

from video_converter.utils.command_runner import CommandRunner runner = CommandRunner() processor = MetadataProcessor(runner) metadata = processor.extract(Path("video.mp4")) print(metadata.get("QuickTime:CreateDate"))

Copy all metadata¶

processor.copy_all(source_path, dest_path)

Verify critical tags¶

results = processor.verify_critical_tags(original, converted) for tag, matched in results.items(): ... print(f"{tag}: {'OK' if matched else 'MISMATCH'}")

MetadataExtractionError ¶

MetadataExtractionError(path, reason)

Bases: Exception

Raised when metadata extraction fails.

Source code in src/video_converter/processors/metadata.py

def __init__(self, path: Path, reason: str) -> None:
    self.path = path
    self.reason = reason
    super().__init__(f"Failed to extract metadata from {path}: {reason}")

MetadataApplicationError ¶

MetadataApplicationError(path, reason)

Bases: Exception

Raised when metadata application fails.

Source code in src/video_converter/processors/metadata.py

def __init__(self, path: Path, reason: str) -> None:
    self.path = path
    self.reason = reason
    super().__init__(f"Failed to apply metadata to {path}: {reason}")

GPSCoordinates `dataclass` ¶

GPSCoordinates(latitude, longitude, altitude=None)

GPS coordinate data.

Attributes:

Name	Type	Description
`latitude`	`float`	Latitude in decimal degrees.
`longitude`	`float`	Longitude in decimal degrees.
`altitude`	`float \| None`	Altitude in meters (optional).

str ¶

__str__()

Return human-readable coordinate string.

Source code in src/video_converter/processors/metadata.py

def __str__(self) -> str:
    """Return human-readable coordinate string."""
    lat_dir = "N" if self.latitude >= 0 else "S"
    lon_dir = "E" if self.longitude >= 0 else "W"
    result = f"{abs(self.latitude):.6f}°{lat_dir}, {abs(self.longitude):.6f}°{lon_dir}"
    if self.altitude is not None:
        result += f", {self.altitude:.1f}m"
    return result

MetadataVerificationResult `dataclass` ¶

MetadataVerificationResult(all_matched, tag_results=dict(), missing_in_source=list(), missing_in_dest=list())

Result of metadata verification between two files.

Attributes:

Name	Type	Description
`all_matched`	`bool`	True if all critical tags matched.
`tag_results`	`dict[str, bool]`	Dictionary mapping tag names to match status.
`missing_in_source`	`list[str]`	Tags present in source but missing in destination.
`missing_in_dest`	`list[str]`	Tags missing in destination that were in source.

MetadataProcessor ¶

MetadataProcessor(command_runner=None)

Handle video metadata extraction and application using ExifTool.

This class provides methods to extract metadata from video files, copy metadata between files, and verify that critical metadata was preserved during conversion.

Attributes:

Name	Type	Description
`EXIFTOOL_CMD`		The exiftool command name.
`CRITICAL_TAGS`		List of metadata tags that must be preserved.
`GPS_TAGS`		Tags related to GPS/location data.
`DATE_TAGS`		Tags related to dates and times.

Example

runner = CommandRunner() processor = MetadataProcessor(runner) metadata = processor.extract(Path("video.mp4")) gps = processor.extract_gps(Path("video.mp4")) if gps: ... print(f"Location: {gps}")

Initialize MetadataProcessor.

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/processors/metadata.py

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

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

is_available ¶

is_available()

Check if ExifTool is available.

Returns:

Type	Description
`bool`	True if exiftool command is available.

Source code in src/video_converter/processors/metadata.py

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

    Returns:
        True if exiftool command is available.
    """
    return self._runner.check_command_exists(self.EXIFTOOL_CMD)

ensure_available ¶

ensure_available()

Ensure ExifTool is available, raising error if not.

Raises:

Type	Description
`CommandNotFoundError`	If exiftool is not installed.

Source code in src/video_converter/processors/metadata.py

def ensure_available(self) -> None:
    """Ensure ExifTool is available, raising error if not.

    Raises:
        CommandNotFoundError: If exiftool is not installed.
    """
    self._runner.ensure_command_exists(self.EXIFTOOL_CMD)

extract ¶

extract(path, *, include_binary=False)

Extract all metadata from a video file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required
`include_binary`	`bool`	Include binary data in output (default: False).	`False`

Returns:

Type	Description
`dict[str, Any]`	Dictionary of metadata tags and values.

Raises:

Type	Description
`FileNotFoundError`	If the file doesn't exist.
`MetadataExtractionError`	If extraction fails.
`CommandNotFoundError`	If exiftool is not installed.

Source code in src/video_converter/processors/metadata.py

def extract(self, path: Path, *, include_binary: bool = False) -> dict[str, Any]:
    """Extract all metadata from a video file.

    Args:
        path: Path to the video file.
        include_binary: Include binary data in output (default: False).

    Returns:
        Dictionary of metadata tags and values.

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

    args = [
        self.EXIFTOOL_CMD,
        "-json",
        "-G",  # Show group names
        "-a",  # Allow duplicate tags
        "-u",  # Show unknown tags
        "-n",  # Numeric output (no formatting)
    ]

    if not include_binary:
        args.append("-b")
        # Actually -b includes binary, we want to exclude
        # Use -X for XMP or just don't include -b
        args.remove("-b")

    args.append(str(path))

    try:
        result = self._runner.run(args, timeout=30.0)
        if not result.success:
            raise MetadataExtractionError(path, result.stderr)

        data: list[dict[str, Any]] = json.loads(result.stdout)
        return data[0] if data else {}

    except json.JSONDecodeError as e:
        raise MetadataExtractionError(path, f"Invalid JSON output: {e}") from e
    except CommandExecutionError as e:
        raise MetadataExtractionError(path, str(e)) from e

extract_gps ¶

extract_gps(path)

Extract GPS coordinates from a video file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required

Returns:

Type	Description
`GPSCoordinates \| None`	GPSCoordinates if found, None otherwise.

Raises:

Type	Description
`FileNotFoundError`	If the file doesn't exist.
`MetadataExtractionError`	If extraction fails.

Source code in src/video_converter/processors/metadata.py

def extract_gps(self, path: Path) -> GPSCoordinates | None:
    """Extract GPS coordinates from a video file.

    Args:
        path: Path to the video file.

    Returns:
        GPSCoordinates if found, None otherwise.

    Raises:
        FileNotFoundError: If the file doesn't exist.
        MetadataExtractionError: If extraction fails.
    """
    metadata = self.extract(path)

    # Try different GPS tag formats
    lat = self._find_tag_value(metadata, "GPSLatitude")
    lon = self._find_tag_value(metadata, "GPSLongitude")

    if lat is None or lon is None:
        # Try Composite:GPSPosition format
        gps_pos = self._find_tag_value(metadata, "GPSPosition")
        if gps_pos and isinstance(gps_pos, str):
            coords = self._parse_gps_position(gps_pos)
            if coords:
                lat, lon = coords

    if lat is not None and lon is not None:
        try:
            lat_float = self._convert_gps_to_decimal(lat, metadata, "Latitude")
            lon_float = self._convert_gps_to_decimal(lon, metadata, "Longitude")

            alt = self._find_tag_value(metadata, "GPSAltitude")
            alt_float = float(alt) if alt is not None else None

            return GPSCoordinates(
                latitude=lat_float,
                longitude=lon_float,
                altitude=alt_float,
            )
        except (ValueError, TypeError):
            return None

    return None

copy_all ¶

copy_all(source, dest, *, overwrite_original=True)

Copy all metadata from source to destination file.

Parameters:

Name	Type	Description	Default
`source`	`Path`	Source file with metadata.	required
`dest`	`Path`	Destination file to receive metadata.	required
`overwrite_original`	`bool`	If True, modify dest in place without backup.	`True`

Returns:

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

Raises:

Type	Description
`FileNotFoundError`	If source or dest doesn't exist.
`MetadataApplicationError`	If copying fails.

Source code in src/video_converter/processors/metadata.py

def copy_all(
    self,
    source: Path,
    dest: Path,
    *,
    overwrite_original: bool = True,
) -> bool:
    """Copy all metadata from source to destination file.

    Args:
        source: Source file with metadata.
        dest: Destination file to receive metadata.
        overwrite_original: If True, modify dest in place without backup.

    Returns:
        True if metadata was copied successfully.

    Raises:
        FileNotFoundError: If source or dest doesn't exist.
        MetadataApplicationError: If copying fails.
    """
    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")

    args.extend(
        [
            "-tagsFromFile",
            str(source),
            "-all:all",
            str(dest),
        ]
    )

    try:
        result = self._runner.run(args, timeout=60.0)
        return result.success
    except CommandExecutionError as e:
        raise MetadataApplicationError(dest, str(e)) from e

copy_tags ¶

copy_tags(source, dest, tags, *, overwrite_original=True)

Copy specific metadata tags from source to destination.

Parameters:

Name	Type	Description	Default
`source`	`Path`	Source file with metadata.	required
`dest`	`Path`	Destination file to receive metadata.	required
`tags`	`list[str]`	List of tag patterns (e.g., ["GPS*", "CreateDate"]).	required
`overwrite_original`	`bool`	If True, modify dest in place.	`True`

Returns:

Type	Description
`bool`	True if tags were copied successfully.

Raises:

Type	Description
`FileNotFoundError`	If source or dest doesn't exist.
`MetadataApplicationError`	If copying fails.

Source code in src/video_converter/processors/metadata.py

def copy_tags(
    self,
    source: Path,
    dest: Path,
    tags: list[str],
    *,
    overwrite_original: bool = True,
) -> bool:
    """Copy specific metadata tags from source to destination.

    Args:
        source: Source file with metadata.
        dest: Destination file to receive metadata.
        tags: List of tag patterns (e.g., ["GPS*", "CreateDate"]).
        overwrite_original: If True, modify dest in place.

    Returns:
        True if tags were copied successfully.

    Raises:
        FileNotFoundError: If source or dest doesn't exist.
        MetadataApplicationError: If copying fails.
    """
    if not source.exists():
        raise FileNotFoundError(f"Source file not found: {source}")
    if not dest.exists():
        raise FileNotFoundError(f"Destination file not found: {dest}")
    if not tags:
        return True  # Nothing to copy

    args = [self.EXIFTOOL_CMD]

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

    args.extend(["-tagsFromFile", str(source)])

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

    args.append(str(dest))

    try:
        result = self._runner.run(args, timeout=60.0)
        return result.success
    except CommandExecutionError as e:
        raise MetadataApplicationError(dest, str(e)) from e

copy_gps ¶

copy_gps(source, dest, *, overwrite_original=True)

Copy GPS metadata from source to destination.

This is a convenience method that copies all GPS-related tags.

Parameters:

Name	Type	Description	Default
`source`	`Path`	Source file with GPS data.	required
`dest`	`Path`	Destination file.	required
`overwrite_original`	`bool`	If True, modify dest in place.	`True`

Returns:

Type	Description
`bool`	True if GPS data was copied successfully.

Source code in src/video_converter/processors/metadata.py

def copy_gps(
    self,
    source: Path,
    dest: Path,
    *,
    overwrite_original: bool = True,
) -> bool:
    """Copy GPS metadata from source to destination.

    This is a convenience method that copies all GPS-related tags.

    Args:
        source: Source file with GPS data.
        dest: Destination file.
        overwrite_original: If True, modify dest in place.

    Returns:
        True if GPS data was copied successfully.
    """
    return self.copy_tags(
        source,
        dest,
        ["GPS*", "GPSCoordinates", "GPSPosition"],
        overwrite_original=overwrite_original,
    )

copy_dates ¶

copy_dates(source, dest, *, overwrite_original=True)

Copy date/time metadata from source to destination.

Parameters:

Name	Type	Description	Default
`source`	`Path`	Source file with date metadata.	required
`dest`	`Path`	Destination file.	required
`overwrite_original`	`bool`	If True, modify dest in place.	`True`

Returns:

Type	Description
`bool`	True if dates were copied successfully.

Source code in src/video_converter/processors/metadata.py

def copy_dates(
    self,
    source: Path,
    dest: Path,
    *,
    overwrite_original: bool = True,
) -> bool:
    """Copy date/time metadata from source to destination.

    Args:
        source: Source file with date metadata.
        dest: Destination file.
        overwrite_original: If True, modify dest in place.

    Returns:
        True if dates were copied successfully.
    """
    return self.copy_tags(
        source,
        dest,
        self.DATE_TAGS,
        overwrite_original=overwrite_original,
    )

verify_critical_tags ¶

verify_critical_tags(original, converted)

Verify that critical metadata was preserved during conversion.

Parameters:

Name	Type	Description	Default
`original`	`Path`	Original source file.	required
`converted`	`Path`	Converted output file.	required

Returns:

Type	Description
`MetadataVerificationResult`	MetadataVerificationResult with comparison details.

Raises:

Type	Description
`FileNotFoundError`	If either file doesn't exist.
`MetadataExtractionError`	If metadata extraction fails.

Source code in src/video_converter/processors/metadata.py

def verify_critical_tags(
    self,
    original: Path,
    converted: Path,
) -> MetadataVerificationResult:
    """Verify that critical metadata was preserved during conversion.

    Args:
        original: Original source file.
        converted: Converted output file.

    Returns:
        MetadataVerificationResult with comparison details.

    Raises:
        FileNotFoundError: If either file doesn't exist.
        MetadataExtractionError: If metadata extraction fails.
    """
    orig_meta = self.extract(original)
    conv_meta = self.extract(converted)

    tag_results: dict[str, bool] = {}
    missing_in_source: list[str] = []
    missing_in_dest: list[str] = []

    for tag in self.CRITICAL_TAGS:
        orig_value = self._find_tag_value(orig_meta, tag)
        conv_value = self._find_tag_value(conv_meta, tag)

        if orig_value is None:
            missing_in_source.append(tag)
            continue

        if conv_value is None:
            missing_in_dest.append(tag)
            tag_results[tag] = False
        else:
            tag_results[tag] = self._values_match(orig_value, conv_value)

    all_matched = all(tag_results.values()) if tag_results else True

    return MetadataVerificationResult(
        all_matched=all_matched,
        tag_results=tag_results,
        missing_in_source=missing_in_source,
        missing_in_dest=missing_in_dest,
    )

set_tag ¶

set_tag(path, tag, value, *, overwrite_original=True)

Set a single metadata tag on a file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the file.	required
`tag`	`str`	Tag name to set.	required
`value`	`str`	Value to set.	required
`overwrite_original`	`bool`	If True, modify file in place.	`True`

Returns:

Type	Description
`bool`	True if the tag was set successfully.

Raises:

Type	Description
`FileNotFoundError`	If the file doesn't exist.
`MetadataApplicationError`	If setting fails.

Source code in src/video_converter/processors/metadata.py

def set_tag(
    self,
    path: Path,
    tag: str,
    value: str,
    *,
    overwrite_original: bool = True,
) -> bool:
    """Set a single metadata tag on a file.

    Args:
        path: Path to the file.
        tag: Tag name to set.
        value: Value to set.
        overwrite_original: If True, modify file in place.

    Returns:
        True if the tag was set successfully.

    Raises:
        FileNotFoundError: If the file doesn't exist.
        MetadataApplicationError: If setting fails.
    """
    if not path.exists():
        raise FileNotFoundError(f"File not found: {path}")

    args = [self.EXIFTOOL_CMD]

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

    # Escape special characters in value
    escaped_value = self._escape_value(value)
    args.append(f"-{tag}={escaped_value}")
    args.append(str(path))

    try:
        result = self._runner.run(args, timeout=30.0)
        return result.success
    except CommandExecutionError as e:
        raise MetadataApplicationError(path, str(e)) from e

batch_copy_all ¶

batch_copy_all(source_dest_pairs, *, overwrite_original=True)

Copy metadata for multiple file pairs.

Parameters:

Name	Type	Description	Default
`source_dest_pairs`	`list[tuple[Path, Path]]`	List of (source, dest) path tuples.	required
`overwrite_original`	`bool`	If True, modify dest files in place.	`True`

Returns:

Type	Description
`dict[Path, bool]`	Dictionary mapping destination paths to success status.

Source code in src/video_converter/processors/metadata.py

def batch_copy_all(
    self,
    source_dest_pairs: list[tuple[Path, Path]],
    *,
    overwrite_original: bool = True,
) -> dict[Path, bool]:
    """Copy metadata for multiple file pairs.

    Args:
        source_dest_pairs: List of (source, dest) path tuples.
        overwrite_original: If True, modify dest files in place.

    Returns:
        Dictionary mapping destination paths to success status.
    """
    results: dict[Path, bool] = {}

    for source, dest in source_dest_pairs:
        try:
            success = self.copy_all(source, dest, overwrite_original=overwrite_original)
            results[dest] = success
        except (FileNotFoundError, MetadataApplicationError):
            results[dest] = False

    return results

Quality Validator¶

quality_validator ¶

Video quality validation module.

This module provides video file integrity validation using FFprobe to ensure converted files are playable and not corrupted. It also includes compression ratio validation to detect encoding issues.

SDS Reference: SDS-P01-003 SRS Reference: SRS-501 (Conversion Result Verification) SRS Reference: SRS-503 (Compression Ratio Validation)

Example

validator = VideoValidator() result = validator.validate(Path("output.mp4")) if result.valid: ... print("Video is valid!") ... else: ... print(f"Errors: {result.errors}")

Compression ratio validation¶

comp_validator = CompressionValidator() result = comp_validator.validate(100_000_000, 40_000_000) print(f"Compression ratio: {result.compression_ratio:.1%}")

ValidationStrictness ¶

Bases: Enum

Validation strictness levels.

Attributes:

Name	Type	Description
`QUICK`		Only check if file is readable (fastest).
`STANDARD`		Check integrity and basic properties (default).
`STRICT`		Full validation including all streams and metadata.

StreamInfo `dataclass` ¶

StreamInfo(index, codec_type, codec_name, duration=None, bit_rate=None, width=None, height=None, fps=None, channels=None, sample_rate=None)

Information about a media stream.

Attributes:

Name	Type	Description
`index`	`int`	Stream index in the file.
`codec_type`	`str`	Type of stream (video, audio, subtitle, etc.).
`codec_name`	`str`	Name of the codec (h264, hevc, aac, etc.).
`duration`	`float \| None`	Duration in seconds (if available).
`bit_rate`	`int \| None`	Bit rate in bits/second (if available).
`width`	`int \| None`	Width in pixels (video only).
`height`	`int \| None`	Height in pixels (video only).
`fps`	`float \| None`	Frames per second (video only).
`channels`	`int \| None`	Number of audio channels (audio only).
`sample_rate`	`int \| None`	Audio sample rate (audio only).

VideoInfo `dataclass` ¶

VideoInfo(path, format_name, duration, size, bit_rate, streams=list())

Complete video file information.

Attributes:

Name	Type	Description
`path`	`Path`	Path to the video file.
`format_name`	`str`	Container format name.
`duration`	`float`	Total duration in seconds.
`size`	`int`	File size in bytes.
`bit_rate`	`int \| None`	Overall bit rate.
`streams`	`list[StreamInfo]`	List of stream information.
`video_streams`	`list[StreamInfo]`	List of video stream information.
`audio_streams`	`list[StreamInfo]`	List of audio stream information.

video_streams `property` ¶

video_streams

Get all video streams.

audio_streams `property` ¶

audio_streams

Get all audio streams.

has_video `property` ¶

has_video

Check if file has at least one video stream.

has_audio `property` ¶

has_audio

Check if file has at least one audio stream.

primary_video_stream `property` ¶

primary_video_stream

Get the primary (first) video stream.

primary_audio_stream `property` ¶

primary_audio_stream

Get the primary (first) audio stream.

ValidationResult `dataclass` ¶

ValidationResult(valid, integrity_ok, properties_match=True, compression_normal=True, vmaf_score=None, errors=list(), warnings=list(), video_info=None)

Result of video validation.

Attributes:

Name	Type	Description
`valid`	`bool`	Overall validation result (True if all checks passed).
`integrity_ok`	`bool`	File integrity check passed.
`properties_match`	`bool`	Properties match expected values.
`compression_normal`	`bool`	Compression ratio is within normal range.
`vmaf_score`	`float \| None`	VMAF quality score (if measured, None otherwise).
`errors`	`list[str]`	List of error messages (validation failures).
`warnings`	`list[str]`	List of warning messages (non-critical issues).
`video_info`	`VideoInfo \| None`	Parsed video information (if successful).

add_error ¶

add_error(message)

Add an error message and mark as invalid.

Source code in src/video_converter/processors/quality_validator.py

def add_error(self, message: str) -> None:
    """Add an error message and mark as invalid."""
    self.errors.append(message)
    self.valid = False

add_warning ¶

add_warning(message)

Add a warning message.

Source code in src/video_converter/processors/quality_validator.py

def add_warning(self, message: str) -> None:
    """Add a warning message."""
    self.warnings.append(message)

VideoValidator ¶

VideoValidator(*, strictness=STANDARD, timeout=30.0, ffprobe=None)

Video file integrity validator using FFprobe.

This class provides comprehensive validation for video files to ensure they are playable and not corrupted. It supports configurable strictness levels for different use cases.

SDS Reference: SDS-P01-003

Validation Checks

File existence and size > 0
FFprobe can read the file without errors
Video stream exists and is decodable
Audio stream integrity (if present)
Duration is valid (> 0)
Codec information is extractable

Example

validator = VideoValidator()

Quick validation¶

if validator.quick_validate(Path("video.mp4")): ... print("Video is playable")

Full validation¶

result = validator.validate(Path("video.mp4")) print(f"Valid: {result.valid}, Errors: {result.errors}")

Attributes:

Name	Type	Description
`ffprobe`		FFprobe runner instance.
`strictness`		Validation strictness level.
`timeout`		Maximum time for FFprobe commands.

Initialize the video validator.

Parameters:

Name	Type	Description	Default
`strictness`	`ValidationStrictness`	Validation strictness level.	`STANDARD`
`timeout`	`float`	Maximum time for FFprobe operations (seconds).	`30.0`
`ffprobe`	`FFprobeRunner \| None`	FFprobe runner to use (creates new one if None).	`None`

Source code in src/video_converter/processors/quality_validator.py

def __init__(
    self,
    *,
    strictness: ValidationStrictness = ValidationStrictness.STANDARD,
    timeout: float = 30.0,
    ffprobe: FFprobeRunner | None = None,
) -> None:
    """Initialize the video validator.

    Args:
        strictness: Validation strictness level.
        timeout: Maximum time for FFprobe operations (seconds).
        ffprobe: FFprobe runner to use (creates new one if None).
    """
    self.strictness = strictness
    self.timeout = timeout
    self.ffprobe = ffprobe or FFprobeRunner()

quick_validate ¶

quick_validate(path)

Quickly check if a video file is valid.

This performs a minimal check to see if the file can be opened. Use this for fast validation when detailed error information is not needed.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required

Returns:

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

Source code in src/video_converter/processors/quality_validator.py

def quick_validate(self, path: Path) -> bool:
    """Quickly check if a video file is valid.

    This performs a minimal check to see if the file can be opened.
    Use this for fast validation when detailed error information
    is not needed.

    Args:
        path: Path to the video file.

    Returns:
        True if the file appears valid, False otherwise.
    """
    return self.ffprobe.quick_check(path, timeout=self.timeout)

validate ¶

validate(path, *, strictness=None)

Validate a video file comprehensively.

Performs full validation according to the specified strictness level. Returns detailed information about any issues found.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file to validate.	required
`strictness`	`ValidationStrictness \| None`	Override the default strictness level.	`None`

Returns:

Type	Description
`ValidationResult`	ValidationResult containing validation details.

Raises:

Type	Description
`CommandNotFoundError`	If FFprobe is not installed.

Source code in src/video_converter/processors/quality_validator.py

def validate(
    self,
    path: Path,
    *,
    strictness: ValidationStrictness | None = None,
) -> ValidationResult:
    """Validate a video file comprehensively.

    Performs full validation according to the specified strictness level.
    Returns detailed information about any issues found.

    Args:
        path: Path to the video file to validate.
        strictness: Override the default strictness level.

    Returns:
        ValidationResult containing validation details.

    Raises:
        CommandNotFoundError: If FFprobe is not installed.
    """
    level = strictness or self.strictness
    result = ValidationResult(valid=True, integrity_ok=False)

    # Step 1: Check file exists and has size > 0
    if not self._check_file_exists(path, result):
        return result

    # Step 2: Probe the file with FFprobe
    try:
        probe_data = self.ffprobe.probe(
            path,
            show_format=True,
            show_streams=True,
            timeout=self.timeout,
        )
    except CommandNotFoundError:
        raise
    except CommandExecutionError as e:
        result.add_error(f"FFprobe failed: {e.stderr}")
        return result
    except FileNotFoundError:
        result.add_error(f"File not found: {path}")
        return result
    except Exception as e:
        result.add_error(f"Error probing file: {e}")
        return result

    # Step 3: Parse and validate the probe data
    video_info = self._parse_probe_data(path, probe_data, result)
    if video_info is None:
        return result

    result.video_info = video_info
    result.integrity_ok = True

    # Step 4: Validate video streams
    self._validate_video_streams(video_info, result, level)

    # Step 5: Validate audio streams (if strict mode)
    if level in (ValidationStrictness.STANDARD, ValidationStrictness.STRICT):
        self._validate_audio_streams(video_info, result, level)

    # Step 6: Validate duration
    self._validate_duration(video_info, result)

    # Step 7: Additional strict checks
    if level == ValidationStrictness.STRICT:
        self._validate_strict(video_info, result)

    return result

validate_async `async` ¶

validate_async(path, *, strictness=None)

Validate a video file asynchronously.

Async version of validate() for use in async contexts.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file to validate.	required
`strictness`	`ValidationStrictness \| None`	Override the default strictness level.	`None`

Returns:

Type	Description
`ValidationResult`	ValidationResult containing validation details.

Source code in src/video_converter/processors/quality_validator.py

async def validate_async(
    self,
    path: Path,
    *,
    strictness: ValidationStrictness | None = None,
) -> ValidationResult:
    """Validate a video file asynchronously.

    Async version of validate() for use in async contexts.

    Args:
        path: Path to the video file to validate.
        strictness: Override the default strictness level.

    Returns:
        ValidationResult containing validation details.
    """
    level = strictness or self.strictness
    result = ValidationResult(valid=True, integrity_ok=False)

    if not self._check_file_exists(path, result):
        return result

    try:
        probe_data = await self.ffprobe.probe_async(
            path,
            show_format=True,
            show_streams=True,
            timeout=self.timeout,
        )
    except CommandNotFoundError:
        raise
    except CommandExecutionError as e:
        result.add_error(f"FFprobe failed: {e.stderr}")
        return result
    except FileNotFoundError:
        result.add_error(f"File not found: {path}")
        return result
    except Exception as e:
        result.add_error(f"Error probing file: {e}")
        return result

    video_info = self._parse_probe_data(path, probe_data, result)
    if video_info is None:
        return result

    result.video_info = video_info
    result.integrity_ok = True

    self._validate_video_streams(video_info, result, level)

    if level in (ValidationStrictness.STANDARD, ValidationStrictness.STRICT):
        self._validate_audio_streams(video_info, result, level)

    self._validate_duration(video_info, result)

    if level == ValidationStrictness.STRICT:
        self._validate_strict(video_info, result)

    return result

ComparisonSeverity ¶

Bases: Enum

Severity level for property mismatch.

Attributes:

Name	Type	Description
`ERROR`		Mismatch indicates a critical problem.
`WARNING`		Mismatch is notable but not critical.

PropertyComparison `dataclass` ¶

PropertyComparison(property_name, original_value, converted_value, matches, tolerance=None, severity=WARNING, message='')

Result of comparing a single video property.

SDS Reference: SDS-P05-002 SRS Reference: SRS-502 (Video Property Comparison)

Attributes:

Name	Type	Description
`property_name`	`str`	Name of the property being compared.
`original_value`	`Any`	Value from the original file.
`converted_value`	`Any`	Value from the converted file.
`matches`	`bool`	Whether the values match within tolerance.
`tolerance`	`float \| None`	Tolerance used for comparison (if applicable).
`severity`	`ComparisonSeverity`	Severity level if mismatch occurs.
`message`	`str`	Human-readable description of the comparison result.

PropertyComparisonResult `dataclass` ¶

PropertyComparisonResult(all_match=True, comparisons=list(), errors=list(), warnings=list())

Result of comparing all video properties.

SDS Reference: SDS-P05-002

Attributes:

Name	Type	Description
`all_match`	`bool`	True if all critical properties match.
`comparisons`	`list[PropertyComparison]`	List of individual property comparisons.
`errors`	`list[str]`	List of error messages (critical mismatches).
`warnings`	`list[str]`	List of warning messages (non-critical mismatches).

add_comparison ¶

add_comparison(comparison)

Add a property comparison result.

Parameters:

Name	Type	Description	Default
`comparison`	`PropertyComparison`	The comparison result to add.	required

Source code in src/video_converter/processors/quality_validator.py

def add_comparison(self, comparison: PropertyComparison) -> None:
    """Add a property comparison result.

    Args:
        comparison: The comparison result to add.
    """
    self.comparisons.append(comparison)

    if not comparison.matches:
        if comparison.severity == ComparisonSeverity.ERROR:
            self.all_match = False
            self.errors.append(comparison.message or f"{comparison.property_name} mismatch")
        else:
            self.warnings.append(comparison.message or f"{comparison.property_name} mismatch")

PropertyComparer ¶

PropertyComparer(*, fps_tolerance=FPS_TOLERANCE, duration_tolerance=DURATION_TOLERANCE)

Compare video properties between original and converted files.

This class provides comparison of video properties to ensure that the conversion process did not alter critical parameters.

SDS Reference: SDS-P05-002 SRS Reference: SRS-502 (Video Property Comparison)

Tolerance Settings

Resolution: Exact match required (Error on mismatch)
Frame Rate: ±0.001 fps tolerance (Warning on mismatch)
Duration: ±0.5 second tolerance (Warning on mismatch)
Aspect Ratio: Exact match required (Error on mismatch)
Audio Codec: Exact match required (Warning on mismatch)
Audio Channels: Exact match required (Error on mismatch)

Example

comparer = PropertyComparer() result = comparer.compare(original_info, converted_info) if result.all_match: ... print("All properties match!") else: ... print(f"Errors: {result.errors}") ... print(f"Warnings: {result.warnings}")

Initialize the property comparer.

Parameters:

Name	Type	Description	Default
`fps_tolerance`	`float`	Tolerance for frame rate comparison (default: ±0.001 fps).	`FPS_TOLERANCE`
`duration_tolerance`	`float`	Tolerance for duration comparison (default: ±0.5 seconds).	`DURATION_TOLERANCE`

Source code in src/video_converter/processors/quality_validator.py

def __init__(
    self,
    *,
    fps_tolerance: float = FPS_TOLERANCE,
    duration_tolerance: float = DURATION_TOLERANCE,
) -> None:
    """Initialize the property comparer.

    Args:
        fps_tolerance: Tolerance for frame rate comparison (default: ±0.001 fps).
        duration_tolerance: Tolerance for duration comparison (default: ±0.5 seconds).
    """
    self.fps_tolerance = fps_tolerance
    self.duration_tolerance = duration_tolerance

compare ¶

compare(original, converted)

Compare all properties between original and converted video.

Parameters:

Name	Type	Description	Default
`original`	`VideoInfo`	VideoInfo from the original file.	required
`converted`	`VideoInfo`	VideoInfo from the converted file.	required

Returns:

Type	Description
`PropertyComparisonResult`	PropertyComparisonResult containing all comparison details.

Source code in src/video_converter/processors/quality_validator.py

def compare(
    self,
    original: VideoInfo,
    converted: VideoInfo,
) -> PropertyComparisonResult:
    """Compare all properties between original and converted video.

    Args:
        original: VideoInfo from the original file.
        converted: VideoInfo from the converted file.

    Returns:
        PropertyComparisonResult containing all comparison details.
    """
    result = PropertyComparisonResult()

    # Compare resolution
    result.add_comparison(self._compare_resolution(original, converted))

    # Compare frame rate
    result.add_comparison(self._compare_fps(original, converted))

    # Compare duration
    result.add_comparison(self._compare_duration(original, converted))

    # Compare aspect ratio
    result.add_comparison(self._compare_aspect_ratio(original, converted))

    # Compare audio codec
    result.add_comparison(self._compare_audio_codec(original, converted))

    # Compare audio channels
    result.add_comparison(self._compare_audio_channels(original, converted))

    return result

ContentType ¶

Bases: Enum

Content type for compression ratio validation.

Different content types have different expected compression ratios. High-motion content (sports, action) compresses less efficiently, while low-motion content (interviews, static scenes) compresses better.

Attributes:

Name	Type	Description
`STANDARD`		Normal video content (default).
`HIGH_MOTION`		High-motion content like sports or action.
`LOW_MOTION`		Low-motion content like interviews or static scenes.

CompressionRange `dataclass` ¶

CompressionRange(min_ratio, max_ratio, content_type=STANDARD)

Expected compression ratio range for a content type.

SDS Reference: SDS-P05-003

Attributes:

Name	Type	Description
`min_ratio`	`float`	Minimum expected compression ratio (0.0 to 1.0).
`max_ratio`	`float`	Maximum expected compression ratio (0.0 to 1.0).
`content_type`	`ContentType`	The content type this range applies to.

__post_init__ ¶

__post_init__()

Validate range values.

Source code in src/video_converter/processors/quality_validator.py

def __post_init__(self) -> None:
    """Validate range values."""
    if not 0.0 <= self.min_ratio <= 1.0:
        raise ValueError(f"min_ratio must be between 0.0 and 1.0, got {self.min_ratio}")
    if not 0.0 <= self.max_ratio <= 1.0:
        raise ValueError(f"max_ratio must be between 0.0 and 1.0, got {self.max_ratio}")
    if self.min_ratio > self.max_ratio:
        raise ValueError(
            f"min_ratio ({self.min_ratio}) must be <= max_ratio ({self.max_ratio})"
        )

CompressionSeverity ¶

Bases: Enum

Severity level for compression ratio issues.

Attributes:

Name	Type	Description
`NORMAL`		Compression ratio is within expected range.
`WARNING`		Compression ratio is outside expected range but not critical.
`ERROR`		Compression ratio indicates a serious encoding issue.

CompressionValidationResult `dataclass` ¶

CompressionValidationResult(valid, compression_ratio, original_size, converted_size, content_type, severity, message='', expected_range=None)

Result of compression ratio validation.

SDS Reference: SDS-P05-003 SRS Reference: SRS-503 (Compression Ratio Validation)

Attributes:

Name	Type	Description
`valid`	`bool`	Overall validation result (True if within acceptable range).
`compression_ratio`	`float`	Calculated compression ratio (0.0 to 1.0).
`original_size`	`int`	Original file size in bytes.
`converted_size`	`int`	Converted file size in bytes.
`content_type`	`ContentType`	Content type used for validation.
`severity`	`CompressionSeverity`	Severity level of the result.
`message`	`str`	Human-readable description of the result.
`expected_range`	`CompressionRange \| None`	The expected compression range for the content type.

size_reduction_percent `property` ¶

size_reduction_percent

Get size reduction as percentage.

Returns:

Type	Description
`float`	Size reduction percentage (0-100).

file_grew `property` ¶

file_grew

Check if file size increased after conversion.

Returns:

Type	Description
`bool`	True if converted file is larger than original.

CompressionValidator ¶

CompressionValidator(*, ranges=None, critical_low=DEFAULT_CRITICAL_LOW, critical_high=DEFAULT_CRITICAL_HIGH)

Validate compression ratio is within expected range.

This class validates that the compression ratio of a converted video falls within expected ranges for different content types. It helps detect encoding issues such as: - File grew significantly (encoding issue) - Below expected compression (quality setting too high) - Too aggressive compression (quality loss)

SDS Reference: SDS-P05-003 SRS Reference: SRS-503 (Compression Ratio Validation)

Expected Compression Ranges

Standard Video: 30% - 70%
High Motion: 20% - 60%
Low Motion: 40% - 80%

Warning Thresholds

< 20%: File grew significantly (encoding issue)
< 30%: Below expected (quality setting too high?)
80%: Too aggressive (quality loss?)

Example

validator = CompressionValidator() result = validator.validate(100_000_000, 40_000_000) if result.valid: ... print(f"Compression OK: {result.compression_ratio:.1%}") else: ... print(f"Warning: {result.message}")

Attributes:

Name	Type	Description
`ranges`		Dictionary mapping content types to compression ranges.
`critical_low`		Threshold below which indicates serious encoding issue.
`critical_high`		Threshold above which indicates excessive compression.

Initialize the compression validator.

Parameters:

Name	Type	Description	Default
`ranges`	`dict[ContentType, CompressionRange] \| None`	Custom compression ranges per content type. Uses default ranges if not provided.	`None`
`critical_low`	`float`	Threshold below which is a critical issue (default: 0.20).	`DEFAULT_CRITICAL_LOW`
`critical_high`	`float`	Threshold above which is a critical issue (default: 0.80).	`DEFAULT_CRITICAL_HIGH`

Source code in src/video_converter/processors/quality_validator.py

def __init__(
    self,
    *,
    ranges: dict[ContentType, CompressionRange] | None = None,
    critical_low: float = DEFAULT_CRITICAL_LOW,
    critical_high: float = DEFAULT_CRITICAL_HIGH,
) -> None:
    """Initialize the compression validator.

    Args:
        ranges: Custom compression ranges per content type.
               Uses default ranges if not provided.
        critical_low: Threshold below which is a critical issue (default: 0.20).
        critical_high: Threshold above which is a critical issue (default: 0.80).
    """
    self.ranges = ranges if ranges is not None else self.DEFAULT_RANGES.copy()
    self.critical_low = critical_low
    self.critical_high = critical_high

validate ¶

validate(original_size, converted_size, content_type=STANDARD)

Validate compression ratio is within expected range.

Calculates the compression ratio and checks if it falls within the expected range for the given content type.

Parameters:

Name	Type	Description	Default
`original_size`	`int`	Original file size in bytes.	required
`converted_size`	`int`	Converted file size in bytes.	required
`content_type`	`ContentType`	Type of video content (default: STANDARD).	`STANDARD`

Returns:

Type	Description
`CompressionValidationResult`	CompressionValidationResult containing validation details.

Raises:

Type	Description
`ValueError`	If original_size is not positive.

Source code in src/video_converter/processors/quality_validator.py

def validate(
    self,
    original_size: int,
    converted_size: int,
    content_type: ContentType = ContentType.STANDARD,
) -> CompressionValidationResult:
    """Validate compression ratio is within expected range.

    Calculates the compression ratio and checks if it falls within
    the expected range for the given content type.

    Args:
        original_size: Original file size in bytes.
        converted_size: Converted file size in bytes.
        content_type: Type of video content (default: STANDARD).

    Returns:
        CompressionValidationResult containing validation details.

    Raises:
        ValueError: If original_size is not positive.
    """
    if original_size <= 0:
        raise ValueError(f"original_size must be positive, got {original_size}")

    if converted_size < 0:
        raise ValueError(f"converted_size cannot be negative, got {converted_size}")

    # Calculate compression ratio: 1 - (converted / original)
    # A ratio of 0.5 means the file is 50% smaller
    compression_ratio = 1 - (converted_size / original_size)

    expected_range = self.ranges.get(content_type)

    # Determine severity and validity
    severity, valid, message = self._evaluate_ratio(
        compression_ratio, expected_range, content_type
    )

    return CompressionValidationResult(
        valid=valid,
        compression_ratio=compression_ratio,
        original_size=original_size,
        converted_size=converted_size,
        content_type=content_type,
        severity=severity,
        message=message,
        expected_range=expected_range,
    )

get_expected_range ¶

get_expected_range(content_type)

Get the expected compression range for a content type.

Parameters:

Name	Type	Description	Default
`content_type`	`ContentType`	The content type to get the range for.	required

Returns:

Type	Description
`CompressionRange \| None`	CompressionRange for the content type, or None if not defined.

Source code in src/video_converter/processors/quality_validator.py

def get_expected_range(self, content_type: ContentType) -> CompressionRange | None:
    """Get the expected compression range for a content type.

    Args:
        content_type: The content type to get the range for.

    Returns:
        CompressionRange for the content type, or None if not defined.
    """
    return self.ranges.get(content_type)

set_range ¶

set_range(content_type, range_)

Set a custom compression range for a content type.

Parameters:

Name	Type	Description	Default
`content_type`	`ContentType`	The content type to set the range for.	required
`range_`	`CompressionRange`	The compression range to set.	required

Source code in src/video_converter/processors/quality_validator.py

def set_range(self, content_type: ContentType, range_: CompressionRange) -> None:
    """Set a custom compression range for a content type.

    Args:
        content_type: The content type to set the range for.
        range_: The compression range to set.
    """
    self.ranges[content_type] = range_

GPS Handler¶

gps ¶

GPS coordinate handling for video metadata.

This module provides specialized handling for GPS coordinate preservation during video conversion. It ensures location data is accurately maintained across different metadata container formats.

SDS Reference: SDS-P01-003 SRS Reference: SRS-402 (GPS Preservation)

Example

from video_converter.utils.command_runner import CommandRunner from video_converter.processors.gps import GPSHandler handler = GPSHandler(CommandRunner()) coords = handler.extract(Path("video.mp4")) if coords: ... print(f"Location: {coords}") ... print(f"QuickTime format: {coords.to_quicktime()}")

Verify GPS was preserved¶

result = handler.verify(original_path, converted_path) if result.passed: ... print("GPS preserved successfully")

GPSFormat ¶

Bases: Enum

GPS coordinate storage formats.

Different metadata containers store GPS data in different formats. This enum identifies the source format for proper handling.

GPSCoordinates `dataclass` ¶

GPSCoordinates(latitude, longitude, altitude=None, accuracy=None, source_format=DECIMAL)

GPS coordinate data with format conversion support.

This class represents GPS coordinates with support for converting between different metadata container formats. It preserves accuracy to 6 decimal places (~0.1 meter precision).

Attributes:

Name	Type	Description
`latitude`	`float`	Latitude in decimal degrees (-90 to 90).
`longitude`	`float`	Longitude in decimal degrees (-180 to 180).
`altitude`	`float \| None`	Altitude in meters (optional).
`accuracy`	`float \| None`	Horizontal accuracy in meters (optional).
`source_format`	`GPSFormat`	Original format the coordinates were parsed from.

Example

coords = GPSCoordinates( ... latitude=37.7749, ... longitude=-122.4194, ... altitude=10.5 ... ) print(coords.to_quicktime()) +37.774900-122.419400/

__post_init__ ¶

__post_init__()

Validate and normalize coordinates.

Source code in src/video_converter/processors/gps.py

def __post_init__(self) -> None:
    """Validate and normalize coordinates."""
    if not -90 <= self.latitude <= 90:
        raise ValueError(f"Latitude must be between -90 and 90: {self.latitude}")
    if not -180 <= self.longitude <= 180:
        raise ValueError(f"Longitude must be between -180 and 180: {self.longitude}")

str ¶

__str__()

Return human-readable coordinate string.

Source code in src/video_converter/processors/gps.py

def __str__(self) -> str:
    """Return human-readable coordinate string."""
    lat_dir = "N" if self.latitude >= 0 else "S"
    lon_dir = "E" if self.longitude >= 0 else "W"
    result = (
        f"{abs(self.latitude):.{self.PRECISION}f}°{lat_dir}, "
        f"{abs(self.longitude):.{self.PRECISION}f}°{lon_dir}"
    )
    if self.altitude is not None:
        result += f", {self.altitude:.1f}m"
    if self.accuracy is not None:
        result += f" (±{self.accuracy:.1f}m)"
    return result

to_quicktime ¶

to_quicktime()

Convert to QuickTime GPS format.

QuickTime uses ISO 6709 format: +DD.DDDDDD-DDD.DDDDDD/

Returns:

Type	Description
`str`	GPS string in QuickTime format.

Example

coords = GPSCoordinates(37.7749, -122.4194) coords.to_quicktime() '+37.774900-122.419400/'

Source code in src/video_converter/processors/gps.py

def to_quicktime(self) -> str:
    """Convert to QuickTime GPS format.

    QuickTime uses ISO 6709 format: +DD.DDDDDD-DDD.DDDDDD/

    Returns:
        GPS string in QuickTime format.

    Example:
        >>> coords = GPSCoordinates(37.7749, -122.4194)
        >>> coords.to_quicktime()
        '+37.774900-122.419400/'
    """
    lat_sign = "+" if self.latitude >= 0 else ""
    lon_sign = "+" if self.longitude >= 0 else ""
    result = (
        f"{lat_sign}{self.latitude:.{self.PRECISION}f}"
        f"{lon_sign}{self.longitude:.{self.PRECISION}f}"
    )
    if self.altitude is not None:
        alt_sign = "+" if self.altitude >= 0 else ""
        result += f"{alt_sign}{self.altitude:.2f}"
    return result + "/"

to_xmp ¶

to_xmp()

Convert to XMP GPS format.

XMP uses "DD.DDDDDD,D" format with direction suffix.

Returns:

Type	Description
`tuple[str, str]`	Tuple of (latitude_string, longitude_string).

Example

coords = GPSCoordinates(37.7749, -122.4194) coords.to_xmp() ('37.774900 N', '122.419400 W')

Source code in src/video_converter/processors/gps.py

def to_xmp(self) -> tuple[str, str]:
    """Convert to XMP GPS format.

    XMP uses "DD.DDDDDD,D" format with direction suffix.

    Returns:
        Tuple of (latitude_string, longitude_string).

    Example:
        >>> coords = GPSCoordinates(37.7749, -122.4194)
        >>> coords.to_xmp()
        ('37.774900 N', '122.419400 W')
    """
    lat_dir = "N" if self.latitude >= 0 else "S"
    lon_dir = "E" if self.longitude >= 0 else "W"
    return (
        f"{abs(self.latitude):.{self.PRECISION}f} {lat_dir}",
        f"{abs(self.longitude):.{self.PRECISION}f} {lon_dir}",
    )

to_exif_dms ¶

to_exif_dms()

Convert to EXIF DMS (degrees, minutes, seconds) format.

Returns:

Type	Description
`tuple[str, str, str, str]`	Tuple of (latitude, lat_ref, longitude, lon_ref).

Example

coords = GPSCoordinates(37.775, -122.4194) lat, lat_ref, lon, lon_ref = coords.to_exif_dms() print(f"{lat} {lat_ref}") 37 deg 46' 30.00" N

Source code in src/video_converter/processors/gps.py

def to_exif_dms(self) -> tuple[str, str, str, str]:
    """Convert to EXIF DMS (degrees, minutes, seconds) format.

    Returns:
        Tuple of (latitude, lat_ref, longitude, lon_ref).

    Example:
        >>> coords = GPSCoordinates(37.775, -122.4194)
        >>> lat, lat_ref, lon, lon_ref = coords.to_exif_dms()
        >>> print(f"{lat} {lat_ref}")
        37 deg 46' 30.00" N
    """
    lat_dir = "N" if self.latitude >= 0 else "S"
    lon_dir = "E" if self.longitude >= 0 else "W"

    lat_dms = self._decimal_to_dms(abs(self.latitude))
    lon_dms = self._decimal_to_dms(abs(self.longitude))

    return (lat_dms, lat_dir, lon_dms, lon_dir)

matches ¶

matches(other, tolerance=None)

Check if coordinates match within tolerance.

Parameters:

Name	Type	Description	Default
`other`	`GPSCoordinates`	Another GPSCoordinates to compare.	required
`tolerance`	`float \| None`	Optional custom tolerance in decimal degrees. Defaults to TOLERANCE (0.000001 = ~0.1m).	`None`

Returns:

Type	Description
`bool`	True if coordinates match within tolerance.

Source code in src/video_converter/processors/gps.py

def matches(self, other: GPSCoordinates, tolerance: float | None = None) -> bool:
    """Check if coordinates match within tolerance.

    Args:
        other: Another GPSCoordinates to compare.
        tolerance: Optional custom tolerance in decimal degrees.
                  Defaults to TOLERANCE (0.000001 = ~0.1m).

    Returns:
        True if coordinates match within tolerance.
    """
    tol = tolerance if tolerance is not None else self.TOLERANCE
    lat_match = abs(self.latitude - other.latitude) <= tol
    lon_match = abs(self.longitude - other.longitude) <= tol
    return lat_match and lon_match

distance_to ¶

distance_to(other)

Calculate approximate distance to another coordinate in meters.

Uses the Haversine formula for spherical distance calculation.

Parameters:

Name	Type	Description	Default
`other`	`GPSCoordinates`	Another GPSCoordinates.	required

Returns:

Type	Description
`float`	Approximate distance in meters.

Source code in src/video_converter/processors/gps.py

def distance_to(self, other: GPSCoordinates) -> float:
    """Calculate approximate distance to another coordinate in meters.

    Uses the Haversine formula for spherical distance calculation.

    Args:
        other: Another GPSCoordinates.

    Returns:
        Approximate distance in meters.
    """
    import math

    R = 6371000  # Earth's radius in meters

    lat1 = math.radians(self.latitude)
    lat2 = math.radians(other.latitude)
    dlat = math.radians(other.latitude - self.latitude)
    dlon = math.radians(other.longitude - self.longitude)

    a = math.sin(dlat / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2
    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))

    return R * c

from_quicktime `classmethod` ¶

from_quicktime(value)

Parse QuickTime GPS format (ISO 6709).

Parameters:

Name	Type	Description	Default
`value`	`str`	GPS string like "+37.7749-122.4194/" or "+37.7749-122.4194+10.5/"	required

Returns:

Type	Description
`GPSCoordinates \| None`	GPSCoordinates if parsed successfully, None otherwise.

Source code in src/video_converter/processors/gps.py

@classmethod
def from_quicktime(cls, value: str) -> GPSCoordinates | None:
    """Parse QuickTime GPS format (ISO 6709).

    Args:
        value: GPS string like "+37.7749-122.4194/" or "+37.7749-122.4194+10.5/"

    Returns:
        GPSCoordinates if parsed successfully, None otherwise.
    """
    value = value.strip().rstrip("/")
    if not value:
        return None

    # Find all signed numbers in the string
    # Matches patterns like: +37.774900, -122.419400, +10.50
    parts = re.findall(r"[+-]?\d+(?:\.\d+)?", value)

    if len(parts) < 2:
        return None

    try:
        lat = float(parts[0])
        lon = float(parts[1])
        alt = float(parts[2]) if len(parts) > 2 else None
        return cls(
            latitude=lat,
            longitude=lon,
            altitude=alt,
            source_format=GPSFormat.QUICKTIME,
        )
    except (ValueError, TypeError, IndexError):
        return None

from_xmp `classmethod` ¶

from_xmp(lat_str, lon_str)

Parse XMP GPS format.

Parameters:

Name	Type	Description	Default
`lat_str`	`str`	Latitude string like "37.7749 N" or "37.7749,N"	required
`lon_str`	`str`	Longitude string like "122.4194 W" or "122.4194,W"	required

Returns:

Type	Description
`GPSCoordinates \| None`	GPSCoordinates if parsed successfully, None otherwise.

Source code in src/video_converter/processors/gps.py

@classmethod
def from_xmp(cls, lat_str: str, lon_str: str) -> GPSCoordinates | None:
    """Parse XMP GPS format.

    Args:
        lat_str: Latitude string like "37.7749 N" or "37.7749,N"
        lon_str: Longitude string like "122.4194 W" or "122.4194,W"

    Returns:
        GPSCoordinates if parsed successfully, None otherwise.
    """
    try:
        lat_match = re.match(r"([\d.]+)[,\s]*([NS])", lat_str.strip(), re.I)
        lon_match = re.match(r"([\d.]+)[,\s]*([EW])", lon_str.strip(), re.I)

        if not lat_match or not lon_match:
            return None

        lat = float(lat_match.group(1))
        if lat_match.group(2).upper() == "S":
            lat = -lat

        lon = float(lon_match.group(1))
        if lon_match.group(2).upper() == "W":
            lon = -lon

        return cls(
            latitude=lat,
            longitude=lon,
            source_format=GPSFormat.XMP,
        )
    except (ValueError, TypeError):
        return None

from_exif_dms `classmethod` ¶

from_exif_dms(lat_dms, lat_ref, lon_dms, lon_ref)

Parse EXIF DMS format.

Parameters:

Name	Type	Description	Default
`lat_dms`	`str`	Latitude in DMS like "37 deg 46' 30.00""	required
`lat_ref`	`str`	Latitude reference ("N" or "S")	required
`lon_dms`	`str`	Longitude in DMS like "122 deg 25' 10.00""	required
`lon_ref`	`str`	Longitude reference ("E" or "W")	required

Returns:

Type	Description
`GPSCoordinates \| None`	GPSCoordinates if parsed successfully, None otherwise.

Source code in src/video_converter/processors/gps.py

@classmethod
def from_exif_dms(
    cls,
    lat_dms: str,
    lat_ref: str,
    lon_dms: str,
    lon_ref: str,
) -> GPSCoordinates | None:
    """Parse EXIF DMS format.

    Args:
        lat_dms: Latitude in DMS like "37 deg 46' 30.00\""
        lat_ref: Latitude reference ("N" or "S")
        lon_dms: Longitude in DMS like "122 deg 25' 10.00\""
        lon_ref: Longitude reference ("E" or "W")

    Returns:
        GPSCoordinates if parsed successfully, None otherwise.
    """
    try:
        lat = cls._parse_dms(lat_dms)
        if lat_ref.upper() == "S":
            lat = -lat

        lon = cls._parse_dms(lon_dms)
        if lon_ref.upper() == "W":
            lon = -lon

        return cls(
            latitude=lat,
            longitude=lon,
            source_format=GPSFormat.EXIF,
        )
    except (ValueError, TypeError):
        return None

GPSVerificationResult `dataclass` ¶

GPSVerificationResult(passed, original=None, converted=None, distance_meters=None, tolerance_used=TOLERANCE, details='')

Result of GPS coordinate verification.

Attributes:

Name	Type	Description
`passed`	`bool`	True if GPS was preserved within tolerance.
`original`	`GPSCoordinates \| None`	Original GPS coordinates (None if not present).
`converted`	`GPSCoordinates \| None`	Converted GPS coordinates (None if not present).
`distance_meters`	`float \| None`	Distance between coordinates in meters (if both exist).
`tolerance_used`	`float`	Tolerance used for comparison in decimal degrees.
`details`	`str`	Additional verification details.

GPSHandler ¶

GPSHandler(metadata_processor=None)

Handle GPS coordinate preservation during video conversion.

This class provides specialized methods for extracting, applying, and verifying GPS coordinates across different metadata formats.

Example

handler = GPSHandler(CommandRunner()) coords = handler.extract(Path("video.mp4")) if coords: ... handler.apply(Path("output.mp4"), coords) ... result = handler.verify(Path("video.mp4"), Path("output.mp4")) ... print(f"Verification: {result.passed}")

Initialize GPSHandler.

Parameters:

Name	Type	Description	Default
`metadata_processor`	`MetadataProcessor \| None`	MetadataProcessor to use for metadata operations. If None, creates a new one.	`None`

Source code in src/video_converter/processors/gps.py

def __init__(self, metadata_processor: MetadataProcessor | None = None) -> None:
    """Initialize GPSHandler.

    Args:
        metadata_processor: MetadataProcessor to use for metadata operations.
                          If None, creates a new one.
    """
    self._processor = metadata_processor or MetadataProcessor()

extract ¶

extract(path)

Extract GPS coordinates from a video file.

Checks all possible GPS tag locations (QuickTime, XMP, EXIF, Keys) and returns coordinates if found.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required

Returns:

Type	Description
`GPSCoordinates \| None`	GPSCoordinates if found, None otherwise.

Raises:

Type	Description
`FileNotFoundError`	If the file doesn't exist.
`MetadataExtractionError`	If extraction fails.

Source code in src/video_converter/processors/gps.py

def extract(self, path: Path) -> GPSCoordinates | None:
    """Extract GPS coordinates from a video file.

    Checks all possible GPS tag locations (QuickTime, XMP, EXIF, Keys)
    and returns coordinates if found.

    Args:
        path: Path to the video file.

    Returns:
        GPSCoordinates if found, None otherwise.

    Raises:
        FileNotFoundError: If the file doesn't exist.
        MetadataExtractionError: If extraction fails.
    """
    if not path.exists():
        raise FileNotFoundError(f"File not found: {path}")

    metadata = self._processor.extract(path)
    return self._parse_gps_from_metadata(metadata)

apply ¶

apply(path, coords, *, overwrite_original=True)

Apply GPS coordinates to a video file.

Writes GPS data in multiple formats to ensure maximum compatibility.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required
`coords`	`GPSCoordinates`	GPS coordinates to apply.	required
`overwrite_original`	`bool`	If True, modify file in place.	`True`

Returns:

Type	Description
`bool`	True if GPS was applied successfully.

Raises:

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

Source code in src/video_converter/processors/gps.py

def apply(
    self,
    path: Path,
    coords: GPSCoordinates,
    *,
    overwrite_original: bool = True,
) -> bool:
    """Apply GPS coordinates to a video file.

    Writes GPS data in multiple formats to ensure maximum compatibility.

    Args:
        path: Path to the video file.
        coords: GPS coordinates to apply.
        overwrite_original: If True, modify file in place.

    Returns:
        True if GPS was applied successfully.

    Raises:
        FileNotFoundError: If the file doesn't exist.
    """
    if not path.exists():
        raise FileNotFoundError(f"File not found: {path}")

    # Set GPS tags in multiple formats for maximum compatibility
    tags_to_set = self._build_gps_tags(coords)

    success = True
    for tag, value in tags_to_set.items():
        result = self._processor.set_tag(
            path, tag, value, overwrite_original=overwrite_original
        )
        if not result:
            success = False

    return success

copy ¶

copy(source, dest, *, overwrite_original=True)

Copy GPS data from source to destination file.

Parameters:

Name	Type	Description	Default
`source`	`Path`	Source file with GPS data.	required
`dest`	`Path`	Destination file.	required
`overwrite_original`	`bool`	If True, modify dest in place.	`True`

Returns:

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

Source code in src/video_converter/processors/gps.py

def copy(
    self,
    source: Path,
    dest: Path,
    *,
    overwrite_original: bool = True,
) -> bool:
    """Copy GPS data from source to destination file.

    Args:
        source: Source file with GPS data.
        dest: Destination file.
        overwrite_original: If True, modify dest in place.

    Returns:
        True if GPS was copied successfully.
    """
    return self._processor.copy_gps(source, dest, overwrite_original=overwrite_original)

verify ¶

verify(original, converted, *, tolerance=None)

Verify GPS was preserved during conversion.

Parameters:

Name	Type	Description	Default
`original`	`Path`	Original source file.	required
`converted`	`Path`	Converted output file.	required
`tolerance`	`float \| None`	Custom tolerance in decimal degrees. Defaults to GPSCoordinates.TOLERANCE (~0.1m).	`None`

Returns:

Type	Description
`GPSVerificationResult`	GPSVerificationResult with detailed comparison.

Raises:

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

Source code in src/video_converter/processors/gps.py

def verify(
    self,
    original: Path,
    converted: Path,
    *,
    tolerance: float | None = None,
) -> GPSVerificationResult:
    """Verify GPS was preserved during conversion.

    Args:
        original: Original source file.
        converted: Converted output file.
        tolerance: Custom tolerance in decimal degrees.
                  Defaults to GPSCoordinates.TOLERANCE (~0.1m).

    Returns:
        GPSVerificationResult with detailed comparison.

    Raises:
        FileNotFoundError: If either file doesn't exist.
    """
    if not original.exists():
        raise FileNotFoundError(f"Original file not found: {original}")
    if not converted.exists():
        raise FileNotFoundError(f"Converted file not found: {converted}")

    tol = tolerance if tolerance is not None else GPSCoordinates.TOLERANCE

    try:
        orig_gps = self.extract(original)
    except MetadataExtractionError:
        orig_gps = None

    try:
        conv_gps = self.extract(converted)
    except MetadataExtractionError:
        conv_gps = None

    # Case 1: No GPS in original (OK)
    if orig_gps is None:
        return GPSVerificationResult(
            passed=True,
            original=None,
            converted=conv_gps,
            tolerance_used=tol,
            details="No GPS in original file",
        )

    # Case 2: GPS in original but not in converted (FAIL)
    if conv_gps is None:
        return GPSVerificationResult(
            passed=False,
            original=orig_gps,
            converted=None,
            tolerance_used=tol,
            details="GPS missing in converted file",
        )

    # Case 3: Compare coordinates
    distance = orig_gps.distance_to(conv_gps)
    matches = orig_gps.matches(conv_gps, tolerance=tol)

    if matches:
        details = f"GPS preserved (distance: {distance:.2f}m)"
    else:
        details = (
            f"GPS mismatch: original {orig_gps} vs converted {conv_gps} "
            f"(distance: {distance:.2f}m)"
        )

    return GPSVerificationResult(
        passed=matches,
        original=orig_gps,
        converted=conv_gps,
        distance_meters=distance,
        tolerance_used=tol,
        details=details,
    )

has_gps ¶

has_gps(path)

Check if a video file has GPS data.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file.	required

Returns:

Type	Description
`bool`	True if the file has GPS coordinates.

Source code in src/video_converter/processors/gps.py

def has_gps(self, path: Path) -> bool:
    """Check if a video file has GPS data.

    Args:
        path: Path to the video file.

    Returns:
        True if the file has GPS coordinates.
    """
    try:
        return self.extract(path) is not None
    except (FileNotFoundError, MetadataExtractionError):
        return False

VMAF Analyzer¶

vmaf_analyzer ¶

VMAF (Video Multimethod Assessment Fusion) quality measurement module.

This module provides VMAF-based perceptual video quality measurement to compare original and converted video files. VMAF is a metric developed by Netflix that predicts subjective video quality.

SDS Reference: SDS-P05-004 SRS Reference: SRS-504 (VMAF Quality Measurement)

VMAF Score Interpretation

93+: Visually lossless (indistinguishable from original)
80-93: High quality (minor differences, acceptable for most use cases)
60-80: Good quality (noticeable differences, acceptable for streaming)
<60: Noticeable degradation (quality issues visible)

Example

analyzer = VmafAnalyzer() if analyzer.is_available(): ... result = analyzer.analyze( ... original=Path("original.mp4"), ... converted=Path("converted.mp4"), ... ) ... print(f"VMAF Score: {result.scores.mean:.2f}") ... print(f"Quality: {result.quality_level.name}") ... else: ... print("VMAF not available (libvmaf not installed)")

VmafNotAvailableError ¶

VmafNotAvailableError(message='libvmaf is not available')

Bases: Exception

Raised when VMAF analysis is requested but libvmaf is not available.

This exception indicates that FFmpeg was not compiled with libvmaf support or libvmaf is not installed on the system.

Attributes:

Name	Type	Description
`message`		Explanation of why VMAF is not available.

Source code in src/video_converter/processors/vmaf_analyzer.py

def __init__(self, message: str = "libvmaf is not available") -> None:
    self.message = message
    super().__init__(
        f"{message}. Install FFmpeg with libvmaf support: "
        "brew install ffmpeg (ensure --with-libvmaf or similar option)"
    )

VmafAnalysisError ¶

VmafAnalysisError(original, converted, reason)

Bases: Exception

Raised when VMAF analysis fails.

Attributes:

Name	Type	Description
`original`		Path to the original video.
`converted`		Path to the converted video.
`reason`		Explanation of the failure.

Source code in src/video_converter/processors/vmaf_analyzer.py

def __init__(
    self,
    original: Path,
    converted: Path,
    reason: str,
) -> None:
    self.original = original
    self.converted = converted
    self.reason = reason
    super().__init__(f"VMAF analysis failed for {original.name} vs {converted.name}: {reason}")

VmafQualityLevel ¶

Bases: Enum

VMAF quality level classification.

Based on Netflix's VMAF score interpretation guidelines.

Attributes:

Name	Type	Description
`VISUALLY_LOSSLESS`		Score >= 93, indistinguishable from original.
`HIGH_QUALITY`		Score 80-93, minor differences acceptable for most use cases.
`GOOD_QUALITY`		Score 60-80, noticeable differences acceptable for streaming.
`NOTICEABLE_DEGRADATION`		Score < 60, visible quality issues.

from_score `classmethod` ¶

from_score(score)

Determine quality level from VMAF score.

Parameters:

Name	Type	Description	Default
`score`	`float`	VMAF score (0-100).	required

Returns:

Type	Description
`VmafQualityLevel`	Corresponding VmafQualityLevel.

Source code in src/video_converter/processors/vmaf_analyzer.py

@classmethod
def from_score(cls, score: float) -> VmafQualityLevel:
    """Determine quality level from VMAF score.

    Args:
        score: VMAF score (0-100).

    Returns:
        Corresponding VmafQualityLevel.
    """
    if score >= VMAF_THRESHOLD_VISUALLY_LOSSLESS:
        return cls.VISUALLY_LOSSLESS
    elif score >= VMAF_THRESHOLD_HIGH_QUALITY:
        return cls.HIGH_QUALITY
    elif score >= VMAF_THRESHOLD_GOOD_QUALITY:
        return cls.GOOD_QUALITY
    else:
        return cls.NOTICEABLE_DEGRADATION

VmafScores `dataclass` ¶

VmafScores(mean, min, max, percentile_5, percentile_95, harmonic_mean=None, std_dev=None)

VMAF score statistics.

Contains various percentile and aggregate scores from VMAF analysis.

Attributes:

Name	Type	Description
`mean`	`float`	Mean VMAF score across all frames.
`min`	`float`	Minimum VMAF score (worst frame).
`max`	`float`	Maximum VMAF score (best frame).
`percentile_5`	`float`	5th percentile score (worst 5% of frames).
`percentile_95`	`float`	95th percentile score (best 5% excluded).
`harmonic_mean`	`float \| None`	Harmonic mean (if available).
`std_dev`	`float \| None`	Standard deviation (if available).

quality_level `property` ¶

quality_level

Get quality level based on mean score.

str ¶

__str__()

Return human-readable summary.

Source code in src/video_converter/processors/vmaf_analyzer.py

def __str__(self) -> str:
    """Return human-readable summary."""
    return (
        f"VMAF: mean={self.mean:.2f}, min={self.min:.2f}, max={self.max:.2f}, "
        f"5th={self.percentile_5:.2f}, 95th={self.percentile_95:.2f}"
    )

VmafResult `dataclass` ¶

VmafResult(original_path, converted_path, scores, quality_level, frame_count, sampled=False, sample_interval=1, model_version='vmaf_v0.6.1', warnings=list(), raw_data=None)

Complete VMAF analysis result.

SDS Reference: SDS-P05-004

Attributes:

Name	Type	Description
`original_path`	`Path`	Path to the original video file.
`converted_path`	`Path`	Path to the converted video file.
`scores`	`VmafScores`	VMAF score statistics.
`quality_level`	`VmafQualityLevel`	Quality classification based on mean score.
`frame_count`	`int`	Number of frames analyzed.
`sampled`	`bool`	Whether sampling was used for faster analysis.
`sample_interval`	`int`	Sampling interval used (1 = every frame).
`model_version`	`str`	VMAF model version used.
`warnings`	`list[str]`	List of warning messages.
`raw_data`	`dict[str, Any] \| None`	Raw JSON output from VMAF (if available).

is_visually_lossless `property` ¶

is_visually_lossless

Check if quality is visually lossless (VMAF >= 93).

is_acceptable `property` ¶

is_acceptable

Check if quality is acceptable for most use cases (VMAF >= 80).

add_warning ¶

add_warning(message)

Add a warning message.

Source code in src/video_converter/processors/vmaf_analyzer.py

def add_warning(self, message: str) -> None:
    """Add a warning message."""
    self.warnings.append(message)

str ¶

__str__()

Return human-readable summary.

Source code in src/video_converter/processors/vmaf_analyzer.py

def __str__(self) -> str:
    """Return human-readable summary."""
    sampled_str = f" (sampled 1:{self.sample_interval})" if self.sampled else ""
    return (
        f"VMAF Analysis{sampled_str}: {self.scores.mean:.2f} "
        f"({self.quality_level.value}) - {self.frame_count} frames"
    )

VmafAnalyzer ¶

VmafAnalyzer(*, ffmpeg_path=None, timeout=VMAF_ANALYSIS_TIMEOUT, model_path=None, command_runner=None)

VMAF quality analyzer for video comparison.

This class provides VMAF-based quality measurement to compare original and converted videos. It supports sampling for faster analysis and gracefully handles missing libvmaf.

SDS Reference: SDS-P05-004 SRS Reference: SRS-504 (VMAF Quality Measurement)

Features

Calculate VMAF scores between original and converted videos
Support sampling for faster analysis of long videos
Report percentile scores (min, 5th, mean, 95th)
Gracefully handle missing libvmaf

Example

analyzer = VmafAnalyzer()

Check availability first¶

if not analyzer.is_available(): ... print("VMAF not available") ... return

Full analysis¶

result = analyzer.analyze( ... original=Path("original.mp4"), ... converted=Path("converted.mp4"), ... ) print(f"Score: {result.scores.mean:.2f}")

Sampled analysis for long videos (faster)¶

result = analyzer.analyze( ... original=Path("long_video.mp4"), ... converted=Path("converted.mp4"), ... sample_interval=10, # Analyze every 10th frame ... )

Attributes:

Name	Type	Description
`ffmpeg_path`		Path to FFmpeg executable.
`timeout`		Default timeout for VMAF analysis.
`model_path`		Path to VMAF model file (optional).

Initialize VMAF analyzer.

Parameters:

Name	Type	Description	Default
`ffmpeg_path`	`str \| None`	Custom path to FFmpeg. Uses PATH if None.	`None`
`timeout`	`float`	Default timeout for analysis in seconds.	`VMAF_ANALYSIS_TIMEOUT`
`model_path`	`str \| None`	Custom path to VMAF model file. Uses default if None.	`None`
`command_runner`	`CommandRunner \| None`	CommandRunner instance. Creates new one if None.	`None`

Source code in src/video_converter/processors/vmaf_analyzer.py

def __init__(
    self,
    *,
    ffmpeg_path: str | None = None,
    timeout: float = VMAF_ANALYSIS_TIMEOUT,
    model_path: str | None = None,
    command_runner: CommandRunner | None = None,
) -> None:
    """Initialize VMAF analyzer.

    Args:
        ffmpeg_path: Custom path to FFmpeg. Uses PATH if None.
        timeout: Default timeout for analysis in seconds.
        model_path: Custom path to VMAF model file. Uses default if None.
        command_runner: CommandRunner instance. Creates new one if None.
    """
    self._ffmpeg_path = ffmpeg_path or self.FFMPEG_CMD
    self._timeout = timeout
    self._model_path = model_path
    self._runner = command_runner or CommandRunner()
    self._availability_checked = False
    self._is_available = False

is_available ¶

is_available()

Check if VMAF analysis is available.

Checks if FFmpeg is installed and has libvmaf support.

Returns:

Type	Description
`bool`	True if VMAF is available, False otherwise.

Source code in src/video_converter/processors/vmaf_analyzer.py

def is_available(self) -> bool:
    """Check if VMAF analysis is available.

    Checks if FFmpeg is installed and has libvmaf support.

    Returns:
        True if VMAF is available, False otherwise.
    """
    if self._availability_checked:
        return self._is_available

    self._is_available = self._check_vmaf_support()
    self._availability_checked = True
    return self._is_available

analyze ¶

analyze(original, converted, *, sample_interval=1, timeout=None, resolution=None)

Analyze video quality using VMAF.

Compares the converted video against the original and returns VMAF scores and quality assessment.

Parameters:

Name	Type	Description	Default
`original`	`Path`	Path to the original (reference) video.	required
`converted`	`Path`	Path to the converted (distorted) video.	required
`sample_interval`	`int`	Analyze every Nth frame (1 = all frames). Use higher values for faster analysis of long videos.	`1`
`timeout`	`float \| None`	Analysis timeout in seconds. Uses default if None.	`None`
`resolution`	`tuple[int, int] \| None`	Target resolution for comparison (width, height). If None, videos are scaled to match if needed.	`None`

Returns:

Type	Description
`VmafResult`	VmafResult containing scores and quality assessment.

Raises:

Type	Description
`VmafNotAvailableError`	If libvmaf is not available.
`VmafAnalysisError`	If analysis fails.
`FileNotFoundError`	If either video file doesn't exist.

Example

result = analyzer.analyze( ... original=Path("original.mp4"), ... converted=Path("converted.mp4"), ... sample_interval=5, # Every 5th frame for speed ... ) if result.is_visually_lossless: ... print("Excellent quality!")

Source code in src/video_converter/processors/vmaf_analyzer.py

def analyze(
    self,
    original: Path,
    converted: Path,
    *,
    sample_interval: int = 1,
    timeout: float | None = None,
    resolution: tuple[int, int] | None = None,
) -> VmafResult:
    """Analyze video quality using VMAF.

    Compares the converted video against the original and returns
    VMAF scores and quality assessment.

    Args:
        original: Path to the original (reference) video.
        converted: Path to the converted (distorted) video.
        sample_interval: Analyze every Nth frame (1 = all frames).
            Use higher values for faster analysis of long videos.
        timeout: Analysis timeout in seconds. Uses default if None.
        resolution: Target resolution for comparison (width, height).
            If None, videos are scaled to match if needed.

    Returns:
        VmafResult containing scores and quality assessment.

    Raises:
        VmafNotAvailableError: If libvmaf is not available.
        VmafAnalysisError: If analysis fails.
        FileNotFoundError: If either video file doesn't exist.

    Example:
        >>> result = analyzer.analyze(
        ...     original=Path("original.mp4"),
        ...     converted=Path("converted.mp4"),
        ...     sample_interval=5,  # Every 5th frame for speed
        ... )
        >>> if result.is_visually_lossless:
        ...     print("Excellent quality!")
    """
    if not self.is_available():
        raise VmafNotAvailableError()

    if not original.exists():
        raise FileNotFoundError(f"Original video not found: {original}")
    if not converted.exists():
        raise FileNotFoundError(f"Converted video not found: {converted}")

    timeout = timeout or self._timeout

    # Create temporary file for JSON output
    with tempfile.NamedTemporaryFile(
        mode="w",
        suffix=".json",
        delete=False,
    ) as tmp_file:
        json_path = Path(tmp_file.name)

    try:
        # Build and run FFmpeg command
        args = self._build_vmaf_command(
            original=original,
            converted=converted,
            json_output=json_path,
            sample_interval=sample_interval,
            resolution=resolution,
        )

        logger.info(f"Running VMAF analysis: {original.name} vs {converted.name}")
        logger.debug(f"Command: {' '.join(args)}")

        result = self._runner.run(args, timeout=timeout)

        if not result.success:
            raise VmafAnalysisError(original, converted, f"FFmpeg failed: {result.stderr}")

        # Parse results
        return self._parse_vmaf_output(
            original=original,
            converted=converted,
            json_path=json_path,
            sample_interval=sample_interval,
            stderr=result.stderr,
        )

    except CommandTimeoutError as e:
        raise VmafAnalysisError(
            original, converted, f"Analysis timed out after {timeout}s"
        ) from e
    finally:
        # Clean up temporary file
        if json_path.exists():
            json_path.unlink()

analyze_async `async` ¶

analyze_async(original, converted, *, sample_interval=1, timeout=None, resolution=None)

Analyze video quality using VMAF asynchronously.

Async version of analyze() for use in async contexts.

Parameters:

Name	Type	Description	Default
`original`	`Path`	Path to the original (reference) video.	required
`converted`	`Path`	Path to the converted (distorted) video.	required
`sample_interval`	`int`	Analyze every Nth frame (1 = all frames).	`1`
`timeout`	`float \| None`	Analysis timeout in seconds. Uses default if None.	`None`
`resolution`	`tuple[int, int] \| None`	Target resolution for comparison (width, height).	`None`

Returns:

Type	Description
`VmafResult`	VmafResult containing scores and quality assessment.

Raises:

Type	Description
`VmafNotAvailableError`	If libvmaf is not available.
`VmafAnalysisError`	If analysis fails.
`FileNotFoundError`	If either video file doesn't exist.

Source code in src/video_converter/processors/vmaf_analyzer.py

async def analyze_async(
    self,
    original: Path,
    converted: Path,
    *,
    sample_interval: int = 1,
    timeout: float | None = None,
    resolution: tuple[int, int] | None = None,
) -> VmafResult:
    """Analyze video quality using VMAF asynchronously.

    Async version of analyze() for use in async contexts.

    Args:
        original: Path to the original (reference) video.
        converted: Path to the converted (distorted) video.
        sample_interval: Analyze every Nth frame (1 = all frames).
        timeout: Analysis timeout in seconds. Uses default if None.
        resolution: Target resolution for comparison (width, height).

    Returns:
        VmafResult containing scores and quality assessment.

    Raises:
        VmafNotAvailableError: If libvmaf is not available.
        VmafAnalysisError: If analysis fails.
        FileNotFoundError: If either video file doesn't exist.
    """
    if not self.is_available():
        raise VmafNotAvailableError()

    if not original.exists():
        raise FileNotFoundError(f"Original video not found: {original}")
    if not converted.exists():
        raise FileNotFoundError(f"Converted video not found: {converted}")

    timeout = timeout or self._timeout

    # Create temporary file for JSON output
    with tempfile.NamedTemporaryFile(
        mode="w",
        suffix=".json",
        delete=False,
    ) as tmp_file:
        json_path = Path(tmp_file.name)

    try:
        # Build FFmpeg command
        args = self._build_vmaf_command(
            original=original,
            converted=converted,
            json_output=json_path,
            sample_interval=sample_interval,
            resolution=resolution,
        )

        logger.info(f"Running async VMAF analysis: {original.name} vs {converted.name}")

        result = await self._runner.run_async(args, timeout=timeout)

        if not result.success:
            raise VmafAnalysisError(original, converted, f"FFmpeg failed: {result.stderr}")

        # Parse results
        return self._parse_vmaf_output(
            original=original,
            converted=converted,
            json_path=json_path,
            sample_interval=sample_interval,
            stderr=result.stderr,
        )

    except asyncio.TimeoutError as e:
        raise VmafAnalysisError(
            original, converted, f"Analysis timed out after {timeout}s"
        ) from e
    finally:
        # Clean up temporary file
        if json_path.exists():
            json_path.unlink()

quick_analyze ¶

quick_analyze(original, converted, *, timeout=None)

Perform quick VMAF analysis with sampling.

Uses aggressive sampling for a quick quality estimate. Suitable for initial screening of conversion quality.

Parameters:

Name	Type	Description	Default
`original`	`Path`	Path to original video.	required
`converted`	`Path`	Path to converted video.	required
`timeout`	`float \| None`	Analysis timeout in seconds.	`None`

Returns:

Type	Description
`float \| None`	Mean VMAF score, or None if analysis fails.

Source code in src/video_converter/processors/vmaf_analyzer.py

def quick_analyze(
    self,
    original: Path,
    converted: Path,
    *,
    timeout: float | None = None,
) -> float | None:
    """Perform quick VMAF analysis with sampling.

    Uses aggressive sampling for a quick quality estimate.
    Suitable for initial screening of conversion quality.

    Args:
        original: Path to original video.
        converted: Path to converted video.
        timeout: Analysis timeout in seconds.

    Returns:
        Mean VMAF score, or None if analysis fails.
    """
    try:
        result = self.analyze(
            original=original,
            converted=converted,
            sample_interval=VMAF_DEFAULT_SAMPLE_INTERVAL,  # Every 30th frame
            timeout=timeout or VMAF_QUICK_TIMEOUT,  # 5 minute timeout
        )
        return result.scores.mean
    except (VmafNotAvailableError, VmafAnalysisError, FileNotFoundError):
        return None

get_quality_assessment ¶

get_quality_assessment(score)

Get human-readable quality assessment for a VMAF score.

Parameters:

Name	Type	Description	Default
`score`	`float`	VMAF score (0-100).	required

Returns:

Type	Description
`str`	Human-readable quality assessment string.

Source code in src/video_converter/processors/vmaf_analyzer.py

def get_quality_assessment(self, score: float) -> str:
    """Get human-readable quality assessment for a VMAF score.

    Args:
        score: VMAF score (0-100).

    Returns:
        Human-readable quality assessment string.
    """
    level = VmafQualityLevel.from_score(score)
    assessments = {
        VmafQualityLevel.VISUALLY_LOSSLESS: (
            f"Excellent ({score:.1f}): Visually lossless - indistinguishable from original"
        ),
        VmafQualityLevel.HIGH_QUALITY: (
            f"High Quality ({score:.1f}): Minor differences - acceptable for most use cases"
        ),
        VmafQualityLevel.GOOD_QUALITY: (
            f"Good Quality ({score:.1f}): Noticeable differences - acceptable for streaming"
        ),
        VmafQualityLevel.NOTICEABLE_DEGRADATION: (
            f"Degraded ({score:.1f}): Visible quality issues - "
            "consider re-encoding with higher quality"
        ),
    }
    return assessments[level]

Verification¶

verification ¶

Metadata verification system for video conversion.

This module provides a comprehensive metadata verification system that compares original and converted files to ensure all critical metadata was preserved during conversion.

SDS Reference: SDS-P01-004 SRS Reference: SRS-403 (Metadata Verification)

Example

from video_converter.processors.verification import MetadataVerifier verifier = MetadataVerifier() result = verifier.verify(Path("original.mp4"), Path("converted.mp4")) if result.passed: ... print("All metadata preserved successfully") else: ... for check in result.failed_checks: ... print(f"FAILED: {check.category} - {check.details}")

VerificationCategory ¶

Bases: Enum

Categories of metadata verification.

Attributes:

Name	Type	Description
`DATE_TIME`		Date and time metadata (CreateDate, ModifyDate).
`GPS`		GPS location data (Latitude, Longitude, Altitude).
`CAMERA`		Camera information (Make, Model).
`VIDEO`		Video properties (Duration, Rotation).
`AUDIO`		Audio properties (Codec, Channels).

CheckStatus ¶

Bases: Enum

Status of a single verification check.

Attributes:

Name	Type	Description
`PASSED`		Check passed successfully.
`FAILED`		Check failed - values don't match within tolerance.
`MISSING_IN_SOURCE`		Field not present in original file.
`MISSING_IN_CONVERTED`		Field was in original but missing in converted.
`ERROR`		An error occurred during verification.

ToleranceSettings `dataclass` ¶

ToleranceSettings(date_seconds=1.0, gps_degrees=1e-06, duration_seconds=0.1, numeric_relative=0.001)

Configurable tolerance settings for metadata verification.

Attributes:

Name	Type	Description
`date_seconds`	`float`	Tolerance for date/time comparison in seconds.
`gps_degrees`	`float`	Tolerance for GPS coordinates in decimal degrees.
`duration_seconds`	`float`	Tolerance for video duration in seconds.
`numeric_relative`	`float`	Relative tolerance for other numeric values.

strict `classmethod` ¶

strict()

Create strict tolerance settings with minimal margins.

Source code in src/video_converter/processors/verification.py

@classmethod
def strict(cls) -> ToleranceSettings:
    """Create strict tolerance settings with minimal margins."""
    return cls(
        date_seconds=0.0,
        gps_degrees=0.0,
        duration_seconds=0.0,
        numeric_relative=0.0,
    )

relaxed `classmethod` ¶

relaxed()

Create relaxed tolerance settings for problematic files.

Source code in src/video_converter/processors/verification.py

@classmethod
def relaxed(cls) -> ToleranceSettings:
    """Create relaxed tolerance settings for problematic files."""
    return cls(
        date_seconds=60.0,
        gps_degrees=0.0001,
        duration_seconds=1.0,
        numeric_relative=0.01,
    )

CheckResult `dataclass` ¶

CheckResult(category, field_name, status, original_value=None, converted_value=None, tolerance_used=None, details='')

Result of a single metadata verification check.

Attributes:

Name	Type	Description
`category`	`VerificationCategory`	The verification category this check belongs to.
`field_name`	`str`	Name of the metadata field being verified.
`status`	`CheckStatus`	Status of the verification check.
`original_value`	`Any`	Value from the original file.
`converted_value`	`Any`	Value from the converted file.
`tolerance_used`	`float \| None`	Tolerance value applied to comparison.
`details`	`str`	Additional details about the check result.

passed `property` ¶

passed

Return True if the check passed.

str ¶

__str__()

Return human-readable check result.

Source code in src/video_converter/processors/verification.py

def __str__(self) -> str:
    """Return human-readable check result."""
    status_str = "✓" if self.passed else "✗"
    return f"{status_str} {self.category.value}/{self.field_name}: {self.details}"

VerificationResult `dataclass` ¶

VerificationResult(passed, original_path, converted_path, checks=list(), verification_time=now(), tolerance_settings=ToleranceSettings())

Result of complete metadata verification between two files.

Attributes:

Name	Type	Description
`passed`	`bool`	True if all critical checks passed.
`original_path`	`Path`	Path to the original file.
`converted_path`	`Path`	Path to the converted file.
`checks`	`list[CheckResult]`	List of individual check results.
`verification_time`	`datetime`	When the verification was performed.
`tolerance_settings`	`ToleranceSettings`	Tolerance settings used for verification.

failed_checks `property` ¶

failed_checks

Return list of failed checks.

passed_checks `property` ¶

passed_checks

Return list of passed checks.

checks_by_category `property` ¶

checks_by_category

Group checks by category.

get_summary ¶

get_summary()

Generate a summary of the verification result.

Source code in src/video_converter/processors/verification.py

def get_summary(self) -> str:
    """Generate a summary of the verification result."""
    total = len(self.checks)
    passed = len(self.passed_checks)
    failed = len(self.failed_checks)

    lines = [
        f"Verification Result: {'PASSED' if self.passed else 'FAILED'}",
        f"Original: {self.original_path.name}",
        f"Converted: {self.converted_path.name}",
        f"Checks: {passed}/{total} passed, {failed} failed",
    ]

    if self.failed_checks:
        lines.append("\nFailed checks:")
        for check in self.failed_checks:
            lines.append(f"  - {check}")

    return "\n".join(lines)

MetadataVerifier ¶

MetadataVerifier(metadata_processor=None, gps_handler=None, tolerance=None)

Verify metadata preservation during video conversion.

This class provides comprehensive metadata verification comparing original and converted files across all critical metadata categories.

Attributes:

Name	Type	Description
`DATE_TIME_FIELDS`		Fields to check for date/time category.
`CAMERA_FIELDS`		Fields to check for camera information.
`VIDEO_FIELDS`		Fields to check for video properties.
`AUDIO_FIELDS`		Fields to check for audio properties.

Example

verifier = MetadataVerifier() result = verifier.verify(Path("original.mp4"), Path("converted.mp4")) print(result.get_summary()) Verification Result: PASSED Original: original.mp4 Converted: converted.mp4 Checks: 10/10 passed, 0 failed

Initialize MetadataVerifier.

Parameters:

Name	Type	Description	Default
`metadata_processor`	`MetadataProcessor \| None`	MetadataProcessor instance for metadata extraction.	`None`
`gps_handler`	`GPSHandler \| None`	GPSHandler instance for GPS verification.	`None`
`tolerance`	`ToleranceSettings \| None`	ToleranceSettings for comparison. Defaults to standard settings.	`None`

Source code in src/video_converter/processors/verification.py

def __init__(
    self,
    metadata_processor: MetadataProcessor | None = None,
    gps_handler: GPSHandler | None = None,
    tolerance: ToleranceSettings | None = None,
) -> None:
    """Initialize MetadataVerifier.

    Args:
        metadata_processor: MetadataProcessor instance for metadata extraction.
        gps_handler: GPSHandler instance for GPS verification.
        tolerance: ToleranceSettings for comparison. Defaults to standard settings.
    """
    self._processor = metadata_processor or MetadataProcessor()
    self._gps_handler = gps_handler or GPSHandler(self._processor)
    self._tolerance = tolerance or ToleranceSettings()

tolerance `property` `writable` ¶

tolerance

Get current tolerance settings.

verify ¶

verify(original, converted, *, tolerance=None, categories=None)

Run all verification checks on two files.

Parameters:

Name	Type	Description	Default
`original`	`Path`	Path to the original video file.	required
`converted`	`Path`	Path to the converted video file.	required
`tolerance`	`ToleranceSettings \| None`	Optional custom tolerance settings for this verification.	`None`
`categories`	`list[VerificationCategory] \| None`	Optional list of categories to verify. If None, verifies all.	`None`

Returns:

Type	Description
`VerificationResult`	VerificationResult with all check details.

Raises:

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

Source code in src/video_converter/processors/verification.py

def verify(
    self,
    original: Path,
    converted: Path,
    *,
    tolerance: ToleranceSettings | None = None,
    categories: list[VerificationCategory] | None = None,
) -> VerificationResult:
    """Run all verification checks on two files.

    Args:
        original: Path to the original video file.
        converted: Path to the converted video file.
        tolerance: Optional custom tolerance settings for this verification.
        categories: Optional list of categories to verify. If None, verifies all.

    Returns:
        VerificationResult with all check details.

    Raises:
        FileNotFoundError: If either file doesn't exist.
    """
    if not original.exists():
        raise FileNotFoundError(f"Original file not found: {original}")
    if not converted.exists():
        raise FileNotFoundError(f"Converted file not found: {converted}")

    tol = tolerance or self._tolerance
    cats = categories or list(VerificationCategory)

    checks: list[CheckResult] = []
    orig_meta = self._processor.extract(original)
    conv_meta = self._processor.extract(converted)

    if VerificationCategory.DATE_TIME in cats:
        checks.extend(self._verify_dates(orig_meta, conv_meta, tol))

    if VerificationCategory.GPS in cats:
        checks.extend(self._verify_gps(original, converted, tol))

    if VerificationCategory.CAMERA in cats:
        checks.extend(self._verify_camera(orig_meta, conv_meta))

    if VerificationCategory.VIDEO in cats:
        checks.extend(self._verify_video(orig_meta, conv_meta, tol))

    if VerificationCategory.AUDIO in cats:
        checks.extend(self._verify_audio(orig_meta, conv_meta))

    all_passed = all(c.passed for c in checks) if checks else True

    return VerificationResult(
        passed=all_passed,
        original_path=original,
        converted_path=converted,
        checks=checks,
        tolerance_settings=tol,
    )

Timestamp¶

timestamp ¶

File timestamp synchronization for video conversion.

This module provides functionality to synchronize filesystem timestamps (creation date, modification date) between original and converted video files. This ensures converted videos appear in the correct chronological order when sorted by date in macOS Finder.

SDS Reference: SDS-P01-004 SRS Reference: SRS-403 (Timestamp Preservation)

Example

from video_converter.processors.timestamp import TimestampSynchronizer synchronizer = TimestampSynchronizer()

Copy all timestamps from original to converted¶

result = synchronizer.sync_from_file( ... source=Path("original.mov"), ... dest=Path("converted.mp4"), ... ) if result.success: ... print("Timestamps synchronized successfully")

Verify timestamps were preserved¶

verification = synchronizer.verify( ... original=Path("original.mov"), ... converted=Path("converted.mp4"), ... ) print(f"Passed: {verification.passed}")

TimestampError ¶

TimestampError(path, reason)

Bases: Exception

Base exception for timestamp operations.

Attributes:

Name	Type	Description
`path`		The path where the error occurred.
`reason`		Description of the error.

Source code in src/video_converter/processors/timestamp.py

def __init__(self, path: Path, reason: str) -> None:
    self.path = path
    self.reason = reason
    super().__init__(f"Timestamp error for {path}: {reason}")

TimestampSyncError ¶

TimestampSyncError(path, reason)

Bases: TimestampError

Raised when timestamp synchronization fails.

Source code in src/video_converter/processors/timestamp.py

def __init__(self, path: Path, reason: str) -> None:
    self.path = path
    self.reason = reason
    super().__init__(f"Timestamp error for {path}: {reason}")

TimestampReadError ¶

TimestampReadError(path, reason)

Bases: TimestampError

Raised when reading timestamps fails.

Source code in src/video_converter/processors/timestamp.py

def __init__(self, path: Path, reason: str) -> None:
    self.path = path
    self.reason = reason
    super().__init__(f"Timestamp error for {path}: {reason}")

FileTimestamps `dataclass` ¶

FileTimestamps(path, birth_time=None, modification_time=None, access_time=None, metadata_change_time=None)

Container for file timestamp information.

Attributes:

Name	Type	Description
`path`	`Path`	Path to the file.
`birth_time`	`datetime \| None`	File creation time (macOS-specific, may be None on other OS).
`modification_time`	`datetime \| None`	Last modification time.
`access_time`	`datetime \| None`	Last access time.
`metadata_change_time`	`datetime \| None`	Last metadata change time (read-only on most systems).

Example

timestamps = FileTimestamps.from_file(Path("video.mp4")) print(f"Created: {timestamps.birth_time}") print(f"Modified: {timestamps.modification_time}")

from_file `classmethod` ¶

from_file(path)

Extract timestamps from a file.

Parameters:

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

Returns:

Type	Description
`FileTimestamps`	FileTimestamps with all available timestamp information.

Raises:

Type	Description
`FileNotFoundError`	If the file doesn't exist.
`TimestampReadError`	If reading timestamps fails.

Source code in src/video_converter/processors/timestamp.py

@classmethod
def from_file(cls, path: Path) -> FileTimestamps:
    """Extract timestamps from a file.

    Args:
        path: Path to the file.

    Returns:
        FileTimestamps with all available timestamp information.

    Raises:
        FileNotFoundError: If the file doesn't exist.
        TimestampReadError: If reading timestamps fails.
    """
    if not path.exists():
        raise FileNotFoundError(f"File not found: {path}")

    try:
        stat = path.stat()

        # Get modification and access times (always available)
        mtime = datetime.fromtimestamp(stat.st_mtime)
        atime = datetime.fromtimestamp(stat.st_atime)
        ctime = datetime.fromtimestamp(stat.st_ctime)

        # Get birth time (macOS-specific)
        birth_time = None
        if hasattr(stat, "st_birthtime"):
            birth_time = datetime.fromtimestamp(stat.st_birthtime)

        return cls(
            path=path,
            birth_time=birth_time,
            modification_time=mtime,
            access_time=atime,
            metadata_change_time=ctime,
        )
    except OSError as e:
        raise TimestampReadError(path, str(e)) from e

str ¶

__str__()

Return human-readable timestamp information.

Source code in src/video_converter/processors/timestamp.py

def __str__(self) -> str:
    """Return human-readable timestamp information."""
    lines = [f"Timestamps for {self.path.name}:"]
    if self.birth_time:
        lines.append(f"  Created:  {self.birth_time.isoformat()}")
    if self.modification_time:
        lines.append(f"  Modified: {self.modification_time.isoformat()}")
    if self.access_time:
        lines.append(f"  Accessed: {self.access_time.isoformat()}")
    return "\n".join(lines)

TimestampSyncResult `dataclass` ¶

TimestampSyncResult(success, source, dest, birth_time_synced=False, modification_time_synced=False, access_time_synced=False, error_message=None, warnings=list())

Result of a timestamp synchronization operation.

Attributes:

Name	Type	Description
`success`	`bool`	Whether synchronization completed successfully.
`source`	`Path`	Source file path.
`dest`	`Path`	Destination file path.
`birth_time_synced`	`bool`	Whether birth time was synchronized.
`modification_time_synced`	`bool`	Whether modification time was synchronized.
`access_time_synced`	`bool`	Whether access time was synchronized.
`error_message`	`str \| None`	Error message if synchronization failed.
`warnings`	`list[str]`	Non-critical issues encountered.

Example

result = synchronizer.sync_from_file(source, dest) if result.success: ... print("All timestamps synced") else: ... print(f"Error: {result.error_message}")

TimestampVerificationResult `dataclass` ¶

TimestampVerificationResult(passed, original=None, converted=None, birth_time_match=False, modification_time_match=False, tolerance_seconds=TIMESTAMP_TOLERANCE_SECONDS, details='')

Result of timestamp verification between two files.

Attributes:

Name	Type	Description
`passed`	`bool`	Whether timestamps match within tolerance.
`original`	`FileTimestamps \| None`	Timestamps from original file.
`converted`	`FileTimestamps \| None`	Timestamps from converted file.
`birth_time_match`	`bool`	Whether birth times match.
`modification_time_match`	`bool`	Whether modification times match.
`tolerance_seconds`	`float`	Tolerance used for comparison.
`details`	`str`	Additional verification details.

Example

result = synchronizer.verify(original, converted) if result.passed: ... print("Timestamps preserved correctly") else: ... print(result.details)

TimestampSynchronizer ¶

TimestampSynchronizer()

Synchronize filesystem timestamps between video files.

This class handles copying timestamps from original files to converted files, ensuring that Finder sorting by date works correctly. It supports macOS-specific features like birth time (creation date).

Attributes:

Name	Type	Description
`is_macos`		Whether running on macOS.

Example

synchronizer = TimestampSynchronizer()

Sync from original file¶

result = synchronizer.sync_from_file( ... source=Path("original.mov"), ... dest=Path("converted.mp4"), ... )

Or sync from specific datetime¶

result = synchronizer.sync_from_datetime( ... path=Path("converted.mp4"), ... creation_date=datetime(2023, 1, 15, 10, 30, 0), ... )

Initialize TimestampSynchronizer.

Source code in src/video_converter/processors/timestamp.py

def __init__(self) -> None:
    """Initialize TimestampSynchronizer."""
    self.is_macos = platform.system() == "Darwin"
    logger.debug("TimestampSynchronizer initialized (macOS: %s)", self.is_macos)

sync_from_file ¶

sync_from_file(source, dest, *, sync_birth_time=True, sync_modification_time=True, sync_access_time=True)

Copy timestamps from source file to destination file.

Copies creation date, modification date, and access time from the source file to the destination file.

Parameters:

Name	Type	Description	Default
`source`	`Path`	Source file to copy timestamps from.	required
`dest`	`Path`	Destination file to apply timestamps to.	required
`sync_birth_time`	`bool`	Whether to sync birth time (creation date).	`True`
`sync_modification_time`	`bool`	Whether to sync modification time.	`True`
`sync_access_time`	`bool`	Whether to sync access time.	`True`

Returns:

Type	Description
`TimestampSyncResult`	TimestampSyncResult with details of the synchronization.

Example

result = synchronizer.sync_from_file( ... source=Path("original.mov"), ... dest=Path("converted.mp4"), ... ) print(f"Success: {result.success}")

Source code in src/video_converter/processors/timestamp.py

def sync_from_file(
    self,
    source: Path,
    dest: Path,
    *,
    sync_birth_time: bool = True,
    sync_modification_time: bool = True,
    sync_access_time: bool = True,
) -> TimestampSyncResult:
    """Copy timestamps from source file to destination file.

    Copies creation date, modification date, and access time from the
    source file to the destination file.

    Args:
        source: Source file to copy timestamps from.
        dest: Destination file to apply timestamps to.
        sync_birth_time: Whether to sync birth time (creation date).
        sync_modification_time: Whether to sync modification time.
        sync_access_time: Whether to sync access time.

    Returns:
        TimestampSyncResult with details of the synchronization.

    Example:
        >>> result = synchronizer.sync_from_file(
        ...     source=Path("original.mov"),
        ...     dest=Path("converted.mp4"),
        ... )
        >>> print(f"Success: {result.success}")
    """
    logger.debug("Syncing timestamps from %s to %s", source, dest)

    # Validate files exist
    if not source.exists():
        return TimestampSyncResult(
            success=False,
            source=source,
            dest=dest,
            error_message=f"Source file not found: {source}",
        )

    if not dest.exists():
        return TimestampSyncResult(
            success=False,
            source=source,
            dest=dest,
            error_message=f"Destination file not found: {dest}",
        )

    try:
        # Get source timestamps
        source_timestamps = FileTimestamps.from_file(source)
    except TimestampReadError as e:
        return TimestampSyncResult(
            success=False,
            source=source,
            dest=dest,
            error_message=f"Failed to read source timestamps: {e.reason}",
        )

    warnings: list[str] = []
    birth_synced = False
    mtime_synced = False
    atime_synced = False

    # Sync modification and access times using os.utime
    if sync_modification_time or sync_access_time:
        try:
            stat = source.stat()
            atime = stat.st_atime if sync_access_time else dest.stat().st_atime
            mtime = stat.st_mtime if sync_modification_time else dest.stat().st_mtime
            os.utime(dest, (atime, mtime))
            mtime_synced = sync_modification_time
            atime_synced = sync_access_time
            logger.debug("Set mtime/atime on %s", dest)
        except OSError as e:
            warnings.append(f"Failed to set modification/access time: {e}")
            logger.warning("Failed to set mtime/atime on %s: %s", dest, e)

    # Sync birth time (macOS-specific)
    if sync_birth_time and source_timestamps.birth_time:
        birth_synced = self._set_birth_time(dest, source_timestamps.birth_time, warnings)

    success = (
        (not sync_modification_time or mtime_synced)
        and (not sync_access_time or atime_synced)
        and (not sync_birth_time or birth_synced or not self.is_macos)
    )

    result = TimestampSyncResult(
        success=success,
        source=source,
        dest=dest,
        birth_time_synced=birth_synced,
        modification_time_synced=mtime_synced,
        access_time_synced=atime_synced,
        warnings=warnings,
    )

    if success:
        logger.info("Timestamps synchronized: %s -> %s", source.name, dest.name)
    else:
        logger.warning(
            "Timestamp sync incomplete: %s -> %s (warnings: %s)",
            source.name,
            dest.name,
            warnings,
        )

    return result

sync_from_datetime ¶

sync_from_datetime(path, creation_date=None, modification_date=None, access_date=None)

Set file timestamps from datetime values.

Allows setting specific timestamps on a file without a source file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the file to modify.	required
`creation_date`	`datetime \| None`	Creation date to set (macOS only).	`None`
`modification_date`	`datetime \| None`	Modification date to set.	`None`
`access_date`	`datetime \| None`	Access date to set.	`None`

Returns:

Type	Description
`TimestampSyncResult`	TimestampSyncResult with details of the synchronization.

Example

from datetime import datetime result = synchronizer.sync_from_datetime( ... path=Path("video.mp4"), ... creation_date=datetime(2023, 1, 15, 10, 30, 0), ... modification_date=datetime(2023, 1, 15, 10, 30, 0), ... )

Source code in src/video_converter/processors/timestamp.py

def sync_from_datetime(
    self,
    path: Path,
    creation_date: datetime | None = None,
    modification_date: datetime | None = None,
    access_date: datetime | None = None,
) -> TimestampSyncResult:
    """Set file timestamps from datetime values.

    Allows setting specific timestamps on a file without a source file.

    Args:
        path: Path to the file to modify.
        creation_date: Creation date to set (macOS only).
        modification_date: Modification date to set.
        access_date: Access date to set.

    Returns:
        TimestampSyncResult with details of the synchronization.

    Example:
        >>> from datetime import datetime
        >>> result = synchronizer.sync_from_datetime(
        ...     path=Path("video.mp4"),
        ...     creation_date=datetime(2023, 1, 15, 10, 30, 0),
        ...     modification_date=datetime(2023, 1, 15, 10, 30, 0),
        ... )
    """
    logger.debug("Setting timestamps on %s from datetime values", path)

    if not path.exists():
        return TimestampSyncResult(
            success=False,
            source=path,
            dest=path,
            error_message=f"File not found: {path}",
        )

    warnings: list[str] = []
    birth_synced = False
    mtime_synced = False
    atime_synced = False

    # Set modification and access times
    if modification_date or access_date:
        try:
            current_stat = path.stat()
            atime = access_date.timestamp() if access_date else current_stat.st_atime
            mtime = (
                modification_date.timestamp() if modification_date else current_stat.st_mtime
            )
            os.utime(path, (atime, mtime))
            mtime_synced = modification_date is not None
            atime_synced = access_date is not None
            logger.debug("Set mtime/atime on %s", path)
        except OSError as e:
            warnings.append(f"Failed to set modification/access time: {e}")

    # Set birth time (macOS-specific)
    if creation_date:
        birth_synced = self._set_birth_time(path, creation_date, warnings)

    success = not warnings or (mtime_synced or atime_synced or birth_synced)

    return TimestampSyncResult(
        success=success,
        source=path,
        dest=path,
        birth_time_synced=birth_synced,
        modification_time_synced=mtime_synced,
        access_time_synced=atime_synced,
        warnings=warnings,
    )

get_timestamps ¶

get_timestamps(path)

Get all timestamps from a file.

Parameters:

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

Returns:

Type	Description
`FileTimestamps`	FileTimestamps with all available timestamp information.

Raises:

Type	Description
`FileNotFoundError`	If the file doesn't exist.
`TimestampReadError`	If reading timestamps fails.

Source code in src/video_converter/processors/timestamp.py

def get_timestamps(self, path: Path) -> FileTimestamps:
    """Get all timestamps from a file.

    Args:
        path: Path to the file.

    Returns:
        FileTimestamps with all available timestamp information.

    Raises:
        FileNotFoundError: If the file doesn't exist.
        TimestampReadError: If reading timestamps fails.
    """
    return FileTimestamps.from_file(path)

verify ¶

verify(original, converted, *, tolerance_seconds=TIMESTAMP_TOLERANCE_SECONDS, check_birth_time=True, check_modification_time=True)

Verify timestamps were preserved during conversion.

Compares timestamps between original and converted files to ensure they match within the specified tolerance.

Parameters:

Name	Type	Description	Default
`original`	`Path`	Original source file.	required
`converted`	`Path`	Converted output file.	required
`tolerance_seconds`	`float`	Maximum allowed difference in seconds.	`TIMESTAMP_TOLERANCE_SECONDS`
`check_birth_time`	`bool`	Whether to check birth time match.	`True`
`check_modification_time`	`bool`	Whether to check modification time match.	`True`

Returns:

Type	Description
`TimestampVerificationResult`	TimestampVerificationResult with detailed comparison.

Raises:

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

Example

result = synchronizer.verify( ... original=Path("original.mov"), ... converted=Path("converted.mp4"), ... ) if result.passed: ... print("Timestamps preserved correctly")

Source code in src/video_converter/processors/timestamp.py

def verify(
    self,
    original: Path,
    converted: Path,
    *,
    tolerance_seconds: float = TIMESTAMP_TOLERANCE_SECONDS,
    check_birth_time: bool = True,
    check_modification_time: bool = True,
) -> TimestampVerificationResult:
    """Verify timestamps were preserved during conversion.

    Compares timestamps between original and converted files to ensure
    they match within the specified tolerance.

    Args:
        original: Original source file.
        converted: Converted output file.
        tolerance_seconds: Maximum allowed difference in seconds.
        check_birth_time: Whether to check birth time match.
        check_modification_time: Whether to check modification time match.

    Returns:
        TimestampVerificationResult with detailed comparison.

    Raises:
        FileNotFoundError: If either file doesn't exist.

    Example:
        >>> result = synchronizer.verify(
        ...     original=Path("original.mov"),
        ...     converted=Path("converted.mp4"),
        ... )
        >>> if result.passed:
        ...     print("Timestamps preserved correctly")
    """
    if not original.exists():
        raise FileNotFoundError(f"Original file not found: {original}")
    if not converted.exists():
        raise FileNotFoundError(f"Converted file not found: {converted}")

    try:
        orig_timestamps = FileTimestamps.from_file(original)
    except TimestampReadError as e:
        return TimestampVerificationResult(
            passed=False,
            tolerance_seconds=tolerance_seconds,
            details=f"Failed to read original timestamps: {e.reason}",
        )

    try:
        conv_timestamps = FileTimestamps.from_file(converted)
    except TimestampReadError as e:
        return TimestampVerificationResult(
            passed=False,
            original=orig_timestamps,
            tolerance_seconds=tolerance_seconds,
            details=f"Failed to read converted timestamps: {e.reason}",
        )

    birth_match = True
    mtime_match = True
    details_parts: list[str] = []

    # Check birth time
    if check_birth_time and orig_timestamps.birth_time:
        if conv_timestamps.birth_time:
            diff = abs(
                (orig_timestamps.birth_time - conv_timestamps.birth_time).total_seconds()
            )
            birth_match = diff <= tolerance_seconds
            if birth_match:
                details_parts.append(f"Birth time match (diff: {diff:.2f}s)")
            else:
                details_parts.append(
                    f"Birth time mismatch: {orig_timestamps.birth_time} vs "
                    f"{conv_timestamps.birth_time} (diff: {diff:.2f}s)"
                )
        else:
            birth_match = False
            details_parts.append("Birth time missing in converted file")

    # Check modification time
    if check_modification_time:
        if orig_timestamps.modification_time and conv_timestamps.modification_time:
            diff = abs(
                (
                    orig_timestamps.modification_time - conv_timestamps.modification_time
                ).total_seconds()
            )
            mtime_match = diff <= tolerance_seconds
            if mtime_match:
                details_parts.append(f"Modification time match (diff: {diff:.2f}s)")
            else:
                details_parts.append(
                    f"Modification time mismatch: {orig_timestamps.modification_time} "
                    f"vs {conv_timestamps.modification_time} (diff: {diff:.2f}s)"
                )
        elif orig_timestamps.modification_time:
            mtime_match = False
            details_parts.append("Modification time missing in converted file")

    passed = (not check_birth_time or birth_match or not self.is_macos) and (
        not check_modification_time or mtime_match
    )

    return TimestampVerificationResult(
        passed=passed,
        original=orig_timestamps,
        converted=conv_timestamps,
        birth_time_match=birth_match,
        modification_time_match=mtime_match,
        tolerance_seconds=tolerance_seconds,
        details="; ".join(details_parts) if details_parts else "All checks passed",
    )

Processors Module¶

Codec Detector¶

codec_detector ¶

InvalidVideoError ¶

CorruptedVideoError ¶

UnsupportedCodecError ¶

CodecInfo dataclass ¶

is_h264 property ¶

is_hevc property ¶

needs_conversion property ¶

resolution_label property ¶

aspect_ratio property ¶

bitrate_mbps property ¶

size_mb property ¶

size_gb property ¶

__str__ ¶

CodecDetector ¶

Batch analysis¶

analyze ¶

analyze_async async ¶

is_h264 ¶

is_hevc ¶

needs_conversion ¶

get_codec ¶

Metadata Processor¶

metadata ¶

Copy all metadata¶

Verify critical tags¶

MetadataExtractionError ¶

MetadataApplicationError ¶

GPSCoordinates dataclass ¶

__str__ ¶

MetadataVerificationResult dataclass ¶

MetadataProcessor ¶

is_available ¶

ensure_available ¶

extract ¶

extract_gps ¶

copy_all ¶

copy_tags ¶

copy_gps ¶

copy_dates ¶

verify_critical_tags ¶

set_tag ¶

batch_copy_all ¶

Quality Validator¶

quality_validator ¶

Compression ratio validation¶

ValidationStrictness ¶

StreamInfo dataclass ¶

VideoInfo dataclass ¶

video_streams property ¶

audio_streams property ¶

has_video property ¶

has_audio property ¶

primary_video_stream property ¶

primary_audio_stream property ¶

ValidationResult dataclass ¶

add_error ¶

add_warning ¶

VideoValidator ¶

Quick validation¶

Full validation¶

quick_validate ¶

validate ¶

validate_async async ¶

ComparisonSeverity ¶

PropertyComparison dataclass ¶

PropertyComparisonResult dataclass ¶

add_comparison ¶

PropertyComparer ¶

compare ¶

ContentType ¶

CompressionRange dataclass ¶

__post_init__ ¶

CompressionSeverity ¶

CompressionValidationResult dataclass ¶

size_reduction_percent property ¶

file_grew property ¶

CompressionValidator ¶

CodecInfo `dataclass` ¶

is_h264 `property` ¶

is_hevc `property` ¶

needs_conversion `property` ¶

resolution_label `property` ¶

aspect_ratio `property` ¶

bitrate_mbps `property` ¶

size_mb `property` ¶

size_gb `property` ¶

str ¶

analyze_async `async` ¶

GPSCoordinates `dataclass` ¶

str ¶

MetadataVerificationResult `dataclass` ¶

StreamInfo `dataclass` ¶

VideoInfo `dataclass` ¶

video_streams `property` ¶

audio_streams `property` ¶

has_video `property` ¶

has_audio `property` ¶

primary_video_stream `property` ¶

primary_audio_stream `property` ¶

ValidationResult `dataclass` ¶

validate_async `async` ¶

PropertyComparison `dataclass` ¶

PropertyComparisonResult `dataclass` ¶

CompressionRange `dataclass` ¶

CompressionValidationResult `dataclass` ¶

size_reduction_percent `property` ¶

file_grew `property` ¶

GPSCoordinates `dataclass` ¶

str ¶

from_quicktime `classmethod` ¶

from_xmp `classmethod` ¶

from_exif_dms `classmethod` ¶

GPSVerificationResult `dataclass` ¶

from_score `classmethod` ¶

VmafScores `dataclass` ¶

quality_level `property` ¶

str ¶

VmafResult `dataclass` ¶

is_visually_lossless `property` ¶

is_acceptable `property` ¶

str ¶

analyze_async `async` ¶

ToleranceSettings `dataclass` ¶

strict `classmethod` ¶

relaxed `classmethod` ¶

CheckResult `dataclass` ¶

passed `property` ¶

str ¶

VerificationResult `dataclass` ¶

failed_checks `property` ¶

passed_checks `property` ¶

checks_by_category `property` ¶

tolerance `property` `writable` ¶