Core Module

Core components for orchestration and configuration.

Orchestrator

orchestrator

Orchestrator for video conversion workflow.

This module implements the central Orchestrator class that coordinates the entire video conversion workflow from discovery to completion.

SDS Reference: SDS-C01-002 SRS Reference: SRS-601 (Orchestrator Workflow)

Example

from video_converter.core.orchestrator import Orchestrator from video_converter.core import ConversionRequest

orchestrator = Orchestrator()

Convert a single file

result = await orchestrator.convert_single( ... input_path=Path("input.mov"), ... output_path=Path("output.mp4"), ... )

Run batch conversion with progress tracking

def on_progress(progress): ... print(f"Progress: {progress.overall_progress:.1%}")

report = await orchestrator.run( ... input_paths=[Path("video1.mov"), Path("video2.mov")], ... output_dir=Path("converted"), ... on_progress=on_progress, ... ) print(f"Converted: {report.successful}/{report.total_files}")

OrchestratorConfig dataclass

OrchestratorConfig(mode=HARDWARE, quality=DEFAULT_QUALITY, crf=DEFAULT_CRF, preset='medium', output_suffix='_h265', preserve_metadata=True, preserve_timestamps=True, validate_output=True, validation_strictness=STANDARD, max_concurrent=DEFAULT_CONCURRENT_CONVERSIONS, delete_original=False, move_to_processed=None, move_to_failed=None, queue_priority=FIFO, enable_retry=True, retry_config=None, check_disk_space=True, min_free_space=MIN_FREE_DISK_SPACE, pause_on_disk_full=True, enable_notifications=True, notification_config=None, auto_download_icloud=True, icloud_timeout=ICLOUD_DOWNLOAD_TIMEOUT, skip_icloud_on_timeout=True, enable_vmaf=False, vmaf_threshold=VMAF_THRESHOLD_VISUALLY_LOSSLESS, vmaf_sample_interval=VMAF_DEFAULT_SAMPLE_INTERVAL, vmaf_fail_action='warn')

Configuration for the Orchestrator.

Attributes:

Name	Type	Description
mode	ConversionMode	Preferred conversion mode (hardware or software).
quality	int	Quality setting for encoding (1-100).
crf	int	Constant Rate Factor for software encoding.
preset	str	Encoding preset for software encoding.
output_suffix	str	Suffix to add to output filenames.
preserve_metadata	bool	Whether to copy metadata from original.
preserve_timestamps	bool	Whether to sync timestamps from original.
validate_output	bool	Whether to validate converted files.
validation_strictness	ValidationStrictness	How strictly to validate output.
max_concurrent	int	Maximum concurrent conversions.
delete_original	bool	Whether to delete original after success.
move_to_processed	Path | None	Directory to move processed originals.
move_to_failed	Path | None	Directory to move failed files.
queue_priority	QueuePriority	Priority ordering for batch queue.
enable_retry	bool	Whether to retry failed conversions.
retry_config	RetryConfig | None	Configuration for retry behavior.
check_disk_space	bool	Whether to check disk space before processing.
min_free_space	int	Minimum free disk space in bytes (default 1GB).
pause_on_disk_full	bool	Whether to pause on low disk space.
enable_notifications	bool	Whether to send macOS notifications.
notification_config	NotificationConfig | None	Configuration for notification behavior.
auto_download_icloud	bool	Whether to auto-download iCloud files for folders.
icloud_timeout	int	Maximum time to wait for iCloud downloads in seconds.
skip_icloud_on_timeout	bool	Whether to skip files if download times out.
enable_vmaf	bool	Whether to measure VMAF quality score after conversion.
vmaf_threshold	float	Minimum acceptable VMAF score (default 93.0 for visually lossless).
vmaf_sample_interval	int	Frame sampling interval for VMAF analysis (1=all, 30=faster).
vmaf_fail_action	str	Action when VMAF is below threshold ("warn", "retry", "fail").

ConversionTask dataclass

ConversionTask(input_path, output_path, status=PENDING, result=None, error=None)

A single video conversion task in the queue.

Attributes:

Name	Type	Description
input_path	Path	Path to input video.
output_path	Path	Path for output video.
status	ConversionStatus	Current status of the task.
result	ConversionResult | None	Conversion result (when complete).
error	str | None	Error message (if failed).

Orchestrator

Orchestrator(config=None, converter_factory=None, validator=None, timestamp_synchronizer=None, session_manager=None, enable_session_persistence=True, retry_manager=None, error_recovery_manager=None)

Coordinates the video conversion workflow.

The Orchestrator manages the entire conversion pipeline: 1. Discovery - Finding videos to convert 2. Queueing - Managing the conversion queue 3. Converting - Running the encoder 4. Validating - Verifying output integrity 5. Cleanup - Moving/deleting files 6. Reporting - Generating conversion reports

Attributes:

Name	Type	Description
config		Orchestrator configuration.
converter_factory		Factory for creating converters.
validator		Video file validator.
timestamp_synchronizer		Timestamp synchronization handler.
session_manager	SessionStateManager | None	Session state manager for persistence.

Initialize the Orchestrator.

Parameters:

Name	Type	Description	Default
config	OrchestratorConfig | None	Optional configuration. Uses defaults if not provided.	None
converter_factory	ConverterFactory | None	Optional converter factory.	None
validator	VideoValidator | None	Optional video validator.	None
timestamp_synchronizer	TimestampSynchronizer | None	Optional timestamp synchronizer.	None
session_manager	SessionStateManager | None	Optional session state manager.	None
enable_session_persistence	bool	Whether to enable session persistence.	True
retry_manager	RetryManager | None	Optional retry manager for failed conversions.	None
error_recovery_manager	ErrorRecoveryManager | None	Optional error recovery manager.	None

Source code in src/video_converter/core/orchestrator.py

def __init__(
    self,
    config: OrchestratorConfig | None = None,
    converter_factory: ConverterFactory | None = None,
    validator: VideoValidator | None = None,
    timestamp_synchronizer: TimestampSynchronizer | None = None,
    session_manager: SessionStateManager | None = None,
    enable_session_persistence: bool = True,
    retry_manager: RetryManager | None = None,
    error_recovery_manager: ErrorRecoveryManager | None = None,
) -> None:
    """Initialize the Orchestrator.

    Args:
        config: Optional configuration. Uses defaults if not provided.
        converter_factory: Optional converter factory.
        validator: Optional video validator.
        timestamp_synchronizer: Optional timestamp synchronizer.
        session_manager: Optional session state manager.
        enable_session_persistence: Whether to enable session persistence.
        retry_manager: Optional retry manager for failed conversions.
        error_recovery_manager: Optional error recovery manager.
    """
    self.config = config or OrchestratorConfig()
    self.converter_factory = converter_factory or ConverterFactory()
    self.validator = validator or VideoValidator()
    self.timestamp_synchronizer = timestamp_synchronizer or TimestampSynchronizer()
    self._enable_session_persistence = enable_session_persistence

    self.session_manager: SessionStateManager | None
    if enable_session_persistence:
        self.session_manager = session_manager or SessionStateManager()
    else:
        self.session_manager = None

    self.retry_manager: RetryManager | None
    if self.config.enable_retry:
        retry_config = self.config.retry_config or RetryConfig()
        self.retry_manager = retry_manager or RetryManager(retry_config)
    else:
        self.retry_manager = None

    # Initialize error recovery manager
    self.error_recovery_manager = error_recovery_manager or ErrorRecoveryManager(
        failed_dir=self.config.move_to_failed,
        min_free_space=self.config.min_free_space,
    )

    self._converter: BaseConverter | None = None
    self._cancelled = False
    self._paused = False
    self._paused_reason: str | None = None
    self._pause_event = asyncio.Event()
    self._pause_event.set()  # Not paused initially
    self._batch_status = BatchStatus.IDLE
    self._current_session_id: str | None = None
    self._tasks: list[ConversionTask] = []
    self._current_session: SessionState | None = None

    # Concurrent processing support
    self._concurrent_processor = ConcurrentProcessor(
        max_concurrent=self.config.max_concurrent,
        enable_resource_monitoring=True,
        adaptive_concurrency=False,
    )

    # Notification manager
    if self.config.enable_notifications:
        notification_config = self.config.notification_config or NotificationConfig()
        self._notification_manager: NotificationManager | None = NotificationManager(
            config=notification_config
        )
    else:
        self._notification_manager = None

    # VMAF analyzer (lazy initialization for availability check)
    self._vmaf_analyzer: VmafAnalyzer | None = None
    if self.config.enable_vmaf:
        self._vmaf_analyzer = VmafAnalyzer()
        if not self._vmaf_analyzer.is_available():
            logger.warning(
                "VMAF measurement requested but libvmaf is not available. "
                "VMAF analysis will be skipped."
            )
            self._vmaf_analyzer = None

convert_single async

convert_single(input_path, output_path=None, on_progress=None, video_info=None, on_progress_info=None)

Convert a single video file through the full pipeline.

Handles both local files and iCloud files. For iCloud files, automatically downloads before conversion if auto_download_icloud is enabled.

Parameters:

Name	Type	Description	Default
input_path	Path	Path to the input video.	required
output_path	Path | None	Path for the output video. Auto-generated if None.	None
on_progress	ProgressCallback | None	Optional progress callback for pipeline stage updates.	None
video_info	FolderVideoInfo | None	Optional FolderVideoInfo with iCloud status.	None
on_progress_info	Callable[[ProgressInfo], None] | None	Optional progress callback for detailed FFmpeg progress (percentage, speed, current size, ETA).	None

Returns:

Type	Description
ConversionResult	ConversionResult with success status and statistics.

Source code in src/video_converter/core/orchestrator.py

async def convert_single(
    self,
    input_path: Path,
    output_path: Path | None = None,
    on_progress: ProgressCallback | None = None,
    video_info: FolderVideoInfo | None = None,
    on_progress_info: Callable[[ProgressInfo], None] | None = None,
) -> ConversionResult:
    """Convert a single video file through the full pipeline.

    Handles both local files and iCloud files. For iCloud files,
    automatically downloads before conversion if auto_download_icloud is enabled.

    Args:
        input_path: Path to the input video.
        output_path: Path for the output video. Auto-generated if None.
        on_progress: Optional progress callback for pipeline stage updates.
        video_info: Optional FolderVideoInfo with iCloud status.
        on_progress_info: Optional progress callback for detailed FFmpeg progress
            (percentage, speed, current size, ETA).

    Returns:
        ConversionResult with success status and statistics.
    """
    self._cancelled = False

    # Generate output path if needed (before validation)
    if output_path is None:
        output_path = self._create_output_path(input_path)

    # Ensure file is available (handles iCloud download)
    available, error = await self._ensure_file_available(input_path, video_info, on_progress)

    if not available:
        return ConversionResult(
            success=False,
            request=ConversionRequest(
                input_path=input_path,
                output_path=output_path,
            ),
            error_message=error or f"Input file not available: {input_path}",
        )

    # Validate input (file should exist after download)
    if not input_path.exists():
        return ConversionResult(
            success=False,
            request=ConversionRequest(
                input_path=input_path,
                output_path=output_path,
            ),
            error_message=f"Input file not found after download: {input_path}",
        )

    # Create conversion request
    request = ConversionRequest(
        input_path=input_path,
        output_path=output_path,
        mode=self.config.mode,
        quality=self.config.quality,
        crf=self.config.crf,
        preset=self.config.preset,
        preserve_metadata=self.config.preserve_metadata,
    )

    # Stage 1: Convert
    self._emit_progress(
        on_progress,
        ConversionStage.CONVERT,
        ConversionStatus.IN_PROGRESS,
        current_file=input_path.name,
        message=f"Converting {input_path.name}...",
    )

    try:
        converter = self._get_converter()
    except EncoderNotAvailableError as e:
        return ConversionResult(
            success=False,
            request=request,
            error_message=str(e),
        )

    result = await converter.convert(request, on_progress_info=on_progress_info)

    if not result.success:
        if self.retry_manager:
            return await self._retry_conversion(request, result, on_progress, input_path)
        return result

    if self._cancelled:
        # Clean up output file
        if output_path.exists():
            output_path.unlink()
        result.success = False
        result.error_message = "Conversion cancelled"
        return result

    # Stage 2: Validate
    if self.config.validate_output:
        self._emit_progress(
            on_progress,
            ConversionStage.VALIDATE,
            ConversionStatus.IN_PROGRESS,
            current_file=input_path.name,
            message=f"Validating {output_path.name}...",
        )

        validation = self.validator.validate(
            output_path,
            strictness=self.config.validation_strictness,
        )

        if not validation.valid:
            # Validation failed - try retry if enabled
            if self.retry_manager:
                if output_path.exists():
                    output_path.unlink()
                result.success = False
                result.error_message = f"Validation failed: {', '.join(validation.errors)}"
                result.warnings.extend(validation.warnings)
                return await self._retry_conversion(request, result, on_progress, input_path)

            # No retry - clean up and report
            if output_path.exists():
                output_path.unlink()
            result.success = False
            result.error_message = f"Validation failed: {', '.join(validation.errors)}"
            result.warnings.extend(validation.warnings)
            return result

        result.warnings.extend(validation.warnings)

    # VMAF quality analysis (after validation, before metadata sync)
    if self._vmaf_analyzer and output_path.exists():
        try:
            logger.info(f"Running VMAF analysis for {output_path.name}...")
            vmaf_result = await self._vmaf_analyzer.analyze_async(
                original=input_path,
                converted=output_path,
                sample_interval=self.config.vmaf_sample_interval,
            )
            result.vmaf_score = vmaf_result.scores.mean
            result.vmaf_quality_level = vmaf_result.quality_level.value

            # Add VMAF warnings to result
            result.warnings.extend(vmaf_result.warnings)

            logger.info(
                f"VMAF score: {vmaf_result.scores.mean:.2f} ({vmaf_result.quality_level.value})"
            )

            # Handle VMAF threshold check
            if vmaf_result.scores.mean < self.config.vmaf_threshold:
                vmaf_warning = (
                    f"VMAF score {vmaf_result.scores.mean:.2f} is below "
                    f"threshold {self.config.vmaf_threshold:.1f}"
                )

                if self.config.vmaf_fail_action == "fail":
                    # Delete output and mark as failed
                    if output_path.exists():
                        output_path.unlink()
                    result.success = False
                    result.error_message = vmaf_warning
                    return result

                elif self.config.vmaf_fail_action == "retry" and self.retry_manager:
                    # Attempt retry with adjusted settings
                    if output_path.exists():
                        output_path.unlink()
                    result.success = False
                    result.error_message = vmaf_warning
                    return await self._retry_conversion(
                        request, result, on_progress, input_path
                    )

                else:  # "warn" (default)
                    result.warnings.append(vmaf_warning)

        except VmafAnalysisError as e:
            logger.warning(f"VMAF analysis failed: {e}")
            result.warnings.append(f"VMAF analysis failed: {e.reason}")
        except VmafNotAvailableError:
            logger.warning("VMAF not available, skipping quality measurement")
        except Exception as e:
            logger.warning(f"Unexpected error during VMAF analysis: {e}")
            result.warnings.append(f"VMAF analysis error: {e}")

    # Stage 3: Metadata - Sync timestamps
    if self.config.preserve_timestamps:
        self._emit_progress(
            on_progress,
            ConversionStage.METADATA,
            ConversionStatus.IN_PROGRESS,
            current_file=input_path.name,
            message=f"Syncing timestamps for {output_path.name}...",
        )

        timestamp_result = self.timestamp_synchronizer.sync_from_file(
            source=input_path,
            dest=output_path,
        )

        if not timestamp_result.success:
            result.warnings.append(f"Timestamp sync incomplete: {timestamp_result.warnings}")
        elif timestamp_result.warnings:
            result.warnings.extend(timestamp_result.warnings)

        logger.debug(
            "Timestamp sync result: birth=%s, mtime=%s, atime=%s",
            timestamp_result.birth_time_synced,
            timestamp_result.modification_time_synced,
            timestamp_result.access_time_synced,
        )

    # Stage 4: Cleanup
    self._emit_progress(
        on_progress,
        ConversionStage.CLEANUP,
        ConversionStatus.IN_PROGRESS,
        current_file=input_path.name,
        message="Cleaning up...",
    )

    if self.config.delete_original:
        try:
            input_path.unlink()
            logger.info(f"Deleted original: {input_path}")
        except OSError as e:
            result.warnings.append(f"Could not delete original: {e}")

    elif self.config.move_to_processed:
        try:
            dest = self.config.move_to_processed / input_path.name
            self.config.move_to_processed.mkdir(parents=True, exist_ok=True)
            input_path.rename(dest)
            logger.info(f"Moved original to: {dest}")
        except OSError as e:
            result.warnings.append(f"Could not move original: {e}")

    # Complete
    self._emit_progress(
        on_progress,
        ConversionStage.COMPLETE,
        ConversionStatus.COMPLETED,
        current_file=input_path.name,
        stage_progress=1.0,
        message="Conversion complete",
    )

    return result

run async

run(input_paths, output_dir=None, on_progress=None, on_complete=None, resume_session=False)

Run batch conversion on multiple files.

Parameters:

Name	Type	Description	Default
input_paths	list[Path]	List of input video paths.	required
output_dir	Path | None	Directory for output files. Uses original dirs if None.	None
on_progress	ProgressCallback | None	Optional progress callback.	None
on_complete	CompleteCallback | None	Optional completion callback.	None
resume_session	bool	If True, try to resume an interrupted session.	False

Returns:

Type	Description
ConversionReport	ConversionReport with batch statistics.

Source code in src/video_converter/core/orchestrator.py

async def run(
    self,
    input_paths: list[Path],
    output_dir: Path | None = None,
    on_progress: ProgressCallback | None = None,
    on_complete: CompleteCallback | None = None,
    resume_session: bool = False,
) -> ConversionReport:
    """Run batch conversion on multiple files.

    Args:
        input_paths: List of input video paths.
        output_dir: Directory for output files. Uses original dirs if None.
        on_progress: Optional progress callback.
        on_complete: Optional completion callback.
        resume_session: If True, try to resume an interrupted session.

    Returns:
        ConversionReport with batch statistics.
    """
    self._cancelled = False
    self._paused = False
    self._pause_event.set()
    self._batch_status = BatchStatus.RUNNING
    self._current_session_id = self._generate_session_id()

    report = ConversionReport(
        session_id=self._current_session_id,
        started_at=datetime.now(),
        total_files=len(input_paths),
    )

    if not input_paths:
        report.completed_at = datetime.now()
        self._batch_status = BatchStatus.COMPLETED
        if on_complete:
            on_complete(report)
        return report

    # Sort input paths by priority
    sorted_paths = self._sort_by_priority(list(input_paths))

    # Stage 1: Discovery - Build task queue
    self._emit_progress(
        on_progress,
        ConversionStage.DISCOVERY,
        ConversionStatus.IN_PROGRESS,
        total_files=len(sorted_paths),
        message="Discovering videos...",
    )

    self._tasks = []
    for input_path in sorted_paths:
        output_path = self._create_output_path(input_path, output_dir)

        # Skip if output already exists
        if output_path.exists():
            logger.info(f"Skipping (output exists): {input_path.name}")
            report.skipped += 1
            continue

        self._tasks.append(
            ConversionTask(
                input_path=input_path,
                output_path=output_path,
            )
        )

    if not self._tasks:
        report.completed_at = datetime.now()
        if on_complete:
            on_complete(report)
        return report

    # Create session state for persistence
    if self.session_manager:
        self._current_session = self.session_manager.create_session(
            video_paths=[t.input_path for t in self._tasks],
            output_dir=output_dir,
            config=self.config,
        )
        self._current_session_id = self._current_session.session_id
        report.session_id = self._current_session_id

    # Stage 2: Process queue
    total_tasks = len(self._tasks)

    # Use concurrent processing if max_concurrent > 1
    if self.config.max_concurrent > 1:
        await self._process_tasks_concurrent(report, on_progress, total_tasks)
    else:
        await self._process_tasks_sequential(report, on_progress, total_tasks)

    # Complete
    report.completed_at = datetime.now()
    if not self._cancelled:
        self._batch_status = BatchStatus.COMPLETED
        if self.session_manager:
            self.session_manager.complete_session()

    self._emit_progress(
        on_progress,
        ConversionStage.COMPLETE,
        ConversionStatus.COMPLETED,
        total_files=total_tasks,
        stage_progress=1.0,
        message=(
            f"Completed: {report.successful} succeeded, "
            f"{report.failed} failed, {report.skipped} skipped"
        ),
    )

    # Send notification
    self._send_batch_notification(report)

    if on_complete:
        on_complete(report)

    return report

resume_session async

resume_session(on_progress=None, on_complete=None)

Resume an interrupted or paused session from disk.

This method loads a previously saved session and continues processing any pending videos.

Parameters:

Name	Type	Description	Default
on_progress	ProgressCallback | None	Optional progress callback.	None
on_complete	CompleteCallback | None	Optional completion callback.	None

Returns:

Type	Description
ConversionReport | None	ConversionReport if resumed successfully, None if no session to resume.

Source code in src/video_converter/core/orchestrator.py

async def resume_session(
    self,
    on_progress: ProgressCallback | None = None,
    on_complete: CompleteCallback | None = None,
) -> ConversionReport | None:
    """Resume an interrupted or paused session from disk.

    This method loads a previously saved session and continues
    processing any pending videos.

    Args:
        on_progress: Optional progress callback.
        on_complete: Optional completion callback.

    Returns:
        ConversionReport if resumed successfully, None if no session to resume.
    """
    if not self.session_manager:
        logger.warning("Session persistence is disabled, cannot resume")
        return None

    session = self.session_manager.resume_session()
    if session is None:
        logger.info("No resumable session found")
        return None

    self._current_session = session
    self._current_session_id = session.session_id

    logger.info(
        f"Resuming session {session.session_id} with "
        f"{len(session.pending_videos)} pending videos"
    )

    # Get pending video paths
    pending_paths = [v.path for v in session.pending_videos]

    if not pending_paths:
        logger.info("No pending videos to process")
        self.session_manager.complete_session()
        return ConversionReport(
            session_id=session.session_id,
            started_at=session.started_at,
            completed_at=datetime.now(),
            total_files=session.total_videos,
            successful=len(session.completed_videos),
            failed=len(session.failed_videos),
        )

    return await self.run(
        input_paths=pending_paths,
        output_dir=session.output_dir,
        on_progress=on_progress,
        on_complete=on_complete,
    )

has_resumable_session

has_resumable_session()

Check if there is a resumable session available.

Returns:

Type	Description
bool	True if a session can be resumed, False otherwise.

Source code in src/video_converter/core/orchestrator.py

def has_resumable_session(self) -> bool:
    """Check if there is a resumable session available.

    Returns:
        True if a session can be resumed, False otherwise.
    """
    if not self.session_manager:
        return False
    return self.session_manager.has_resumable_session()

get_session_status

get_session_status()

Get the current session status.

Returns:

Type	Description
dict | None	Session status dictionary, or None if no session.

Source code in src/video_converter/core/orchestrator.py

def get_session_status(self) -> dict | None:
    """Get the current session status.

    Returns:
        Session status dictionary, or None if no session.
    """
    if not self.session_manager:
        return None
    return self.session_manager.get_session_status()

run_directory async

run_directory(input_dir, output_dir=None, recursive=False, on_progress=None, on_complete=None)

Run batch conversion on a directory.

Parameters:

Name	Type	Description	Default
input_dir	Path	Directory containing videos.	required
output_dir	Path | None	Directory for output files.	None
recursive	bool	Whether to search subdirectories.	False
on_progress	ProgressCallback | None	Optional progress callback.	None
on_complete	CompleteCallback | None	Optional completion callback.	None

Returns:

Type	Description
ConversionReport	ConversionReport with batch statistics.

Source code in src/video_converter/core/orchestrator.py

async def run_directory(
    self,
    input_dir: Path,
    output_dir: Path | None = None,
    recursive: bool = False,
    on_progress: ProgressCallback | None = None,
    on_complete: CompleteCallback | None = None,
) -> ConversionReport:
    """Run batch conversion on a directory.

    Args:
        input_dir: Directory containing videos.
        output_dir: Directory for output files.
        recursive: Whether to search subdirectories.
        on_progress: Optional progress callback.
        on_complete: Optional completion callback.

    Returns:
        ConversionReport with batch statistics.
    """
    # Discover video files
    video_files: list[Path] = []

    if recursive:
        for ext in VIDEO_EXTENSIONS:
            video_files.extend(input_dir.rglob(f"*{ext}"))
    else:
        for ext in VIDEO_EXTENSIONS:
            video_files.extend(input_dir.glob(f"*{ext}"))

    # Sort by name for consistent ordering
    video_files.sort()

    return await self.run(
        input_paths=video_files,
        output_dir=output_dir,
        on_progress=on_progress,
        on_complete=on_complete,
    )

cancel

cancel()

Cancel the current conversion operation.

Source code in src/video_converter/core/orchestrator.py

def cancel(self) -> None:
    """Cancel the current conversion operation."""
    self._cancelled = True
    self._batch_status = BatchStatus.CANCELLED
    if self._converter:
        self._converter.cancel()
    # Also resume if paused, so the loop can exit
    self._pause_event.set()
    logger.info("Conversion cancelled by user")

pause

pause()

Pause the batch conversion.

Pauses after the current file completes.

Returns:

Type	Description
bool	True if paused, False if not running.

Source code in src/video_converter/core/orchestrator.py

def pause(self) -> bool:
    """Pause the batch conversion.

    Pauses after the current file completes.

    Returns:
        True if paused, False if not running.
    """
    if self._batch_status != BatchStatus.RUNNING:
        return False

    self._paused = True
    self._pause_event.clear()
    self._batch_status = BatchStatus.PAUSED

    # Update session state
    if self.session_manager:
        self.session_manager.pause_session()

    logger.info("Batch conversion paused")
    return True

resume

resume()

Resume a paused batch conversion.

This resumes a batch that was paused in the current session. To resume a saved session from disk, use resume_session() instead.

Returns:

Type	Description
bool	True if resumed, False if not paused.

Source code in src/video_converter/core/orchestrator.py

def resume(self) -> bool:
    """Resume a paused batch conversion.

    This resumes a batch that was paused in the current session.
    To resume a saved session from disk, use resume_session() instead.

    Returns:
        True if resumed, False if not paused.
    """
    if self._batch_status != BatchStatus.PAUSED:
        return False

    self._paused = False
    self._pause_event.set()
    self._batch_status = BatchStatus.RUNNING

    # Update session state
    if self.session_manager and self._current_session:
        self._current_session.status = SessionStatus.ACTIVE
        self.session_manager.save(force=True)

    logger.info("Batch conversion resumed")
    return True

get_batch_status

get_batch_status()

Get current batch conversion status.

Returns:

Type	Description
BatchStatus	Current BatchStatus.

Source code in src/video_converter/core/orchestrator.py

def get_batch_status(self) -> BatchStatus:
    """Get current batch conversion status.

    Returns:
        Current BatchStatus.
    """
    return self._batch_status

is_paused

is_paused()

Check if batch conversion is paused.

Returns:

Type	Description
bool	True if paused, False otherwise.

Source code in src/video_converter/core/orchestrator.py

def is_paused(self) -> bool:
    """Check if batch conversion is paused.

    Returns:
        True if paused, False otherwise.
    """
    return self._batch_status == BatchStatus.PAUSED

get_pending_tasks

get_pending_tasks()

Get list of pending tasks.

Returns:

Type	Description
list[ConversionTask]	List of tasks not yet completed.

Source code in src/video_converter/core/orchestrator.py

def get_pending_tasks(self) -> list[ConversionTask]:
    """Get list of pending tasks.

    Returns:
        List of tasks not yet completed.
    """
    return [t for t in self._tasks if t.status == ConversionStatus.PENDING]

get_completed_tasks

get_completed_tasks()

Get list of completed tasks.

Returns:

Type	Description
list[ConversionTask]	List of successfully completed tasks.

Source code in src/video_converter/core/orchestrator.py

def get_completed_tasks(self) -> list[ConversionTask]:
    """Get list of completed tasks.

    Returns:
        List of successfully completed tasks.
    """
    return [t for t in self._tasks if t.status == ConversionStatus.COMPLETED]

get_failed_tasks

get_failed_tasks()

Get list of failed tasks.

Returns:

Type	Description
list[ConversionTask]	List of failed tasks.

Source code in src/video_converter/core/orchestrator.py

def get_failed_tasks(self) -> list[ConversionTask]:
    """Get list of failed tasks.

    Returns:
        List of failed tasks.
    """
    return [t for t in self._tasks if t.status == ConversionStatus.FAILED]

convert_single_sync

convert_single_sync(input_path, output_path=None, on_progress=None)

Synchronous wrapper for convert_single.

Parameters:

Name	Type	Description	Default
input_path	Path	Path to the input video.	required
output_path	Path | None	Path for the output video.	None
on_progress	ProgressCallback | None	Optional progress callback.	None

Returns:

Type	Description
ConversionResult	ConversionResult with success status and statistics.

Source code in src/video_converter/core/orchestrator.py

def convert_single_sync(
    self,
    input_path: Path,
    output_path: Path | None = None,
    on_progress: ProgressCallback | None = None,
) -> ConversionResult:
    """Synchronous wrapper for convert_single.

    Args:
        input_path: Path to the input video.
        output_path: Path for the output video.
        on_progress: Optional progress callback.

    Returns:
        ConversionResult with success status and statistics.
    """
    return asyncio.run(self.convert_single(input_path, output_path, on_progress))

run_sync

run_sync(input_paths, output_dir=None, on_progress=None, on_complete=None)

Synchronous wrapper for run.

Parameters:

Name	Type	Description	Default
input_paths	list[Path]	List of input video paths.	required
output_dir	Path | None	Directory for output files.	None
on_progress	ProgressCallback | None	Optional progress callback.	None
on_complete	CompleteCallback | None	Optional completion callback.	None

Returns:

Type	Description
ConversionReport	ConversionReport with batch statistics.

Source code in src/video_converter/core/orchestrator.py

def run_sync(
    self,
    input_paths: list[Path],
    output_dir: Path | None = None,
    on_progress: ProgressCallback | None = None,
    on_complete: CompleteCallback | None = None,
) -> ConversionReport:
    """Synchronous wrapper for run.

    Args:
        input_paths: List of input video paths.
        output_dir: Directory for output files.
        on_progress: Optional progress callback.
        on_complete: Optional completion callback.

    Returns:
        ConversionReport with batch statistics.
    """
    return asyncio.run(self.run(input_paths, output_dir, on_progress, on_complete))

check_disk_space

check_disk_space(path=None)

Check if there is sufficient disk space for conversion.

Parameters:

Name	Type	Description	Default
path	Path | None	Path to check. Uses output directory or home if None.	None

Returns:

Type	Description
tuple[bool, dict]	Tuple of (has_sufficient_space, disk_info_dict).

Source code in src/video_converter/core/orchestrator.py

def check_disk_space(self, path: Path | None = None) -> tuple[bool, dict]:
    """Check if there is sufficient disk space for conversion.

    Args:
        path: Path to check. Uses output directory or home if None.

    Returns:
        Tuple of (has_sufficient_space, disk_info_dict).
    """
    check_path = path
    if check_path is None:
        if self._current_session and self._current_session.output_dir:
            check_path = self._current_session.output_dir
        else:
            check_path = Path.home()

    sufficient, info = self.error_recovery_manager.has_sufficient_space(check_path)
    return sufficient, {
        "path": str(info.path),
        "total_bytes": info.total_bytes,
        "free_bytes": info.free_bytes,
        "used_bytes": info.used_bytes,
        "free_percent": info.free_percent,
        "sufficient": sufficient,
        "required_bytes": self.config.min_free_space,
    }

get_failure_records

get_failure_records()

Get list of all recorded failures.

Returns:

Type	Description
list[FailureRecord]	List of FailureRecord objects from this session.

Source code in src/video_converter/core/orchestrator.py

def get_failure_records(self) -> list[FailureRecord]:
    """Get list of all recorded failures.

    Returns:
        List of FailureRecord objects from this session.
    """
    return self.error_recovery_manager.failure_records

get_retryable_failures

get_retryable_failures()

Get list of failures that can be retried.

Returns:

Type	Description
list[FailureRecord]	List of retryable FailureRecord objects.

Source code in src/video_converter/core/orchestrator.py

def get_retryable_failures(self) -> list[FailureRecord]:
    """Get list of failures that can be retried.

    Returns:
        List of retryable FailureRecord objects.
    """
    return self.error_recovery_manager.get_retryable_failures()

retry_failed async

retry_failed(record, on_progress=None)

Retry a failed conversion.

This method allows manual retry of a specific failed conversion.

Parameters:

Name	Type	Description	Default
record	FailureRecord	The FailureRecord to retry.	required
on_progress	ProgressCallback | None	Optional progress callback.	None

Returns:

Type	Description
ConversionResult	ConversionResult with retry outcome.

Source code in src/video_converter/core/orchestrator.py

async def retry_failed(
    self,
    record: FailureRecord,
    on_progress: ProgressCallback | None = None,
) -> ConversionResult:
    """Retry a failed conversion.

    This method allows manual retry of a specific failed conversion.

    Args:
        record: The FailureRecord to retry.
        on_progress: Optional progress callback.

    Returns:
        ConversionResult with retry outcome.
    """
    # Prepare the file for retry
    input_path = self.error_recovery_manager.prepare_retry(record)
    if input_path is None:
        return ConversionResult(
            success=False,
            request=ConversionRequest(
                input_path=record.input_path,
                output_path=record.output_path,
            ),
            error_message=f"Cannot find file for retry: {record.input_path}",
        )

    # Check disk space before retry
    if self.config.check_disk_space:
        sufficient, _ = self.check_disk_space(record.output_path.parent)
        if not sufficient:
            return ConversionResult(
                success=False,
                request=ConversionRequest(
                    input_path=input_path,
                    output_path=record.output_path,
                ),
                error_message="Insufficient disk space for retry",
            )

    logger.info(f"Retrying failed conversion: {input_path.name} (attempt {record.retry_count})")

    # Perform the conversion
    result = await self.convert_single(
        input_path=input_path,
        output_path=record.output_path,
        on_progress=on_progress,
    )

    # Update failure record based on result
    if result.success:
        self.error_recovery_manager.mark_retry_success(record)

    return result

retry_all_failed async

retry_all_failed(on_progress=None)

Retry all failed conversions that are retryable.

Parameters:

Name	Type	Description	Default
on_progress	ProgressCallback | None	Optional progress callback.	None

Returns:

Type	Description
ConversionReport	ConversionReport with retry results.

Source code in src/video_converter/core/orchestrator.py

async def retry_all_failed(
    self,
    on_progress: ProgressCallback | None = None,
) -> ConversionReport:
    """Retry all failed conversions that are retryable.

    Args:
        on_progress: Optional progress callback.

    Returns:
        ConversionReport with retry results.
    """
    retryable = self.get_retryable_failures()

    if not retryable:
        logger.info("No retryable failures found")
        return ConversionReport(
            session_id=self._current_session_id or "retry",
            started_at=datetime.now(),
            completed_at=datetime.now(),
            total_files=0,
        )

    report = ConversionReport(
        session_id=f"{self._current_session_id or 'manual'}_retry",
        started_at=datetime.now(),
        total_files=len(retryable),
    )

    logger.info(f"Retrying {len(retryable)} failed conversion(s)")

    for i, record in enumerate(retryable):
        self._emit_progress(
            on_progress,
            ConversionStage.CONVERT,
            ConversionStatus.IN_PROGRESS,
            current_file=record.input_path.name,
            current_index=i,
            total_files=len(retryable),
            stage_progress=i / len(retryable),
            message=f"Retrying {record.input_path.name} ({i + 1}/{len(retryable)})...",
        )

        result = await self.retry_failed(record)
        report.add_result(result)

    report.completed_at = datetime.now()
    return report

retry_failed_sync

retry_failed_sync(record, on_progress=None)

Synchronous wrapper for retry_failed.

Parameters:

Name	Type	Description	Default
record	FailureRecord	The FailureRecord to retry.	required
on_progress	ProgressCallback | None	Optional progress callback.	None

Returns:

Type	Description
ConversionResult	ConversionResult with retry outcome.

Source code in src/video_converter/core/orchestrator.py

def retry_failed_sync(
    self,
    record: FailureRecord,
    on_progress: ProgressCallback | None = None,
) -> ConversionResult:
    """Synchronous wrapper for retry_failed.

    Args:
        record: The FailureRecord to retry.
        on_progress: Optional progress callback.

    Returns:
        ConversionResult with retry outcome.
    """
    return asyncio.run(self.retry_failed(record, on_progress))

retry_all_failed_sync

retry_all_failed_sync(on_progress=None)

Synchronous wrapper for retry_all_failed.

Parameters:

Name	Type	Description	Default
on_progress	ProgressCallback | None	Optional progress callback.	None

Returns:

Type	Description
ConversionReport	ConversionReport with retry results.

Source code in src/video_converter/core/orchestrator.py

def retry_all_failed_sync(
    self,
    on_progress: ProgressCallback | None = None,
) -> ConversionReport:
    """Synchronous wrapper for retry_all_failed.

    Args:
        on_progress: Optional progress callback.

    Returns:
        ConversionReport with retry results.
    """
    return asyncio.run(self.retry_all_failed(on_progress))

get_failure_summary

get_failure_summary()

Get summary of all recorded failures.

Returns:

Type	Description
dict	Dictionary with failure statistics.

Source code in src/video_converter/core/orchestrator.py

def get_failure_summary(self) -> dict:
    """Get summary of all recorded failures.

    Returns:
        Dictionary with failure statistics.
    """
    return self.error_recovery_manager.get_failure_summary()

clear_failure_records

clear_failure_records()

Clear all failure records.

Returns:

Type	Description
int	Number of records cleared.

Source code in src/video_converter/core/orchestrator.py

def clear_failure_records(self) -> int:
    """Clear all failure records.

    Returns:
        Number of records cleared.
    """
    return self.error_recovery_manager.clear_failures()

get_pause_reason

get_pause_reason()

Get the reason for pause if paused.

Returns:

Type	Description
str | None	Pause reason string, or None if not paused.

Source code in src/video_converter/core/orchestrator.py

def get_pause_reason(self) -> str | None:
    """Get the reason for pause if paused.

    Returns:
        Pause reason string, or None if not paused.
    """
    return self._paused_reason if self._paused else None

send_notification

send_notification(title, body, sound=None)

Send a custom notification.

Parameters:

Name	Type	Description	Default
title	str	The notification title.	required
body	str	The notification body text.	required
sound	bool | None	Whether to play sound. Uses config default if None.	None

Returns:

Type	Description
bool	True if notification was sent successfully.

Source code in src/video_converter/core/orchestrator.py

def send_notification(
    self,
    title: str,
    body: str,
    sound: bool | None = None,
) -> bool:
    """Send a custom notification.

    Args:
        title: The notification title.
        body: The notification body text.
        sound: Whether to play sound. Uses config default if None.

    Returns:
        True if notification was sent successfully.
    """
    if self._notification_manager is None:
        return False

    result = self._notification_manager.send_notification(
        title=title,
        body=body,
        sound=sound,
    )
    return result.success

Config

config

Configuration management system for video converter.

This module provides a centralized configuration system that loads, validates, and saves user preferences in JSON format. It supports both default and user-customized settings with environment variable overrides.

SDS Reference: SDS-C01-003 SRS Reference: SRS-102 (Configuration Management)

Example

from video_converter.core.config import Config config = Config.load() print(config.encoding.mode) # "hardware" config.encoding.quality = 50 config.save()

Environment variable override

VIDEO_CONVERTER_ENCODING__MODE=software

config = Config.load() print(config.encoding.mode) # "software"

EncodingConfig

Bases: BaseModel

Encoding settings for video conversion.

Attributes:

Name	Type	Description
mode	Literal['hardware', 'software']	Encoding mode - "hardware" (VideoToolbox) or "software" (libx265).
quality	int	Quality setting (1-100, higher is better).
crf	int	Constant Rate Factor for encoding (18-28, lower is better quality).
preset	Literal['fast', 'medium', 'slow']	Encoding preset affecting speed vs compression trade-off.
bit_depth	Literal[8, 10]	Output bit depth (8 or 10). 10-bit for HDR content.
hdr	bool	Enable HDR encoding parameters for 10-bit content.

PathsConfig

Bases: BaseModel

Path settings for file management.

Attributes:

Name	Type	Description
output	Path	Directory for converted output files.
processed	Path	Directory for successfully processed original files.
failed	Path	Directory for files that failed conversion.

expand_path classmethod

expand_path(v)

Expand user home directory in paths.

Parameters:

Name	Type	Description	Default
v	str | Path	Path value (string or Path).	required

Returns:

Type	Description
Path	Path with ~ expanded to user home directory.

Source code in src/video_converter/core/config.py

@field_validator("output", "processed", "failed", mode="before")
@classmethod
def expand_path(cls, v: str | Path) -> Path:
    """Expand user home directory in paths.

    Args:
        v: Path value (string or Path).

    Returns:
        Path with ~ expanded to user home directory.
    """
    if isinstance(v, str):
        v = Path(v)
    return v.expanduser()

AutomationConfig

Bases: BaseModel

Automation settings for scheduled conversions.

Attributes:

Name	Type	Description
enabled	bool	Whether automation is enabled.
schedule	Literal['hourly', 'daily', 'weekly']	Schedule type ("daily", "hourly", etc.).
time	str	Time of day to run (HH:MM format).

PhotosConfig

Bases: BaseModel

Photos library settings.

Attributes:

Name	Type	Description
include_albums	list[str]	List of album names to include (empty = all).
exclude_albums	list[str]	List of album names to exclude.
download_from_icloud	bool	Whether to download videos from iCloud.
icloud_timeout	int	Maximum time to wait for iCloud downloads in seconds.
skip_cloud_only	bool	Skip videos that are only in iCloud (don't download).

FolderConfig

Bases: BaseModel

Folder-based video scanning settings.

Settings for scanning videos from filesystem directories, including iCloud Drive folders with stub files.

SRS Reference: SRS-304 (Folder-based Video Extraction)

Attributes:

Name	Type	Description
recursive	bool	Whether to scan subdirectories.
auto_download_icloud	bool	Whether to auto-download iCloud stub files.
icloud_timeout	int	Maximum time to wait for iCloud downloads in seconds.
skip_icloud_on_timeout	bool	Skip file if download times out.
include_patterns	list[str]	Glob patterns to include (e.g., ["vacation*"]).
exclude_patterns	list[str]	Glob patterns to exclude (e.g., [".tmp", "._"]).

VmafConfig

Bases: BaseModel

VMAF quality analysis settings.

VMAF (Video Multimethod Assessment Fusion) is a perceptual video quality assessment algorithm that predicts subjective quality scores.

Attributes:

Name	Type	Description
enabled	bool	Whether to measure VMAF quality score after conversion.
threshold	float	Minimum acceptable VMAF score (0-100). 93.0 is visually lossless.
sample_interval	int	Frame sampling interval for VMAF analysis (1=all, 30=faster).
fail_action	Literal['warn', 'retry', 'fail']	Action when VMAF is below threshold. - "warn": Log warning but keep file. - "retry": Retry with adjusted settings. - "fail": Delete file and mark as failed.

ProcessingConfig

Bases: BaseModel

Processing settings for conversion workflow.

Attributes:

Name	Type	Description
max_concurrent	int	Maximum concurrent conversions.
validate_quality	bool	Whether to validate output quality.
preserve_original	bool	Whether to preserve original files.
move_processed	bool	Whether to move processed originals to paths.processed.
move_failed	bool	Whether to move failed files to paths.failed.
check_disk_space	bool	Whether to check disk space before processing.
min_free_space_gb	float	Minimum free disk space in gigabytes.

NotificationConfig

Bases: BaseModel

Notification settings.

Attributes:

Name	Type	Description
on_complete	bool	Send notification when conversion completes.
on_error	bool	Send notification on errors.
daily_summary	bool	Send daily summary notification.

Config

Bases: BaseSettings

Main configuration class for video converter.

This class manages all configuration settings for the video converter application. It supports loading from JSON files, environment variable overrides, and runtime modifications.

Attributes:

Name	Type	Description
version	str	Configuration schema version.
encoding	EncodingConfig	Encoding settings.
paths	PathsConfig	Path settings.
automation	AutomationConfig	Automation settings.
photos	PhotosConfig	Photos library settings.
processing	ProcessingConfig	Processing settings.
notification	NotificationConfig	Notification settings.

Example

config = Config.load() print(config.encoding.mode) 'hardware'

Modify and save

config.encoding.quality = 60 config.save()

Environment override (VIDEO_CONVERTER_ENCODING__MODE=software)

config = Config.load() print(config.encoding.mode) 'software'

settings_customise_sources classmethod

settings_customise_sources(settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings)

Customize settings sources priority.

Priority (highest to lowest): 1. init_settings (direct arguments) 2. env_settings (environment variables) 3. JSON file settings 4. Default values

This ensures environment variables override JSON file settings.

Source code in src/video_converter/core/config.py

@classmethod
def settings_customise_sources(
    cls,
    settings_cls: type[BaseSettings],
    init_settings: PydanticBaseSettingsSource,
    env_settings: PydanticBaseSettingsSource,
    dotenv_settings: PydanticBaseSettingsSource,
    file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]:
    """Customize settings sources priority.

    Priority (highest to lowest):
    1. init_settings (direct arguments)
    2. env_settings (environment variables)
    3. JSON file settings
    4. Default values

    This ensures environment variables override JSON file settings.
    """
    # Explicitly mark unused parameters (required by pydantic-settings interface)
    _ = dotenv_settings
    _ = file_secret_settings

    # Find config file
    json_file = cls._find_config_file()

    # Create JSON source if file exists
    json_source = _JsonFileSettingsSource(settings_cls, json_file=json_file)

    return (
        init_settings,
        env_settings,
        json_source,
    )

load classmethod

load(config_path=None, *, force_reload=False)

Load configuration from file.

Loads configuration from the specified path, or falls back to default locations. Environment variables can override loaded values.

Parameters:

Name	Type	Description	Default
config_path	Path | None	Deprecated. Custom paths are not currently supported. Use environment variables or modify the default files.	None
force_reload	bool	Force reload even if already loaded.	False

Returns:

Name	Type	Description
Config	Config	Loaded configuration instance.

Raises:

Type	Description
ValueError	If configuration file contains invalid values.

Example

config = Config.load() # Uses default paths

Environment override: VIDEO_CONVERTER_ENCODING__MODE=software

Source code in src/video_converter/core/config.py

@classmethod
def load(
    cls,
    config_path: Path | None = None,
    *,
    force_reload: bool = False,
) -> Config:
    """Load configuration from file.

    Loads configuration from the specified path, or falls back to default
    locations. Environment variables can override loaded values.

    Args:
        config_path: Deprecated. Custom paths are not currently supported.
                    Use environment variables or modify the default files.
        force_reload: Force reload even if already loaded.

    Returns:
        Config: Loaded configuration instance.

    Raises:
        ValueError: If configuration file contains invalid values.

    Example:
        >>> config = Config.load()  # Uses default paths
        >>> # Environment override: VIDEO_CONVERTER_ENCODING__MODE=software
    """
    global _config_instance, _config_path_cache

    # Mark deprecated parameter as unused
    _ = config_path

    with _config_lock:
        if _config_instance is not None and not force_reload:
            return _config_instance

        # Create instance - settings_customise_sources handles JSON loading
        instance = cls()
        _config_path_cache = cls._find_config_file() or DEFAULT_CONFIG_FILE

        _config_instance = instance
        return instance

save

save(config_path=None)

Save configuration to file.

Parameters:

Name	Type	Description	Default
config_path	Path | None	Optional path to save to. Defaults to loaded path.	None

Raises:

Type	Description
IOError	If file cannot be written.

Example

config = Config.load() config.encoding.quality = 60 config.save()

Source code in src/video_converter/core/config.py

def save(self, config_path: Path | None = None) -> None:
    """Save configuration to file.

    Args:
        config_path: Optional path to save to. Defaults to loaded path.

    Raises:
        IOError: If file cannot be written.

    Example:
        >>> config = Config.load()
        >>> config.encoding.quality = 60
        >>> config.save()
    """
    save_path = config_path or _config_path_cache or DEFAULT_CONFIG_FILE

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

    # Convert to dict and save
    config_dict = self.model_dump(mode="json")

    # Convert Path objects to strings for JSON serialization
    if "paths" in config_dict:
        for key, value in config_dict["paths"].items():
            if isinstance(value, Path):
                config_dict["paths"][key] = str(value)

    with save_path.open("w", encoding="utf-8") as f:
        json.dump(config_dict, f, indent=2, ensure_ascii=False)
        f.write("\n")

reset classmethod

reset()

Reset the singleton instance.

This is primarily useful for testing to ensure a clean state.

Source code in src/video_converter/core/config.py

@classmethod
def reset(cls) -> None:
    """Reset the singleton instance.

    This is primarily useful for testing to ensure a clean state.
    """
    global _config_instance, _config_path_cache

    with _config_lock:
        _config_instance = None
        _config_path_cache = None

get_default_config_path classmethod

get_default_config_path()

Get the default user configuration file path.

Returns:

Type	Description
Path	Path to user configuration file.

Source code in src/video_converter/core/config.py

@classmethod
def get_default_config_path(cls) -> Path:
    """Get the default user configuration file path.

    Returns:
        Path to user configuration file.
    """
    return DEFAULT_CONFIG_FILE

get_config_dir classmethod

get_config_dir()

Get the configuration directory path.

Returns:

Type	Description
Path	Path to configuration directory.

Source code in src/video_converter/core/config.py

@classmethod
def get_config_dir(cls) -> Path:
    """Get the configuration directory path.

    Returns:
        Path to configuration directory.
    """
    return DEFAULT_CONFIG_DIR

to_dict

to_dict()

Convert configuration to dictionary.

Returns:

Type	Description
dict[str, Any]	Dictionary representation of configuration.

Source code in src/video_converter/core/config.py

def to_dict(self) -> dict[str, Any]:
    """Convert configuration to dictionary.

    Returns:
        Dictionary representation of configuration.
    """
    result: dict[str, Any] = self.model_dump()
    return result

Types

types

Core type definitions for video conversion workflow.

This module defines the essential data classes used throughout the video conversion pipeline, including conversion requests, results, progress tracking, and reporting structures.

SDS Reference: SDS-C01-001 SRS Reference: SRS-601 (Orchestrator Workflow)

Example

from video_converter.core.types import ConversionRequest, ConversionResult request = ConversionRequest( ... input_path=Path("input.mov"), ... output_path=Path("output.mp4"), ... mode="hardware", ... )

After conversion

result = ConversionResult( ... success=True, ... request=request, ... original_size=100_000_000, ... converted_size=40_000_000, ... duration_seconds=120.5, ... ) print(f"Compression ratio: {result.compression_ratio:.1%}") Compression ratio: 60.0%

ConversionMode

Bases: Enum

Video conversion mode.

Attributes:

Name	Type	Description
HARDWARE		Use hardware encoder (VideoToolbox on macOS).
SOFTWARE		Use software encoder (libx265).

ErrorCategory

Bases: Enum

Category of conversion error for recovery action determination.

SDS Reference: SDS-C01-004 SRS Reference: SRS-605 (Failure Isolation)

Attributes:

Name	Type	Description
INPUT_ERROR		Input file issue (missing, corrupt, unsupported format).
ENCODING_ERROR		Encoder failure during conversion.
VALIDATION_ERROR		Output file failed quality validation.
METADATA_ERROR		Metadata copy/sync failure (non-fatal).
DISK_SPACE_ERROR		Insufficient disk space.
PERMISSION_ERROR		File permission or access denied.
UNKNOWN_ERROR		Unclassified error.

RecoveryAction

Bases: Enum

Recommended recovery action for an error category.

Attributes:

Name	Type	Description
SKIP		Skip this file and continue with next.
RETRY_WITH_FALLBACK		Retry with alternative encoder/settings.
RETRY_ONCE		Retry one more time with same settings.
KEEP_VIDEO_LOG_WARNING		Keep converted video, log the warning.
PAUSE_NOTIFY_USER		Pause processing and notify user.
ABORT		Stop all processing.

QueuePriority

Bases: Enum

Queue priority ordering for batch processing.

Attributes:

Name	Type	Description
FIFO		First-in, first-out (default).
DATE_OLDEST		Oldest files first (by modification time).
DATE_NEWEST		Newest files first (by modification time).
SIZE_SMALLEST		Smallest files first.
SIZE_LARGEST		Largest files first.

ConversionStage

Bases: Enum

Stages in the conversion pipeline.

Attributes:

Name	Type	Description
DISCOVERY		Finding videos to convert.
EXPORT		Extracting from Photos library.
CONVERT		Running the encoder.
VALIDATE		Verifying output integrity.
METADATA		Copying metadata.
CLEANUP		Moving/deleting files.
COMPLETE		All stages finished.

ConversionStatus

Bases: Enum

Status of a conversion operation.

Attributes:

Name	Type	Description
PENDING		Waiting to be processed.
IN_PROGRESS		Currently being processed.
COMPLETED		Successfully completed.
FAILED		Failed during processing.
SKIPPED		Skipped (already converted or filtered).
CANCELLED		Cancelled by user.

BatchStatus

Bases: Enum

Status of a batch conversion operation.

Attributes:

Name	Type	Description
IDLE		No batch is running.
RUNNING		Batch is actively processing.
PAUSED		Batch is paused by user.
COMPLETED		Batch finished (success or with failures).
CANCELLED		Batch was cancelled.

ConversionRequest dataclass

ConversionRequest(input_path, output_path, mode=HARDWARE, quality=45, crf=22, preset='medium', audio_mode='copy', preserve_metadata=True, bit_depth=8, hdr=False)

Request for video conversion.

Attributes:

Name	Type	Description
input_path	Path	Path to the input video file.
output_path	Path	Path for the converted output file.
mode	ConversionMode	Conversion mode (hardware or software).
quality	int	Quality setting (1-100, higher is better).
crf	int	Constant Rate Factor for encoding.
preset	str	Encoding preset (speed vs compression trade-off).
audio_mode	str	How to handle audio ("copy", "aac", etc.).
preserve_metadata	bool	Whether to copy metadata from original.
bit_depth	int	Output bit depth (8 or 10). 10-bit for HDR content.
hdr	bool	Enable HDR encoding parameters for 10-bit content.

__post_init__

__post_init__()

Validate and normalize fields.

Source code in src/video_converter/core/types.py

def __post_init__(self) -> None:
    """Validate and normalize fields."""
    if isinstance(self.input_path, str):
        self.input_path = Path(self.input_path)
    if isinstance(self.output_path, str):
        self.output_path = Path(self.output_path)
    if isinstance(self.mode, str):
        self.mode = ConversionMode(self.mode)

ConversionResult dataclass

ConversionResult(success, request, original_size=0, converted_size=0, duration_seconds=0.0, speed_ratio=0.0, error_message=None, warnings=list(), started_at=None, completed_at=None, retry_count=0, retry_strategy_used=None, retry_history=list(), vmaf_score=None, vmaf_quality_level=None)

Result of a single video conversion.

Attributes:

Name	Type	Description
success	bool	Whether conversion completed successfully.
request	ConversionRequest	The original conversion request.
original_size	int	Size of input file in bytes.
converted_size	int	Size of output file in bytes.
duration_seconds	float	Time taken to convert in seconds.
speed_ratio	float	Conversion speed relative to realtime.
error_message	str | None	Error message if conversion failed.
warnings	list[str]	Non-critical issues encountered.
started_at	datetime | None	When conversion started.
completed_at	datetime | None	When conversion finished.
retry_count	int	Number of retry attempts made (0 if no retries).
retry_strategy_used	str | None	Final retry strategy that succeeded (or None).
retry_history	list[dict]	Detailed history of all retry attempts.
vmaf_score	float | None	VMAF quality score (0-100) if measured.
vmaf_quality_level	str | None	Quality classification based on VMAF score.

compression_ratio property

compression_ratio

Calculate compression ratio (0.0-1.0).

Returns:

Type	Description
float	Compression ratio where 0.6 means 60% size reduction.

size_saved property

size_saved

Calculate bytes saved by compression.

ConversionProgress dataclass

ConversionProgress(stage, status, current_file='', current_index=0, total_files=0, stage_progress=0.0, overall_progress=0.0, bytes_processed=0, bytes_total=0, estimated_time_remaining=None, message='')

Progress information for conversion tracking.

Attributes:

Name	Type	Description
stage	ConversionStage	Current stage in the pipeline.
status	ConversionStatus	Current status of the operation.
current_file	str	Name of the file being processed.
current_index	int	Index of current file (0-based).
total_files	int	Total number of files to process.
stage_progress	float	Progress within current stage (0.0-1.0).
overall_progress	float	Overall progress (0.0-1.0).
bytes_processed	int	Bytes processed so far.
bytes_total	int	Total bytes to process.
estimated_time_remaining	float | None	Estimated seconds remaining.
message	str	Human-readable status message.

__post_init__

__post_init__()

Validate progress values.

Source code in src/video_converter/core/types.py

def __post_init__(self) -> None:
    """Validate progress values."""
    self.stage_progress = max(0.0, min(1.0, self.stage_progress))
    self.overall_progress = max(0.0, min(1.0, self.overall_progress))

ConversionReport dataclass

ConversionReport(session_id, started_at, completed_at=None, total_files=0, successful=0, failed=0, skipped=0, cancelled=False, total_original_size=0, total_converted_size=0, total_duration_seconds=0.0, results=list(), errors=list(), warnings=list())

Summary report for a conversion batch.

Attributes:

Name	Type	Description
session_id	str	Unique identifier for this conversion session.
started_at	datetime	When the batch started.
completed_at	datetime | None	When the batch finished.
total_files	int	Total files in the batch.
successful	int	Number of successfully converted files.
failed	int	Number of failed conversions.
skipped	int	Number of skipped files.
cancelled	bool	Whether the batch was cancelled.
total_original_size	int	Sum of all original file sizes.
total_converted_size	int	Sum of all converted file sizes.
total_duration_seconds	float	Total conversion time.
results	list[ConversionResult]	Individual conversion results.
errors	list[str]	Aggregated error messages.
warnings	list[str]	Aggregated warning messages.

total_size_saved property

total_size_saved

Calculate total bytes saved.

average_compression_ratio property

average_compression_ratio

Calculate average compression ratio.

success_rate property

success_rate

Calculate success rate (0.0-1.0).

add_result

add_result(result)

Add a conversion result to the report.

Parameters:

Name	Type	Description	Default
result	ConversionResult	The conversion result to add.	required

Source code in src/video_converter/core/types.py

def add_result(self, result: ConversionResult) -> None:
    """Add a conversion result to the report.

    Args:
        result: The conversion result to add.
    """
    self.results.append(result)
    self.total_original_size += result.original_size
    self.total_converted_size += result.converted_size
    self.total_duration_seconds += result.duration_seconds

    if result.success:
        self.successful += 1
    else:
        self.failed += 1
        if result.error_message:
            self.errors.append(f"{result.request.input_path.name}: {result.error_message}")

    self.warnings.extend(result.warnings)

SessionStatus

Bases: Enum

Status of a conversion session for persistence.

Attributes:

Name	Type	Description
ACTIVE		Session is currently running.
PAUSED		Session was paused and can be resumed.
COMPLETED		Session finished successfully.
INTERRUPTED		Session was interrupted (crash, shutdown).
CANCELLED		Session was explicitly cancelled.

VideoEntry dataclass

VideoEntry(path, output_path, status=PENDING, error_message=None, original_size=0, converted_size=0)

Entry for a video file in session state.

Attributes:

Name	Type	Description
path	Path	Absolute path to the video file.
output_path	Path	Absolute path to the output file.
status	ConversionStatus	Current conversion status.
error_message	str | None	Error message if failed.
original_size	int	Size of original file in bytes.
converted_size	int	Size of converted file in bytes.

__post_init__

__post_init__()

Validate and normalize fields.

Source code in src/video_converter/core/types.py

def __post_init__(self) -> None:
    """Validate and normalize fields."""
    if isinstance(self.path, str):
        self.path = Path(self.path)
    if isinstance(self.output_path, str):
        self.output_path = Path(self.output_path)
    if isinstance(self.status, str):
        self.status = ConversionStatus(self.status)

to_dict

to_dict()

Convert to JSON-serializable dictionary.

Source code in src/video_converter/core/types.py

def to_dict(self) -> dict:
    """Convert to JSON-serializable dictionary."""
    return {
        "path": str(self.path),
        "output_path": str(self.output_path),
        "status": self.status.value,
        "error_message": self.error_message,
        "original_size": self.original_size,
        "converted_size": self.converted_size,
    }

from_dict classmethod

from_dict(data)

Create from dictionary.

Source code in src/video_converter/core/types.py

@classmethod
def from_dict(cls, data: dict) -> VideoEntry:
    """Create from dictionary."""
    return cls(
        path=Path(data["path"]),
        output_path=Path(data["output_path"]),
        status=ConversionStatus(data["status"]),
        error_message=data.get("error_message"),
        original_size=data.get("original_size", 0),
        converted_size=data.get("converted_size", 0),
    )

SessionState dataclass

SessionState(session_id, status=ACTIVE, started_at=now(), updated_at=now(), current_index=0, pending_videos=list(), completed_videos=list(), failed_videos=list(), temporary_files=list(), output_dir=None, config_snapshot=dict())

Persistent session state for resumable conversions.

Tracks all information needed to resume an interrupted conversion session, including which files have been processed and any temporary files created.

Attributes:

Name	Type	Description
session_id	str	Unique identifier for this session.
status	SessionStatus	Current session status.
started_at	datetime	When the session started.
updated_at	datetime	When the session was last updated.
current_index	int	Index of the file currently being processed.
pending_videos	list[VideoEntry]	List of videos waiting to be processed.
completed_videos	list[VideoEntry]	List of successfully completed videos.
failed_videos	list[VideoEntry]	List of videos that failed conversion.
temporary_files	list[Path]	List of temporary files created during conversion.
output_dir	Path | None	Output directory for converted files.
config_snapshot	dict	Snapshot of configuration at session start.

total_videos property

total_videos

Get total number of videos in the session.

progress property

progress

Get session progress as a fraction (0.0-1.0).

is_resumable property

is_resumable

Check if this session can be resumed.

__post_init__

__post_init__()

Validate and normalize fields.

Source code in src/video_converter/core/types.py

def __post_init__(self) -> None:
    """Validate and normalize fields."""
    if isinstance(self.status, str):
        self.status = SessionStatus(self.status)
    if isinstance(self.started_at, str):
        self.started_at = datetime.fromisoformat(self.started_at)
    if isinstance(self.updated_at, str):
        self.updated_at = datetime.fromisoformat(self.updated_at)
    if self.output_dir is not None and isinstance(self.output_dir, str):
        self.output_dir = Path(self.output_dir)
    # Normalize temporary files
    self.temporary_files = [Path(f) if isinstance(f, str) else f for f in self.temporary_files]

mark_video_completed

mark_video_completed(video, original_size=0, converted_size=0)

Move a video from pending to completed.

Parameters:

Name	Type	Description	Default
video	VideoEntry	The video entry to mark complete.	required
original_size	int	Size of original file in bytes.	0
converted_size	int	Size of converted file in bytes.	0

Source code in src/video_converter/core/types.py

def mark_video_completed(
    self,
    video: VideoEntry,
    original_size: int = 0,
    converted_size: int = 0,
) -> None:
    """Move a video from pending to completed.

    Args:
        video: The video entry to mark complete.
        original_size: Size of original file in bytes.
        converted_size: Size of converted file in bytes.
    """
    video.status = ConversionStatus.COMPLETED
    video.original_size = original_size
    video.converted_size = converted_size
    if video in self.pending_videos:
        self.pending_videos.remove(video)
    if video not in self.completed_videos:
        self.completed_videos.append(video)
    self.updated_at = datetime.now()

mark_video_failed

mark_video_failed(video, error)

Move a video from pending to failed.

Parameters:

Name	Type	Description	Default
video	VideoEntry	The video entry to mark failed.	required
error	str	Error message describing the failure.	required

Source code in src/video_converter/core/types.py

def mark_video_failed(self, video: VideoEntry, error: str) -> None:
    """Move a video from pending to failed.

    Args:
        video: The video entry to mark failed.
        error: Error message describing the failure.
    """
    video.status = ConversionStatus.FAILED
    video.error_message = error
    if video in self.pending_videos:
        self.pending_videos.remove(video)
    if video not in self.failed_videos:
        self.failed_videos.append(video)
    self.updated_at = datetime.now()

add_temporary_file

add_temporary_file(path)

Track a temporary file created during conversion.

Parameters:

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

Source code in src/video_converter/core/types.py

def add_temporary_file(self, path: Path) -> None:
    """Track a temporary file created during conversion.

    Args:
        path: Path to the temporary file.
    """
    if path not in self.temporary_files:
        self.temporary_files.append(path)
        self.updated_at = datetime.now()

remove_temporary_file

remove_temporary_file(path)

Remove a temporary file from tracking.

Parameters:

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

Source code in src/video_converter/core/types.py

def remove_temporary_file(self, path: Path) -> None:
    """Remove a temporary file from tracking.

    Args:
        path: Path to the temporary file.
    """
    if path in self.temporary_files:
        self.temporary_files.remove(path)
        self.updated_at = datetime.now()

to_dict

to_dict()

Convert to JSON-serializable dictionary.

Source code in src/video_converter/core/types.py

def to_dict(self) -> dict:
    """Convert to JSON-serializable dictionary."""
    return {
        "session_id": self.session_id,
        "status": self.status.value,
        "started_at": self.started_at.isoformat(),
        "updated_at": self.updated_at.isoformat(),
        "current_index": self.current_index,
        "pending_videos": [v.to_dict() for v in self.pending_videos],
        "completed_videos": [v.to_dict() for v in self.completed_videos],
        "failed_videos": [v.to_dict() for v in self.failed_videos],
        "temporary_files": [str(f) for f in self.temporary_files],
        "output_dir": str(self.output_dir) if self.output_dir else None,
        "config_snapshot": self.config_snapshot,
    }

from_dict classmethod

from_dict(data)

Create from dictionary.

Parameters:

Name	Type	Description	Default
data	dict	Dictionary containing session state data.	required

Returns:

Type	Description
SessionState	A new SessionState instance.

Source code in src/video_converter/core/types.py

@classmethod
def from_dict(cls, data: dict) -> SessionState:
    """Create from dictionary.

    Args:
        data: Dictionary containing session state data.

    Returns:
        A new SessionState instance.
    """
    return cls(
        session_id=data["session_id"],
        status=SessionStatus(data["status"]),
        started_at=datetime.fromisoformat(data["started_at"]),
        updated_at=datetime.fromisoformat(data["updated_at"]),
        current_index=data.get("current_index", 0),
        pending_videos=[VideoEntry.from_dict(v) for v in data.get("pending_videos", [])],
        completed_videos=[VideoEntry.from_dict(v) for v in data.get("completed_videos", [])],
        failed_videos=[VideoEntry.from_dict(v) for v in data.get("failed_videos", [])],
        temporary_files=[Path(f) for f in data.get("temporary_files", [])],
        output_dir=Path(data["output_dir"]) if data.get("output_dir") else None,
        config_snapshot=data.get("config_snapshot", {}),
    )

Logger

logger

Logging system with file and console output.

This module provides a comprehensive logging system that outputs to both file and console with configurable log levels, log rotation, and colored console output using the Rich library.

SDS Reference: SDS-C01-002 SRS Reference: SRS-101 (Logging System)

Example

from video_converter.core.logger import get_logger logger = get_logger(name) logger.info("Starting conversion") logger.debug("Processing file: %s", filename) logger.error("Conversion failed", exc_info=True)

Configure log level globally

from video_converter.core.logger import configure_logging configure_logging(level="DEBUG", log_dir=Path("/custom/path"))

LogLevel

Log level constants for convenient access.

Attributes:

Name	Type	Description
DEBUG		Detailed information for debugging.
INFO		General operational information.
WARNING		Indication of potential issues.
ERROR		Error that prevented an operation.
CRITICAL		Critical error, application may not continue.

configure_logging

configure_logging(level=DEFAULT_LOG_LEVEL, log_dir=None, console_output=True, file_output=True)

Configure the global logging settings.

This function should be called once at application startup to configure the logging system. Subsequent calls will update the configuration.

Parameters:

Name	Type	Description	Default
level	int | str	Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL). Can be either an integer or string.	DEFAULT_LOG_LEVEL
log_dir	Path | None	Directory for log files. Default: ~/.local/share/video_converter/logs	None
console_output	bool	Whether to output logs to console. Default: True.	True
file_output	bool	Whether to output logs to file. Default: True.	True

Example

configure_logging(level="DEBUG") configure_logging(level=logging.WARNING, log_dir=Path("/var/log/myapp"))

Source code in src/video_converter/core/logger.py

def configure_logging(
    level: int | str = DEFAULT_LOG_LEVEL,
    log_dir: Path | None = None,
    console_output: bool = True,
    file_output: bool = True,
) -> None:
    """Configure the global logging settings.

    This function should be called once at application startup to configure
    the logging system. Subsequent calls will update the configuration.

    Args:
        level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL).
               Can be either an integer or string.
        log_dir: Directory for log files. Default: ~/.local/share/video_converter/logs
        console_output: Whether to output logs to console. Default: True.
        file_output: Whether to output logs to file. Default: True.

    Example:
        >>> configure_logging(level="DEBUG")
        >>> configure_logging(level=logging.WARNING, log_dir=Path("/var/log/myapp"))
    """
    global _log_dir, _log_level, _initialized

    # Convert string level to int
    if isinstance(level, str):
        _log_level = getattr(logging, level.upper(), DEFAULT_LOG_LEVEL)
    else:
        _log_level = level

    if log_dir is not None:
        _log_dir = log_dir

    # Reset root logger configuration
    root_logger = logging.getLogger("video_converter")
    root_logger.setLevel(_log_level)

    # Remove existing handlers
    for handler in root_logger.handlers[:]:
        root_logger.removeHandler(handler)
        handler.close()

    # Add handlers based on configuration
    if console_output:
        root_logger.addHandler(_create_console_handler())

    if file_output:
        root_logger.addHandler(_create_file_handler())

    _initialized = True

get_logger

get_logger(name)

Get a logger instance with the specified name.

This is the primary function for obtaining a logger. It returns a logger that is a child of the "video_converter" root logger, ensuring consistent configuration across the application.

Parameters:

Name	Type	Description	Default
name	str	The name of the logger, typically name.	required

Returns:

Name	Type	Description
Logger	Logger	Configured logger instance.

Example

logger = get_logger(name) logger.info("Starting conversion") logger.debug("Processing file: %s", filename) logger.error("Conversion failed", exc_info=True)

Source code in src/video_converter/core/logger.py

def get_logger(name: str) -> logging.Logger:
    """Get a logger instance with the specified name.

    This is the primary function for obtaining a logger. It returns a logger
    that is a child of the "video_converter" root logger, ensuring consistent
    configuration across the application.

    Args:
        name: The name of the logger, typically __name__.

    Returns:
        Logger: Configured logger instance.

    Example:
        >>> logger = get_logger(__name__)
        >>> logger.info("Starting conversion")
        >>> logger.debug("Processing file: %s", filename)
        >>> logger.error("Conversion failed", exc_info=True)
    """
    global _initialized

    # Ensure logging is configured on first use
    if not _initialized:
        configure_logging()

    # Create child logger under video_converter namespace
    logger_name = name if name.startswith("video_converter") else f"video_converter.{name}"
    logger = logging.getLogger(logger_name)

    return logger

set_log_level

set_log_level(level)

Set the log level for all video_converter loggers.

Parameters:

Name	Type	Description	Default
level	int | str	Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL). Can be either an integer or string.	required

Example

set_log_level("DEBUG") set_log_level(logging.WARNING)

Source code in src/video_converter/core/logger.py

def set_log_level(level: int | str) -> None:
    """Set the log level for all video_converter loggers.

    Args:
        level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL).
               Can be either an integer or string.

    Example:
        >>> set_log_level("DEBUG")
        >>> set_log_level(logging.WARNING)
    """
    global _log_level

    if isinstance(level, str):
        _log_level = getattr(logging, level.upper(), DEFAULT_LOG_LEVEL)
    else:
        _log_level = level

    # Update all handlers
    root_logger = logging.getLogger("video_converter")
    root_logger.setLevel(_log_level)

    for handler in root_logger.handlers:
        handler.setLevel(_log_level)

get_log_file_path

get_log_file_path()

Get the path to the current log file.

Returns:

Name	Type	Description
Path	Path	Path to the log file.

Source code in src/video_converter/core/logger.py

def get_log_file_path() -> Path:
    """Get the path to the current log file.

    Returns:
        Path: Path to the log file.
    """
    return _log_dir / "video_converter.log"

get_log_dir

get_log_dir()

Get the log directory path.

Returns:

Name	Type	Description
Path	Path	Path to the log directory.

Source code in src/video_converter/core/logger.py

def get_log_dir() -> Path:
    """Get the log directory path.

    Returns:
        Path: Path to the log directory.
    """
    return _log_dir

Session

session

Session state management for resumable conversions.

This module implements persistent session state tracking to enable resuming interrupted video conversions.

SDS Reference: SDS-C01-003 SRS Reference: SRS-603 (Session State Management)

Example

from video_converter.core.session import SessionStateManager from pathlib import Path

manager = SessionStateManager()

Create a new session

session = manager.create_session( ... video_paths=[Path("video1.mov"), Path("video2.mov")], ... output_dir=Path("converted"), ... )

Update session progress

manager.mark_completed(session.pending_videos[0]) manager.save()

Resume an interrupted session

if manager.has_resumable_session(): ... session = manager.load_session() ... # Continue processing pending_videos

SessionStateError

Bases: Exception

Base exception for session state errors.

SessionNotFoundError

Bases: SessionStateError

Raised when a session cannot be found.

SessionCorruptedError

Bases: SessionStateError

Raised when session data is corrupted.

SessionStateManager

SessionStateManager(state_dir=None, auto_save_interval=30)

Manages persistent session state for video conversion.

Provides functionality to: - Create and track conversion sessions - Save session state to disk periodically - Resume from saved state after restart - Handle orphaned temporary files - Query session status

Attributes:

Name	Type	Description
state_dir		Directory for session state files.
current_session	SessionState | None	The currently active session, if any.

Initialize the session state manager.

Parameters:

Name	Type	Description	Default
state_dir	Path | None	Directory for session state files. Defaults to ~/.local/share/video_converter/sessions	None
auto_save_interval	int	Seconds between auto-saves. 0 disables auto-save.	30

Source code in src/video_converter/core/session.py

def __init__(
    self,
    state_dir: Path | None = None,
    auto_save_interval: int = 30,
) -> None:
    """Initialize the session state manager.

    Args:
        state_dir: Directory for session state files.
                  Defaults to ~/.local/share/video_converter/sessions
        auto_save_interval: Seconds between auto-saves. 0 disables auto-save.
    """
    self.state_dir = state_dir or DEFAULT_STATE_DIR
    self.state_dir.mkdir(parents=True, exist_ok=True)

    self._current_session: SessionState | None = None
    self._lock = threading.RLock()
    self._auto_save_interval = auto_save_interval
    self._last_save_time: datetime | None = None
    self._dirty = False

    # Check for interrupted sessions on startup
    self._check_interrupted_sessions()

current_session property

current_session

Get the current session.

state_file_path property

state_file_path

Get the path to the current session state file.

history_dir property

history_dir

Get the path to the session history directory.

create_session

create_session(video_paths, output_dir=None, config=None)

Create a new conversion session.

Parameters:

Name	Type	Description	Default
video_paths	list[Path]	List of video files to convert.	required
output_dir	Path | None	Directory for output files.	None
config	OrchestratorConfig | None	Orchestrator configuration to snapshot.	None

Returns:

Type	Description
SessionState	A new SessionState instance.

Raises:

Type	Description
SessionStateError	If a session is already active.

Source code in src/video_converter/core/session.py

def create_session(
    self,
    video_paths: list[Path],
    output_dir: Path | None = None,
    config: OrchestratorConfig | None = None,
) -> SessionState:
    """Create a new conversion session.

    Args:
        video_paths: List of video files to convert.
        output_dir: Directory for output files.
        config: Orchestrator configuration to snapshot.

    Returns:
        A new SessionState instance.

    Raises:
        SessionStateError: If a session is already active.
    """
    with self._lock:
        if self._current_session and self._current_session.status == SessionStatus.ACTIVE:
            raise SessionStateError(
                f"Session {self._current_session.session_id} is already active. "
                "Complete or cancel it before starting a new session."
            )

        session_id = self._generate_session_id()

        # Create video entries
        pending_videos = []
        for path in video_paths:
            output_path = self._create_output_path(path, output_dir, config)
            pending_videos.append(
                VideoEntry(
                    path=path,
                    output_path=output_path,
                    status=ConversionStatus.PENDING,
                )
            )

        # Snapshot config
        config_snapshot = {}
        if config:
            config_snapshot = {
                "mode": config.mode.value,
                "quality": config.quality,
                "crf": config.crf,
                "preset": config.preset,
                "output_suffix": config.output_suffix,
                "preserve_metadata": config.preserve_metadata,
                "validate_output": config.validate_output,
            }

        self._current_session = SessionState(
            session_id=session_id,
            status=SessionStatus.ACTIVE,
            started_at=datetime.now(),
            updated_at=datetime.now(),
            current_index=0,
            pending_videos=pending_videos,
            output_dir=output_dir,
            config_snapshot=config_snapshot,
        )

        self._dirty = True
        self.save()

        logger.info(f"Created session {session_id} with {len(pending_videos)} videos")
        return self._current_session

save

save(force=False)

Save current session state to disk.

Parameters:

Name	Type	Description	Default
force	bool	If True, save even if not dirty or interval not reached.	False

Source code in src/video_converter/core/session.py

def save(self, force: bool = False) -> None:
    """Save current session state to disk.

    Args:
        force: If True, save even if not dirty or interval not reached.
    """
    with self._lock:
        if self._current_session is None:
            return

        # Check if we should save based on interval
        if not force and not self._dirty:
            return

        now = datetime.now()
        if not force and self._last_save_time is not None and self._auto_save_interval > 0:
            elapsed = (now - self._last_save_time).total_seconds()
            if elapsed < self._auto_save_interval:
                return

        try:
            state_file = self.state_file_path
            self._current_session.updated_at = now

            with open(state_file, "w", encoding="utf-8") as f:
                json.dump(self._current_session.to_dict(), f, indent=2)

            self._last_save_time = now
            self._dirty = False
            logger.debug(f"Saved session state to {state_file}")
        except OSError as e:
            logger.error(f"Failed to save session state: {e}")

load_session

load_session(session_id=None)

Load a session from disk.

Parameters:

Name	Type	Description	Default
session_id	str | None	Specific session ID to load, or None for current.	None

Returns:

Type	Description
SessionState	The loaded SessionState.

Raises:

Type	Description
SessionNotFoundError	If the session does not exist.
SessionCorruptedError	If the session data is corrupted.

Source code in src/video_converter/core/session.py

def load_session(self, session_id: str | None = None) -> SessionState:
    """Load a session from disk.

    Args:
        session_id: Specific session ID to load, or None for current.

    Returns:
        The loaded SessionState.

    Raises:
        SessionNotFoundError: If the session does not exist.
        SessionCorruptedError: If the session data is corrupted.
    """
    with self._lock:
        if session_id:
            state_file = self.history_dir / f"{session_id}.json"
        else:
            state_file = self.state_file_path

        if not state_file.exists():
            raise SessionNotFoundError(f"Session file not found: {state_file}")

        try:
            with open(state_file, encoding="utf-8") as f:
                data = json.load(f)

            session = SessionState.from_dict(data)
            self._current_session = session
            self._dirty = False
            logger.info(f"Loaded session {session.session_id}")
            return session
        except json.JSONDecodeError as e:
            raise SessionCorruptedError(f"Invalid session data: {e}") from e
        except KeyError as e:
            raise SessionCorruptedError(f"Missing session field: {e}") from e

has_resumable_session

has_resumable_session()

Check if there is a resumable session available.

Returns:

Type	Description
bool	True if a session can be resumed, False otherwise.

Source code in src/video_converter/core/session.py

def has_resumable_session(self) -> bool:
    """Check if there is a resumable session available.

    Returns:
        True if a session can be resumed, False otherwise.
    """
    state_file = self.state_file_path
    if not state_file.exists():
        return False

    try:
        with open(state_file, encoding="utf-8") as f:
            data = json.load(f)

        status = SessionStatus(data.get("status", "active"))
        return status in (SessionStatus.PAUSED, SessionStatus.INTERRUPTED)
    except (json.JSONDecodeError, ValueError):
        return False

get_resumable_sessions

get_resumable_sessions()

Get list of all resumable sessions.

Returns:

Type	Description
list[dict]	List of session info dictionaries with id, status, and progress.

Source code in src/video_converter/core/session.py

def get_resumable_sessions(self) -> list[dict]:
    """Get list of all resumable sessions.

    Returns:
        List of session info dictionaries with id, status, and progress.
    """
    sessions = []

    # Check current session
    if self.state_file_path.exists():
        try:
            with open(self.state_file_path, encoding="utf-8") as f:
                data = json.load(f)

            status = SessionStatus(data.get("status", "active"))
            if status in (SessionStatus.PAUSED, SessionStatus.INTERRUPTED):
                total = (
                    len(data.get("pending_videos", []))
                    + len(data.get("completed_videos", []))
                    + len(data.get("failed_videos", []))
                )
                completed = len(data.get("completed_videos", []))
                sessions.append(
                    {
                        "session_id": data.get("session_id"),
                        "status": status.value,
                        "started_at": data.get("started_at"),
                        "total_videos": total,
                        "completed_videos": completed,
                        "progress": completed / total if total > 0 else 0.0,
                    }
                )
        except (json.JSONDecodeError, ValueError):
            pass

    # Check history directory
    if self.history_dir.exists():
        for session_file in self.history_dir.glob("*.json"):
            try:
                with open(session_file, encoding="utf-8") as f:
                    data = json.load(f)

                status = SessionStatus(data.get("status", "active"))
                if status in (SessionStatus.PAUSED, SessionStatus.INTERRUPTED):
                    total = (
                        len(data.get("pending_videos", []))
                        + len(data.get("completed_videos", []))
                        + len(data.get("failed_videos", []))
                    )
                    completed = len(data.get("completed_videos", []))
                    sessions.append(
                        {
                            "session_id": data.get("session_id"),
                            "status": status.value,
                            "started_at": data.get("started_at"),
                            "total_videos": total,
                            "completed_videos": completed,
                            "progress": completed / total if total > 0 else 0.0,
                        }
                    )
            except (json.JSONDecodeError, ValueError):
                continue

    return sessions

mark_video_completed

mark_video_completed(video, original_size=0, converted_size=0)

Mark a video as successfully completed.

Parameters:

Name	Type	Description	Default
video	VideoEntry	The video entry to mark complete.	required
original_size	int	Size of original file in bytes.	0
converted_size	int	Size of converted file in bytes.	0

Source code in src/video_converter/core/session.py

def mark_video_completed(
    self,
    video: VideoEntry,
    original_size: int = 0,
    converted_size: int = 0,
) -> None:
    """Mark a video as successfully completed.

    Args:
        video: The video entry to mark complete.
        original_size: Size of original file in bytes.
        converted_size: Size of converted file in bytes.
    """
    with self._lock:
        if self._current_session is None:
            return

        self._current_session.mark_video_completed(video, original_size, converted_size)
        self._current_session.current_index += 1
        self._dirty = True
        self.save()

mark_video_failed

mark_video_failed(video, error)

Mark a video as failed.

Parameters:

Name	Type	Description	Default
video	VideoEntry	The video entry to mark failed.	required
error	str	Error message describing the failure.	required

Source code in src/video_converter/core/session.py

def mark_video_failed(self, video: VideoEntry, error: str) -> None:
    """Mark a video as failed.

    Args:
        video: The video entry to mark failed.
        error: Error message describing the failure.
    """
    with self._lock:
        if self._current_session is None:
            return

        self._current_session.mark_video_failed(video, error)
        self._current_session.current_index += 1
        self._dirty = True
        self.save()

add_temporary_file

add_temporary_file(path)

Track a temporary file.

Parameters:

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

Source code in src/video_converter/core/session.py

def add_temporary_file(self, path: Path) -> None:
    """Track a temporary file.

    Args:
        path: Path to the temporary file.
    """
    with self._lock:
        if self._current_session is None:
            return

        self._current_session.add_temporary_file(path)
        self._dirty = True

remove_temporary_file

remove_temporary_file(path)

Stop tracking a temporary file.

Parameters:

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

Source code in src/video_converter/core/session.py

def remove_temporary_file(self, path: Path) -> None:
    """Stop tracking a temporary file.

    Args:
        path: Path to the temporary file.
    """
    with self._lock:
        if self._current_session is None:
            return

        self._current_session.remove_temporary_file(path)
        self._dirty = True

pause_session

pause_session()

Pause the current session.

Returns:

Type	Description
bool	True if paused successfully, False if no active session.

Source code in src/video_converter/core/session.py

def pause_session(self) -> bool:
    """Pause the current session.

    Returns:
        True if paused successfully, False if no active session.
    """
    with self._lock:
        if self._current_session is None:
            return False
        if self._current_session.status != SessionStatus.ACTIVE:
            return False

        self._current_session.status = SessionStatus.PAUSED
        self._current_session.updated_at = datetime.now()
        self._dirty = True
        self.save(force=True)

        logger.info(f"Paused session {self._current_session.session_id}")
        return True

resume_session

resume_session()

Resume a paused or interrupted session.

Returns:

Type	Description
SessionState | None	The resumed SessionState, or None if no resumable session.

Source code in src/video_converter/core/session.py

def resume_session(self) -> SessionState | None:
    """Resume a paused or interrupted session.

    Returns:
        The resumed SessionState, or None if no resumable session.
    """
    with self._lock:
        if self._current_session is None:
            try:
                self.load_session()
            except SessionNotFoundError:
                return None

        if self._current_session is None:
            return None

        if not self._current_session.is_resumable:
            return None

        self._current_session.status = SessionStatus.ACTIVE
        self._current_session.updated_at = datetime.now()
        self._dirty = True
        self.save(force=True)

        logger.info(f"Resumed session {self._current_session.session_id}")
        return self._current_session

complete_session

complete_session()

Mark the current session as completed and archive it.

Source code in src/video_converter/core/session.py

def complete_session(self) -> None:
    """Mark the current session as completed and archive it."""
    with self._lock:
        if self._current_session is None:
            return

        self._current_session.status = SessionStatus.COMPLETED
        self._current_session.updated_at = datetime.now()

        # Archive to history
        self._archive_session()

        # Remove current session file
        if self.state_file_path.exists():
            self.state_file_path.unlink()

        logger.info(f"Completed session {self._current_session.session_id}")
        self._current_session = None
        self._dirty = False

cancel_session

cancel_session()

Cancel the current session and archive it.

Source code in src/video_converter/core/session.py

def cancel_session(self) -> None:
    """Cancel the current session and archive it."""
    with self._lock:
        if self._current_session is None:
            return

        self._current_session.status = SessionStatus.CANCELLED
        self._current_session.updated_at = datetime.now()

        # Clean up temporary files
        self._cleanup_temporary_files()

        # Archive to history
        self._archive_session()

        # Remove current session file
        if self.state_file_path.exists():
            self.state_file_path.unlink()

        logger.info(f"Cancelled session {self._current_session.session_id}")
        self._current_session = None
        self._dirty = False

cleanup_orphaned_temp_files

cleanup_orphaned_temp_files()

Clean up orphaned temporary files from interrupted sessions.

Returns:

Type	Description
list[Path]	List of files that were cleaned up.

Source code in src/video_converter/core/session.py

def cleanup_orphaned_temp_files(self) -> list[Path]:
    """Clean up orphaned temporary files from interrupted sessions.

    Returns:
        List of files that were cleaned up.
    """
    cleaned: list[Path] = []

    with self._lock:
        if self._current_session is None:
            try:
                self.load_session()
            except SessionNotFoundError:
                return cleaned

        if self._current_session is None:
            return cleaned

        cleaned = self._cleanup_temporary_files()

    return cleaned

get_session_status

get_session_status()

Get status summary for the current session.

Returns:

Type	Description
dict | None	Dictionary with session status info, or None if no session.

Source code in src/video_converter/core/session.py

def get_session_status(self) -> dict | None:
    """Get status summary for the current session.

    Returns:
        Dictionary with session status info, or None if no session.
    """
    with self._lock:
        if self._current_session is None:
            return None

        return {
            "session_id": self._current_session.session_id,
            "status": self._current_session.status.value,
            "started_at": self._current_session.started_at.isoformat(),
            "updated_at": self._current_session.updated_at.isoformat(),
            "current_index": self._current_session.current_index,
            "total_videos": self._current_session.total_videos,
            "pending": len(self._current_session.pending_videos),
            "completed": len(self._current_session.completed_videos),
            "failed": len(self._current_session.failed_videos),
            "progress": self._current_session.progress,
            "temporary_files": len(self._current_session.temporary_files),
        }

delete_session

delete_session(session_id)

Delete a session from history.

Parameters:

Name	Type	Description	Default
session_id	str	The session ID to delete.	required

Returns:

Type	Description
bool	True if deleted, False if not found.

Source code in src/video_converter/core/session.py

def delete_session(self, session_id: str) -> bool:
    """Delete a session from history.

    Args:
        session_id: The session ID to delete.

    Returns:
        True if deleted, False if not found.
    """
    with self._lock:
        # Check history
        session_file = self.history_dir / f"{session_id}.json"
        if session_file.exists():
            session_file.unlink()
            logger.info(f"Deleted session {session_id}")
            return True

        # Check current
        if self.state_file_path.exists():
            try:
                with open(self.state_file_path, encoding="utf-8") as f:
                    data = json.load(f)
                if data.get("session_id") == session_id:
                    self.state_file_path.unlink()
                    if self._current_session and self._current_session.session_id == session_id:
                        self._current_session = None
                    logger.info(f"Deleted current session {session_id}")
                    return True
            except (json.JSONDecodeError, OSError):
                pass

        return False

clear_history

clear_history(keep_days=30)

Clear old sessions from history.

Parameters:

Name	Type	Description	Default
keep_days	int	Number of days of history to keep.	30

Returns:

Type	Description
int	Number of sessions deleted.

Source code in src/video_converter/core/session.py

def clear_history(self, keep_days: int = 30) -> int:
    """Clear old sessions from history.

    Args:
        keep_days: Number of days of history to keep.

    Returns:
        Number of sessions deleted.
    """
    deleted = 0
    cutoff = datetime.now().timestamp() - (keep_days * 24 * 60 * 60)

    if not self.history_dir.exists():
        return 0

    for session_file in self.history_dir.glob("*.json"):
        try:
            if session_file.stat().st_mtime < cutoff:
                session_file.unlink()
                deleted += 1
        except OSError:
            continue

    if deleted > 0:
        logger.info(f"Cleared {deleted} old sessions from history")

    return deleted

get_session_manager

get_session_manager(state_dir=None)

Get or create the default session state manager.

Parameters:

Name	Type	Description	Default
state_dir	Path | None	Optional state directory. Only used on first call.	None

Returns:

Type	Description
SessionStateManager	The default SessionStateManager instance.

Source code in src/video_converter/core/session.py

def get_session_manager(state_dir: Path | None = None) -> SessionStateManager:
    """Get or create the default session state manager.

    Args:
        state_dir: Optional state directory. Only used on first call.

    Returns:
        The default SessionStateManager instance.
    """
    global _default_manager
    with _manager_lock:
        if _default_manager is None:
            _default_manager = SessionStateManager(state_dir=state_dir)
        return _default_manager

History

history

Conversion history tracking for duplicate prevention.

This module implements a conversion history tracking system to prevent re-converting videos that have already been processed. It uses file hashes or Photos UUIDs to identify videos across renames and moves.

SDS Reference: SDS-C01-003 SRS Reference: SRS-306 (Conversion History)

Example

from video_converter.core.history import ConversionHistory, ConversionRecord from pathlib import Path from datetime import datetime

history = ConversionHistory()

Check before converting

if not history.is_converted(video_uuid): ... # Perform conversion... ... record = ConversionRecord( ... id=video_uuid, ... source_path=str(source), ... output_path=str(output), ... source_codec="h264", ... output_codec="hevc", ... source_size=original_size, ... output_size=converted_size, ... converted_at=datetime.now().isoformat(), ... success=True, ... ) ... history.add_record(record)

StatsPeriod

Bases: Enum

Time period for statistics filtering.

HistoryError

Bases: Exception

Base exception for history operations.

HistoryCorruptedError

Bases: HistoryError

Raised when history data is corrupted or invalid.

ConversionRecord dataclass

ConversionRecord(id, source_path, output_path, source_codec, output_codec, source_size, output_size, converted_at, success, error_message=None)

Record of a video conversion.

Stores all relevant information about a single video conversion, including paths, codecs, sizes, and success status.

Attributes:

Name	Type	Description
id	str	Unique identifier (UUID from Photos or file hash).
source_path	str	Original file path at time of conversion.
output_path	str | None	Converted file path (None if failed).
source_codec	str	Original video codec (e.g., "h264").
output_codec	str	Target video codec (e.g., "hevc").
source_size	int	Original file size in bytes.
output_size	int | None	Converted file size in bytes (None if failed).
converted_at	str	ISO format timestamp of conversion.
success	bool	Whether conversion completed successfully.
error_message	str | None	Error details if conversion failed.

size_saved property

size_saved

Calculate bytes saved by conversion.

Returns:

Type	Description
int	Bytes saved (0 if conversion failed or increased size).

compression_ratio property

compression_ratio

Calculate compression ratio.

Returns:

Type	Description
float	Compression ratio (0.0-1.0) where 0.6 means 60% size reduction.
float	Returns 0.0 if conversion failed or source size is 0.

to_dict

to_dict()

Convert record to JSON-serializable dictionary.

Returns:

Type	Description
dict	Dictionary representation of the record.

Source code in src/video_converter/core/history.py

def to_dict(self) -> dict:
    """Convert record to JSON-serializable dictionary.

    Returns:
        Dictionary representation of the record.
    """
    return asdict(self)

from_dict classmethod

from_dict(data)

Create a ConversionRecord from dictionary.

Parameters:

Name	Type	Description	Default
data	dict	Dictionary containing record data.	required

Returns:

Type	Description
ConversionRecord	A new ConversionRecord instance.

Raises:

Type	Description
KeyError	If required fields are missing.

Source code in src/video_converter/core/history.py

@classmethod
def from_dict(cls, data: dict) -> ConversionRecord:
    """Create a ConversionRecord from dictionary.

    Args:
        data: Dictionary containing record data.

    Returns:
        A new ConversionRecord instance.

    Raises:
        KeyError: If required fields are missing.
    """
    return cls(
        id=data["id"],
        source_path=data["source_path"],
        output_path=data.get("output_path"),
        source_codec=data["source_codec"],
        output_codec=data["output_codec"],
        source_size=data["source_size"],
        output_size=data.get("output_size"),
        converted_at=data["converted_at"],
        success=data["success"],
        error_message=data.get("error_message"),
    )

HistoryStatistics dataclass

HistoryStatistics(total_converted=0, total_failed=0, total_source_bytes=0, total_output_bytes=0, total_saved_bytes=0, first_conversion=None, last_conversion=None, period='all', period_start=None, total_duration_seconds=0.0)

Statistics summary for conversion history.

Attributes:

Name	Type	Description
total_converted	int	Number of successfully converted videos.
total_failed	int	Number of failed conversions.
total_source_bytes	int	Sum of all original file sizes.
total_output_bytes	int	Sum of all converted file sizes.
total_saved_bytes	int	Total bytes saved by compression.
first_conversion	str | None	Timestamp of earliest conversion.
last_conversion	str | None	Timestamp of most recent conversion.
period	str	The time period these statistics cover.
period_start	str | None	Start timestamp of the period.
total_duration_seconds	float	Total conversion time in seconds.

success_rate property

success_rate

Calculate success rate.

Returns:

Type	Description
float	Success rate as a fraction (0.0-1.0).

average_compression_ratio property

average_compression_ratio

Calculate average compression ratio.

Returns:

Type	Description
float	Average compression ratio (0.0-1.0).

storage_saved_percent property

storage_saved_percent

Calculate storage saved percentage.

Returns:

Type	Description
float	Percentage of storage saved (0.0-100.0).

to_dict

to_dict()

Convert statistics to dictionary.

Returns:

Type	Description
dict	Dictionary representation of statistics.

Source code in src/video_converter/core/history.py

def to_dict(self) -> dict:
    """Convert statistics to dictionary.

    Returns:
        Dictionary representation of statistics.
    """
    return {
        "period": self.period,
        "period_start": self.period_start,
        "total_converted": self.total_converted,
        "total_failed": self.total_failed,
        "total_source_bytes": self.total_source_bytes,
        "total_output_bytes": self.total_output_bytes,
        "total_saved_bytes": self.total_saved_bytes,
        "storage_saved_percent": self.storage_saved_percent,
        "first_conversion": self.first_conversion,
        "last_conversion": self.last_conversion,
        "success_rate": self.success_rate,
        "average_compression_ratio": self.average_compression_ratio,
        "total_duration_seconds": self.total_duration_seconds,
    }

ConversionHistory

ConversionHistory(history_path=None)

Manage conversion history for duplicate prevention.

Provides functionality to: - Track converted videos by unique identifier - Store conversion metadata (date, result, output path) - Support Photos UUID and file hash identification - Persist history across application restarts - Generate statistics and reports

Thread-safe for concurrent access.

Attributes:

Name	Type	Description
history_path		Path to the history JSON file.

Example

history = ConversionHistory()

Check if already converted

if history.is_converted(video_id): ... print("Already converted, skipping") ...

Add conversion record

record = ConversionRecord(...) history.add_record(record)

Get statistics

stats = history.get_statistics() print(f"Converted: {stats.total_converted} videos")

Initialize the conversion history manager.

Parameters:

Name	Type	Description	Default
history_path	Path | None	Path to history file. Defaults to ~/.local/share/video_converter/history.json	None

Source code in src/video_converter/core/history.py

def __init__(self, history_path: Path | None = None) -> None:
    """Initialize the conversion history manager.

    Args:
        history_path: Path to history file.
                     Defaults to ~/.local/share/video_converter/history.json
    """
    self.history_path = history_path or DEFAULT_HISTORY_FILE
    self._records: dict[str, ConversionRecord] = {}
    self._lock = threading.RLock()  # RLock for reentrant locking
    self._dirty = False

    # Ensure directory exists
    self.history_path.parent.mkdir(parents=True, exist_ok=True)

    # Load existing history
    self._load()

is_converted

is_converted(identifier)

Check if a video has been successfully converted.

Parameters:

Name	Type	Description	Default
identifier	str	UUID (from Photos) or file hash.	required

Returns:

Type	Description
bool	True if video was successfully converted before.

Source code in src/video_converter/core/history.py

def is_converted(self, identifier: str) -> bool:
    """Check if a video has been successfully converted.

    Args:
        identifier: UUID (from Photos) or file hash.

    Returns:
        True if video was successfully converted before.
    """
    with self._lock:
        record = self._records.get(identifier)
        return record is not None and record.success

add_record

add_record(record)

Add or update a conversion record.

Parameters:

Name	Type	Description	Default
record	ConversionRecord	The conversion record to add.	required

Source code in src/video_converter/core/history.py

def add_record(self, record: ConversionRecord) -> None:
    """Add or update a conversion record.

    Args:
        record: The conversion record to add.
    """
    with self._lock:
        self._records[record.id] = record
        self._dirty = True
        self._save()

    logger.debug(f"Added record for {record.id}")

get_record

get_record(identifier)

Get a conversion record by identifier.

Parameters:

Name	Type	Description	Default
identifier	str	UUID or file hash.	required

Returns:

Type	Description
ConversionRecord | None	The ConversionRecord if found, None otherwise.

Source code in src/video_converter/core/history.py

def get_record(self, identifier: str) -> ConversionRecord | None:
    """Get a conversion record by identifier.

    Args:
        identifier: UUID or file hash.

    Returns:
        The ConversionRecord if found, None otherwise.
    """
    with self._lock:
        return self._records.get(identifier)

remove_record

remove_record(identifier)

Remove a record from history.

Parameters:

Name	Type	Description	Default
identifier	str	UUID or file hash.	required

Returns:

Type	Description
bool	True if record was removed, False if not found.

Source code in src/video_converter/core/history.py

def remove_record(self, identifier: str) -> bool:
    """Remove a record from history.

    Args:
        identifier: UUID or file hash.

    Returns:
        True if record was removed, False if not found.
    """
    with self._lock:
        if identifier in self._records:
            del self._records[identifier]
            self._dirty = True
            self._save()
            logger.debug(f"Removed record for {identifier}")
            return True
        return False

clear

clear()

Clear all history records.

Source code in src/video_converter/core/history.py

def clear(self) -> None:
    """Clear all history records."""
    with self._lock:
        count = len(self._records)
        self._records.clear()
        self._dirty = True
        self._save()

    logger.info(f"Cleared {count} records from history")

get_records_by_period

get_records_by_period(period=ALL)

Get all conversion records filtered by time period.

Parameters:

Name	Type	Description	Default
period	StatsPeriod	Time period to filter by.	ALL

Returns:

Type	Description
list[ConversionRecord]	List of ConversionRecord instances within the period.

Source code in src/video_converter/core/history.py

def get_records_by_period(
    self,
    period: StatsPeriod = StatsPeriod.ALL,
) -> list[ConversionRecord]:
    """Get all conversion records filtered by time period.

    Args:
        period: Time period to filter by.

    Returns:
        List of ConversionRecord instances within the period.
    """
    with self._lock:
        all_records = list(self._records.values())
        return self._filter_records_by_period(all_records, period)

get_statistics

get_statistics(period=ALL)

Get aggregated statistics for conversions.

Parameters:

Name	Type	Description	Default
period	StatsPeriod	Time period to filter statistics by.	ALL

Returns:

Type	Description
HistoryStatistics	HistoryStatistics with totals and averages.

Source code in src/video_converter/core/history.py

def get_statistics(
    self,
    period: StatsPeriod = StatsPeriod.ALL,
) -> HistoryStatistics:
    """Get aggregated statistics for conversions.

    Args:
        period: Time period to filter statistics by.

    Returns:
        HistoryStatistics with totals and averages.
    """
    with self._lock:
        stats = HistoryStatistics()
        stats.period = period.value

        # Calculate period start
        period_start = self._get_period_start(period)
        if period_start:
            stats.period_start = period_start.isoformat()

        # Filter records by period
        all_records = list(self._records.values())
        filtered_records = self._filter_records_by_period(all_records, period)

        successful = [r for r in filtered_records if r.success]
        failed = [r for r in filtered_records if not r.success]

        stats.total_converted = len(successful)
        stats.total_failed = len(failed)

        for record in successful:
            stats.total_source_bytes += record.source_size
            if record.output_size is not None:
                stats.total_output_bytes += record.output_size

        stats.total_saved_bytes = stats.total_source_bytes - stats.total_output_bytes

        # Find first and last conversion times within period
        timestamps = [r.converted_at for r in successful if r.converted_at]
        if timestamps:
            stats.first_conversion = min(timestamps)
            stats.last_conversion = max(timestamps)

        return stats

get_all_records

get_all_records()

Get all conversion records.

Returns:

Type	Description
list[ConversionRecord]	List of all ConversionRecord instances.

Source code in src/video_converter/core/history.py

def get_all_records(self) -> list[ConversionRecord]:
    """Get all conversion records.

    Returns:
        List of all ConversionRecord instances.
    """
    with self._lock:
        return list(self._records.values())

get_failed_records

get_failed_records()

Get all failed conversion records.

Returns:

Type	Description
list[ConversionRecord]	List of failed ConversionRecord instances.

Source code in src/video_converter/core/history.py

def get_failed_records(self) -> list[ConversionRecord]:
    """Get all failed conversion records.

    Returns:
        List of failed ConversionRecord instances.
    """
    with self._lock:
        return [r for r in self._records.values() if not r.success]

get_successful_records

get_successful_records()

Get all successful conversion records.

Returns:

Type	Description
list[ConversionRecord]	List of successful ConversionRecord instances.

Source code in src/video_converter/core/history.py

def get_successful_records(self) -> list[ConversionRecord]:
    """Get all successful conversion records.

    Returns:
        List of successful ConversionRecord instances.
    """
    with self._lock:
        return [r for r in self._records.values() if r.success]

count

count()

Get total number of records.

Returns:

Type	Description
int	Number of records in history.

Source code in src/video_converter/core/history.py

def count(self) -> int:
    """Get total number of records.

    Returns:
        Number of records in history.
    """
    with self._lock:
        return len(self._records)

export_to_json

export_to_json(path)

Export history to a JSON file.

Parameters:

Name	Type	Description	Default
path	Path	Path to export file.	required

Source code in src/video_converter/core/history.py

def export_to_json(self, path: Path) -> None:
    """Export history to a JSON file.

    Args:
        path: Path to export file.
    """
    with self._lock:
        data = {
            "exported_at": datetime.now().isoformat(),
            "statistics": self.get_statistics().to_dict(),
            "records": [r.to_dict() for r in self._records.values()],
        }

        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
            f.write("\n")

    logger.info(f"Exported history to {path}")

export_to_csv

export_to_csv(path)

Export history to a CSV file.

Parameters:

Name	Type	Description	Default
path	Path	Path to export file.	required

Source code in src/video_converter/core/history.py

def export_to_csv(self, path: Path) -> None:
    """Export history to a CSV file.

    Args:
        path: Path to export file.
    """
    import csv

    with self._lock, open(path, "w", newline="", encoding="utf-8") as f:
        fieldnames = [
            "id",
            "source_path",
            "output_path",
            "source_codec",
            "output_codec",
            "source_size",
            "output_size",
            "converted_at",
            "success",
            "error_message",
            "size_saved",
            "compression_ratio",
        ]
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()

        for record in self._records.values():
            row = record.to_dict()
            row["size_saved"] = record.size_saved
            row["compression_ratio"] = f"{record.compression_ratio:.2%}"
            writer.writerow(row)

    logger.info(f"Exported history to {path}")

compute_file_hash staticmethod

compute_file_hash(path)

Compute a hash for file identification.

Uses SHA-256 hash of first and last 1MB of file for efficient identification of large files.

Parameters:

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

Returns:

Type	Description
str	16-character hex hash string.

Raises:

Type	Description
FileNotFoundError	If file does not exist.
OSError	If file cannot be read.

Source code in src/video_converter/core/history.py

@staticmethod
def compute_file_hash(path: Path) -> str:
    """Compute a hash for file identification.

    Uses SHA-256 hash of first and last 1MB of file for
    efficient identification of large files.

    Args:
        path: Path to the file.

    Returns:
        16-character hex hash string.

    Raises:
        FileNotFoundError: If file does not exist.
        OSError: If file cannot be read.
    """
    sha = hashlib.sha256()
    file_size = path.stat().st_size

    with open(path, "rb") as f:
        # Hash first 1MB
        sha.update(f.read(1024 * 1024))

        # If file is larger than 2MB, also hash last 1MB
        if file_size > 2 * 1024 * 1024:
            f.seek(-1024 * 1024, 2)  # Seek to last 1MB
            sha.update(f.read())
        elif file_size > 1024 * 1024:
            # File between 1MB and 2MB, hash remaining
            sha.update(f.read())

    return sha.hexdigest()[:16]

get_history

get_history(history_path=None)

Get or create the default ConversionHistory instance.

Parameters:

Name	Type	Description	Default
history_path	Path | None	Optional path. Only used on first call.	None

Returns:

Type	Description
ConversionHistory	The default ConversionHistory instance.

Source code in src/video_converter/core/history.py

def get_history(history_path: Path | None = None) -> ConversionHistory:
    """Get or create the default ConversionHistory instance.

    Args:
        history_path: Optional path. Only used on first call.

    Returns:
        The default ConversionHistory instance.
    """
    global _default_history
    with _history_lock:
        if _default_history is None:
            _default_history = ConversionHistory(history_path=history_path)
        return _default_history

reset_history

reset_history()

Reset the default history instance.

Primarily useful for testing.

Source code in src/video_converter/core/history.py

def reset_history() -> None:
    """Reset the default history instance.

    Primarily useful for testing.
    """
    global _default_history
    with _history_lock:
        _default_history = None