Automation Module

Automation and scheduling components for background processing.

Service Manager

service_manager

Service manager for macOS launchd automation.

This module provides a high-level interface for installing, uninstalling, and managing the video converter launchd service on macOS.

SDS Reference: SDS-A01-002 SRS Reference: SRS-702 (Service Install/Uninstall)

Example

from video_converter.automation.service_manager import ServiceManager

manager = ServiceManager()

Install service to run daily at 3 AM

result = manager.install(hour=3, minute=0) print(result.message)

Check service status

status = manager.get_status() print(f"Installed: {status.is_installed}")

Uninstall service

result = manager.uninstall()

ServiceState

Bases: Enum

State of the launchd service.

ServiceStatus dataclass

ServiceStatus(state, is_installed, is_loaded, is_running, pid=None, last_exit_status=None, plist_path=None, schedule=None, next_run=None)

Status information for the launchd service.

Attributes:

Name	Type	Description
state	ServiceState	Current state of the service.
is_installed	bool	Whether the plist file exists.
is_loaded	bool	Whether the service is loaded in launchd.
is_running	bool	Whether the service is currently running.
pid	int | None	Process ID if running, None otherwise.
last_exit_status	int | None	Exit status from last run, None if never run.
plist_path	Path | None	Path to the plist file.
schedule	str | None	Human-readable schedule description.
next_run	datetime | None	Estimated next run time, None if not scheduled.

ServiceResult dataclass

ServiceResult(success, message, error=None, plist_path=None)

Result of a service operation.

Attributes:

Name	Type	Description
success	bool	Whether the operation succeeded.
message	str	Human-readable result message.
error	str | None	Error message if failed, None otherwise.
plist_path	Path | None	Path to the plist file, if applicable.

LastRunInfo dataclass

LastRunInfo(timestamp=None, success=None, videos_converted=0, storage_saved_bytes=0, error_message=None)

Information about the last service run.

Attributes:

Name	Type	Description
timestamp	datetime | None	When the service last ran.
success	bool | None	Whether the last run was successful.
videos_converted	int	Number of videos converted in last run.
storage_saved_bytes	int	Bytes saved in last run.
error_message	str | None	Error message if last run failed.

relative_time property

relative_time

Get relative time description (e.g., 'Today', 'Yesterday').

Returns:

Type	Description
str	Human-readable relative time string.

result_text property

result_text

Get result description.

Returns:

Type	Description
str	Human-readable result string.

DetailedServiceStatus dataclass

DetailedServiceStatus(basic_status, next_run=None, next_run_relative='Unknown', last_run=LastRunInfo(), total_videos_converted=0, total_storage_saved_bytes=0, success_rate=0.0)

Extended status information including history and next run.

Attributes:

Name	Type	Description
basic_status	ServiceStatus	Basic ServiceStatus information.
next_run	datetime | None	Estimated next run time.
next_run_relative	str	Human-readable next run description.
last_run	LastRunInfo	Information about the last run.
total_videos_converted	int	Total videos converted (all time).
total_storage_saved_bytes	int	Total storage saved (all time).
success_rate	float	Overall success rate (0.0-1.0).

ServiceManager

ServiceManager(plist_path=None, log_dir=None)

Manager for launchd service operations.

This class provides high-level operations for installing, uninstalling, and querying the status of the video converter launchd service.

Attributes:

Name	Type	Description
plist_path		Path to the launchd plist file.
log_dir		Directory for service log files.

Initialize the service manager.

Parameters:

Name	Type	Description	Default
plist_path	Path | None	Path to plist file. Defaults to LaunchAgents directory.	None
log_dir	Path | None	Directory for log files.	None

Source code in src/video_converter/automation/service_manager.py

def __init__(
    self,
    plist_path: Path | None = None,
    log_dir: Path | None = None,
) -> None:
    """Initialize the service manager.

    Args:
        plist_path: Path to plist file. Defaults to LaunchAgents directory.
        log_dir: Directory for log files.
    """
    self.plist_path = plist_path or (DEFAULT_LAUNCH_AGENTS_DIR / DEFAULT_PLIST_NAME)
    self.log_dir = log_dir or DEFAULT_LOG_DIR
    self._generator = LaunchdPlistGenerator(log_dir=self.log_dir)

install

install(hour=3, minute=0, weekday=None, watch_paths=None, run_at_load=False, force=False)

Install the launchd service.

This method generates a plist file and loads it into launchd. If the service is already installed, it will be updated if force=True.

Parameters:

Name	Type	Description	Default
hour	int | None	Hour to run (0-23). None for watch-only mode.	3
minute	int	Minute to run (0-59).	0
weekday	int | None	Day of week (0=Sunday). None for daily.	None
watch_paths	list[Path] | None	Folders to watch for changes.	None
run_at_load	bool	Whether to run immediately when loaded.	False
force	bool	If True, reinstall even if already installed.	False

Returns:

Type	Description
ServiceResult	ServiceResult with success status and message.

Example

manager = ServiceManager() result = manager.install(hour=3, minute=0) if result.success: ... print("Service installed successfully")

Source code in src/video_converter/automation/service_manager.py

def install(
    self,
    hour: int | None = 3,
    minute: int = 0,
    weekday: int | None = None,
    watch_paths: list[Path] | None = None,
    run_at_load: bool = False,
    force: bool = False,
) -> ServiceResult:
    """Install the launchd service.

    This method generates a plist file and loads it into launchd.
    If the service is already installed, it will be updated if force=True.

    Args:
        hour: Hour to run (0-23). None for watch-only mode.
        minute: Minute to run (0-59).
        weekday: Day of week (0=Sunday). None for daily.
        watch_paths: Folders to watch for changes.
        run_at_load: Whether to run immediately when loaded.
        force: If True, reinstall even if already installed.

    Returns:
        ServiceResult with success status and message.

    Example:
        >>> manager = ServiceManager()
        >>> result = manager.install(hour=3, minute=0)
        >>> if result.success:
        ...     print("Service installed successfully")
    """
    # Check if already installed
    status = self.get_status()
    if status.is_installed and not force:
        return ServiceResult(
            success=False,
            message="Service is already installed. Use --force to reinstall.",
            error="Service already exists",
            plist_path=self.plist_path,
        )

    # Unload existing service if loaded
    if status.is_loaded:
        unload_result = self._unload_service()
        if not unload_result.success:
            return unload_result

    try:
        # Generate plist
        plist = self._generator.generate_plist(
            hour=hour,
            minute=minute,
            weekday=weekday,
            watch_paths=watch_paths,
            run_at_load=run_at_load,
        )

        # Write plist to file
        self._generator.write_plist(plist, self.plist_path)

        # Validate plist syntax
        if not validate_plist_syntax(self.plist_path):
            return ServiceResult(
                success=False,
                message="Generated plist file is invalid",
                error="Plist validation failed",
                plist_path=self.plist_path,
            )

        # Load service
        load_result = self._load_service()
        if not load_result.success:
            return load_result

        # Build schedule description
        schedule_desc = self._build_schedule_description(hour, minute, weekday, watch_paths)

        logger.info(f"Service installed successfully: {schedule_desc}")
        return ServiceResult(
            success=True,
            message=f"Service installed successfully. Schedule: {schedule_desc}",
            plist_path=self.plist_path,
        )

    except Exception as e:
        logger.exception("Failed to install service")
        return ServiceResult(
            success=False,
            message="Failed to install service",
            error=str(e),
        )

uninstall

uninstall(remove_logs=False)

Uninstall the launchd service.

This method unloads the service from launchd and removes the plist file.

Parameters:

Name	Type	Description	Default
remove_logs	bool	If True, also remove log files.	False

Returns:

Type	Description
ServiceResult	ServiceResult with success status and message.

Example

manager = ServiceManager() result = manager.uninstall() if result.success: ... print("Service uninstalled successfully")

Source code in src/video_converter/automation/service_manager.py

def uninstall(self, remove_logs: bool = False) -> ServiceResult:
    """Uninstall the launchd service.

    This method unloads the service from launchd and removes the plist file.

    Args:
        remove_logs: If True, also remove log files.

    Returns:
        ServiceResult with success status and message.

    Example:
        >>> manager = ServiceManager()
        >>> result = manager.uninstall()
        >>> if result.success:
        ...     print("Service uninstalled successfully")
    """
    status = self.get_status()

    if not status.is_installed:
        return ServiceResult(
            success=True,
            message="Service is not installed",
        )

    try:
        # Unload service if loaded
        if status.is_loaded:
            unload_result = self._unload_service()
            if not unload_result.success:
                logger.warning(f"Failed to unload service: {unload_result.error}")

        # Remove plist file
        if self.plist_path.exists():
            self.plist_path.unlink()
            logger.info(f"Removed plist file: {self.plist_path}")

        # Remove logs if requested
        if remove_logs and self.log_dir.exists():
            for log_file in self.log_dir.glob("*.log"):
                log_file.unlink()
                logger.info(f"Removed log file: {log_file}")

        return ServiceResult(
            success=True,
            message="Service uninstalled successfully",
        )

    except Exception as e:
        logger.exception("Failed to uninstall service")
        return ServiceResult(
            success=False,
            message="Failed to uninstall service",
            error=str(e),
        )

get_status

get_status()

Get the current status of the service.

Returns:

Type	Description
ServiceStatus	ServiceStatus with detailed status information.

Example

manager = ServiceManager() status = manager.get_status() print(f"Installed: {status.is_installed}") print(f"Running: {status.is_running}")

Source code in src/video_converter/automation/service_manager.py

def get_status(self) -> ServiceStatus:
    """Get the current status of the service.

    Returns:
        ServiceStatus with detailed status information.

    Example:
        >>> manager = ServiceManager()
        >>> status = manager.get_status()
        >>> print(f"Installed: {status.is_installed}")
        >>> print(f"Running: {status.is_running}")
    """
    is_installed = self.plist_path.exists()
    is_loaded = False
    is_running = False
    pid = None
    last_exit_status = None
    schedule = None

    if is_installed:
        # Query launchctl for service status
        launchctl_info = self._get_launchctl_info()
        is_loaded = launchctl_info.get("loaded", False)
        pid = launchctl_info.get("pid")
        last_exit_status = launchctl_info.get("last_exit_status")
        is_running = pid is not None

        # Parse schedule from plist
        schedule = self._get_schedule_from_plist()

    # Determine state
    if not is_installed:
        state = ServiceState.NOT_INSTALLED
    elif is_running:
        state = ServiceState.INSTALLED_RUNNING
    elif last_exit_status is not None and last_exit_status != 0:
        state = ServiceState.INSTALLED_ERROR
    else:
        state = ServiceState.INSTALLED_IDLE

    return ServiceStatus(
        state=state,
        is_installed=is_installed,
        is_loaded=is_loaded,
        is_running=is_running,
        pid=pid,
        last_exit_status=last_exit_status,
        plist_path=self.plist_path if is_installed else None,
        schedule=schedule,
    )

start

start()

Manually start the service.

This triggers an immediate run of the service, regardless of schedule.

Returns:

Type	Description
ServiceResult	ServiceResult with success status and message.

Source code in src/video_converter/automation/service_manager.py

def start(self) -> ServiceResult:
    """Manually start the service.

    This triggers an immediate run of the service, regardless of schedule.

    Returns:
        ServiceResult with success status and message.
    """
    status = self.get_status()

    if not status.is_installed:
        return ServiceResult(
            success=False,
            message="Service is not installed",
            error="Service not found",
        )

    if not status.is_loaded:
        load_result = self._load_service()
        if not load_result.success:
            return load_result

    try:
        result = subprocess.run(
            ["launchctl", "start", SERVICE_LABEL],
            capture_output=True,
            text=True,
            check=False,
        )

        if result.returncode == 0:
            return ServiceResult(
                success=True,
                message="Service started successfully",
            )
        else:
            return ServiceResult(
                success=False,
                message="Failed to start service",
                error=result.stderr.strip() or "Unknown error",
            )

    except Exception as e:
        return ServiceResult(
            success=False,
            message="Failed to start service",
            error=str(e),
        )

stop

stop()

Stop the currently running service.

Returns:

Type	Description
ServiceResult	ServiceResult with success status and message.

Source code in src/video_converter/automation/service_manager.py

def stop(self) -> ServiceResult:
    """Stop the currently running service.

    Returns:
        ServiceResult with success status and message.
    """
    status = self.get_status()

    if not status.is_running:
        return ServiceResult(
            success=True,
            message="Service is not running",
        )

    try:
        result = subprocess.run(
            ["launchctl", "stop", SERVICE_LABEL],
            capture_output=True,
            text=True,
            check=False,
        )

        if result.returncode == 0:
            return ServiceResult(
                success=True,
                message="Service stopped successfully",
            )
        else:
            return ServiceResult(
                success=False,
                message="Failed to stop service",
                error=result.stderr.strip() or "Unknown error",
            )

    except Exception as e:
        return ServiceResult(
            success=False,
            message="Failed to stop service",
            error=str(e),
        )

load

load()

Load the service into launchd.

This is the public wrapper for loading the service plist into launchd. The service must be installed (plist file exists) before loading.

Returns:

Type	Description
ServiceResult	ServiceResult with success status and message.

Example

manager = ServiceManager() result = manager.load() if result.success: ... print("Service loaded into launchd")

Source code in src/video_converter/automation/service_manager.py

def load(self) -> ServiceResult:
    """Load the service into launchd.

    This is the public wrapper for loading the service plist into launchd.
    The service must be installed (plist file exists) before loading.

    Returns:
        ServiceResult with success status and message.

    Example:
        >>> manager = ServiceManager()
        >>> result = manager.load()
        >>> if result.success:
        ...     print("Service loaded into launchd")
    """
    status = self.get_status()

    if not status.is_installed:
        return ServiceResult(
            success=False,
            message="Service is not installed. Run 'install-service' first.",
            error="Plist file not found",
        )

    if status.is_loaded:
        return ServiceResult(
            success=True,
            message="Service is already loaded",
        )

    # Check plist file permissions
    perm_result = self._check_plist_permissions()
    if not perm_result.success:
        return perm_result

    return self._load_service()

unload

unload()

Unload the service from launchd.

This is the public wrapper for unloading the service from launchd. The service will stop running but the plist file remains installed.

Returns:

Type	Description
ServiceResult	ServiceResult with success status and message.

Example

manager = ServiceManager() result = manager.unload() if result.success: ... print("Service unloaded from launchd")

Source code in src/video_converter/automation/service_manager.py

def unload(self) -> ServiceResult:
    """Unload the service from launchd.

    This is the public wrapper for unloading the service from launchd.
    The service will stop running but the plist file remains installed.

    Returns:
        ServiceResult with success status and message.

    Example:
        >>> manager = ServiceManager()
        >>> result = manager.unload()
        >>> if result.success:
        ...     print("Service unloaded from launchd")
    """
    status = self.get_status()

    if not status.is_installed:
        return ServiceResult(
            success=False,
            message="Service is not installed",
            error="Plist file not found",
        )

    if not status.is_loaded:
        return ServiceResult(
            success=True,
            message="Service is already unloaded",
        )

    return self._unload_service()

restart

restart()

Restart the service.

Unloads and reloads the service from launchd.

Returns:

Type	Description
ServiceResult	ServiceResult with success status and message.

Example

manager = ServiceManager() result = manager.restart() if result.success: ... print("Service restarted")

Source code in src/video_converter/automation/service_manager.py

def restart(self) -> ServiceResult:
    """Restart the service.

    Unloads and reloads the service from launchd.

    Returns:
        ServiceResult with success status and message.

    Example:
        >>> manager = ServiceManager()
        >>> result = manager.restart()
        >>> if result.success:
        ...     print("Service restarted")
    """
    status = self.get_status()

    if not status.is_installed:
        return ServiceResult(
            success=False,
            message="Service is not installed",
            error="Plist file not found",
        )

    # Unload if loaded
    if status.is_loaded:
        unload_result = self._unload_service()
        if not unload_result.success:
            return ServiceResult(
                success=False,
                message="Failed to restart service",
                error=f"Unload failed: {unload_result.error}",
            )

    # Load the service
    load_result = self._load_service()
    if not load_result.success:
        return ServiceResult(
            success=False,
            message="Failed to restart service",
            error=f"Load failed: {load_result.error}",
        )

    return ServiceResult(
        success=True,
        message="Service restarted successfully",
    )

get_log_paths

get_log_paths()

Get paths to stdout and stderr log files.

Returns:

Type	Description
tuple[Path, Path]	Tuple of (stdout_path, stderr_path).

Source code in src/video_converter/automation/service_manager.py

def get_log_paths(self) -> tuple[Path, Path]:
    """Get paths to stdout and stderr log files.

    Returns:
        Tuple of (stdout_path, stderr_path).
    """
    return (
        self.log_dir / "stdout.log",
        self.log_dir / "stderr.log",
    )

read_logs

read_logs(lines=50)

Read recent log entries.

Parameters:

Name	Type	Description	Default
lines	int	Number of lines to read from each log file.	50

Returns:

Type	Description
dict[str, str]	Dictionary with 'stdout' and 'stderr' log contents.

Source code in src/video_converter/automation/service_manager.py

def read_logs(self, lines: int = 50) -> dict[str, str]:
    """Read recent log entries.

    Args:
        lines: Number of lines to read from each log file.

    Returns:
        Dictionary with 'stdout' and 'stderr' log contents.
    """
    stdout_path, stderr_path = self.get_log_paths()
    logs = {"stdout": "", "stderr": ""}

    for key, path in [("stdout", stdout_path), ("stderr", stderr_path)]:
        if path.exists():
            try:
                content = path.read_text()
                log_lines = content.splitlines()
                logs[key] = "\n".join(log_lines[-lines:])
            except Exception as e:
                logs[key] = f"Error reading log: {e}"

    return logs

calculate_next_run

calculate_next_run()

Calculate the next scheduled run time.

Parses the plist file to determine when the service will next run based on StartCalendarInterval settings.

Returns:

Type	Description
datetime | None	Tuple of (next_run_datetime, human_readable_description).
str	Returns (None, "Not scheduled") if no schedule is configured.

Example

manager = ServiceManager() next_run, description = manager.calculate_next_run() print(f"Next run: {description}")

Source code in src/video_converter/automation/service_manager.py

def calculate_next_run(self) -> tuple[datetime | None, str]:
    """Calculate the next scheduled run time.

    Parses the plist file to determine when the service will next run
    based on StartCalendarInterval settings.

    Returns:
        Tuple of (next_run_datetime, human_readable_description).
        Returns (None, "Not scheduled") if no schedule is configured.

    Example:
        >>> manager = ServiceManager()
        >>> next_run, description = manager.calculate_next_run()
        >>> print(f"Next run: {description}")
    """
    if not self.plist_path.exists():
        return None, "Not installed"

    try:
        import plistlib

        with self.plist_path.open("rb") as f:
            plist = plistlib.load(f)

        # Check StartCalendarInterval
        if "StartCalendarInterval" not in plist:
            # Check for WatchPaths only
            if "WatchPaths" in plist:
                return None, "Triggered by folder changes"
            return None, "Not scheduled"

        interval = plist["StartCalendarInterval"]
        hour = interval.get("Hour")
        minute = interval.get("Minute", 0)
        weekday = interval.get("Weekday")

        if hour is None:
            return None, "No time specified"

        now = datetime.now()
        next_run = now.replace(hour=hour, minute=minute, second=0, microsecond=0)

        if weekday is not None:
            # Weekly schedule
            # Convert launchd weekday (0=Sun, 1=Mon, ..., 6=Sat)
            # to Python weekday (0=Mon, 1=Tue, ..., 6=Sun)
            python_weekday = (weekday - 1) % 7
            days_ahead = python_weekday - now.weekday()
            if days_ahead < 0:
                days_ahead += 7
            elif days_ahead == 0 and next_run <= now:
                days_ahead = 7
            next_run += timedelta(days=days_ahead)
        else:
            # Daily schedule
            if next_run <= now:
                next_run += timedelta(days=1)

        # Generate human-readable description
        diff = next_run - now
        if diff.days == 0:
            description = f"Today, {next_run.strftime('%H:%M')}"
        elif diff.days == 1:
            description = f"Tomorrow, {next_run.strftime('%H:%M')}"
        else:
            day_names = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
            day_name = day_names[next_run.weekday()]
            description = f"{day_name}, {next_run.strftime('%H:%M')}"

        return next_run, description

    except Exception as e:
        logger.warning(f"Failed to calculate next run: {e}")
        return None, "Unknown"

get_last_run_info

get_last_run_info()

Get information about the last service run.

Determines last run time from log file modification times and parses log content for success/failure information.

Returns:

Type	Description
LastRunInfo	LastRunInfo with timestamp, success status, and statistics.

Example

manager = ServiceManager() last_run = manager.get_last_run_info() print(f"Last run: {last_run.relative_time} ({last_run.result_text})")

Source code in src/video_converter/automation/service_manager.py

def get_last_run_info(self) -> LastRunInfo:
    """Get information about the last service run.

    Determines last run time from log file modification times and
    parses log content for success/failure information.

    Returns:
        LastRunInfo with timestamp, success status, and statistics.

    Example:
        >>> manager = ServiceManager()
        >>> last_run = manager.get_last_run_info()
        >>> print(f"Last run: {last_run.relative_time} ({last_run.result_text})")
    """
    info = LastRunInfo()
    stdout_path, stderr_path = self.get_log_paths()

    # Get timestamp from log file
    try:
        if stdout_path.exists():
            mtime = stdout_path.stat().st_mtime
            info.timestamp = datetime.fromtimestamp(mtime)

            # Parse log for success/failure and statistics
            content = stdout_path.read_text()
            lines = content.splitlines()

            # Look for completion indicators
            for line in reversed(lines[-100:]):
                line_lower = line.lower()
                if "completed" in line_lower or "success" in line_lower:
                    info.success = True
                    break
                elif "failed" in line_lower or "error" in line_lower:
                    info.success = False
                    break

            # Look for statistics in log
            for line in reversed(lines[-50:]):
                if "converted" in line.lower():
                    # Try to extract number of videos
                    import re

                    match = re.search(r"(\d+)\s*videos?\s*converted", line, re.I)
                    if match:
                        info.videos_converted = int(match.group(1))
                if "saved" in line.lower():
                    # Try to extract bytes saved
                    import re

                    match = re.search(r"(\d+\.?\d*)\s*(GB|MB|KB|B)\s*saved", line, re.I)
                    if match:
                        value = float(match.group(1))
                        unit = match.group(2).upper()
                        multipliers = {"B": 1, "KB": 1024, "MB": 1024**2, "GB": 1024**3}
                        info.storage_saved_bytes = int(value * multipliers.get(unit, 1))

        # Check stderr for errors
        if stderr_path.exists():
            stderr_content = stderr_path.read_text().strip()
            if stderr_content:
                # Only mark as failed if there's actual error content
                last_lines = stderr_content.splitlines()[-5:]
                error_indicators = ["error", "failed", "exception", "traceback"]
                for line in last_lines:
                    if any(ind in line.lower() for ind in error_indicators):
                        info.success = False
                        info.error_message = line.strip()[:200]
                        break

    except Exception as e:
        logger.warning(f"Failed to get last run info: {e}")

    return info

get_detailed_status

get_detailed_status()

Get comprehensive status including history and scheduling.

Combines basic service status with conversion history statistics and next run calculations.

Returns:

Type	Description
DetailedServiceStatus	DetailedServiceStatus with all status information.

Example

manager = ServiceManager() status = manager.get_detailed_status() print(f"Status: {status.basic_status.state}") print(f"Next run: {status.next_run_relative}") print(f"Last run: {status.last_run.relative_time}") print(f"Total converted: {status.total_videos_converted}")

Source code in src/video_converter/automation/service_manager.py

def get_detailed_status(self) -> DetailedServiceStatus:
    """Get comprehensive status including history and scheduling.

    Combines basic service status with conversion history statistics
    and next run calculations.

    Returns:
        DetailedServiceStatus with all status information.

    Example:
        >>> manager = ServiceManager()
        >>> status = manager.get_detailed_status()
        >>> print(f"Status: {status.basic_status.state}")
        >>> print(f"Next run: {status.next_run_relative}")
        >>> print(f"Last run: {status.last_run.relative_time}")
        >>> print(f"Total converted: {status.total_videos_converted}")
    """
    # Get basic status
    basic_status = self.get_status()

    # Calculate next run
    next_run, next_run_relative = self.calculate_next_run()

    # Get last run info
    last_run = self.get_last_run_info()

    # Get conversion history statistics
    total_converted = 0
    total_saved = 0
    success_rate = 0.0

    try:
        from video_converter.core.history import get_history

        history = get_history()
        stats = history.get_statistics()
        total_converted = stats.total_converted
        total_saved = stats.total_saved_bytes
        success_rate = stats.success_rate
    except Exception as e:
        logger.debug(f"Could not load conversion history: {e}")

    return DetailedServiceStatus(
        basic_status=basic_status,
        next_run=next_run,
        next_run_relative=next_run_relative,
        last_run=last_run,
        total_videos_converted=total_converted,
        total_storage_saved_bytes=total_saved,
        success_rate=success_rate,
    )

format_bytes

format_bytes(bytes_value)

Format bytes into human-readable string.

Parameters:

Name	Type	Description	Default
bytes_value	int	Number of bytes.	required

Returns:

Type	Description
str	Formatted string like "1.5 GB" or "256 MB".

Source code in src/video_converter/automation/service_manager.py

def format_bytes(bytes_value: int) -> str:
    """Format bytes into human-readable string.

    Args:
        bytes_value: Number of bytes.

    Returns:
        Formatted string like "1.5 GB" or "256 MB".
    """
    if bytes_value < 1024:
        return f"{bytes_value} B"
    elif bytes_value < 1024**2:
        return f"{bytes_value / 1024:.1f} KB"
    elif bytes_value < 1024**3:
        return f"{bytes_value / 1024**2:.1f} MB"
    else:
        return f"{bytes_value / 1024**3:.1f} GB"

Launchd

launchd

launchd plist generator for macOS automation.

This module generates launchd plist files to enable scheduled automatic video conversion on macOS. It supports time-based scheduling (StartCalendarInterval) and folder-based triggers (WatchPaths).

SDS Reference: SDS-A01-001 SRS Reference: SRS-701 (launchd Integration)

Example

from video_converter.automation.launchd import LaunchdPlistGenerator from video_converter.core.config import Config

config = Config.load() generator = LaunchdPlistGenerator(config)

Generate plist for daily run at 3 AM

plist = generator.generate_plist(hour=3, minute=0) generator.write_plist(plist)

Generate plist with folder watching

plist = generator.generate_plist(watch_paths=[Path("~/Videos/Import")])

LaunchdSchedule dataclass

LaunchdSchedule(hour=None, minute=0, weekday=None)

Schedule configuration for launchd service.

Attributes:

Name	Type	Description
hour	int | None	Hour to run (0-23). None for watch-only mode.
minute	int	Minute to run (0-59).
weekday	int | None	Day of week (0=Sunday, 1=Monday, ..., 6=Saturday). None for daily schedule.

__post_init__

__post_init__()

Validate schedule values.

Source code in src/video_converter/automation/launchd.py

def __post_init__(self) -> None:
    """Validate schedule values."""
    if self.hour is not None and not (0 <= self.hour <= 23):
        raise ValueError(f"Hour must be 0-23, got {self.hour}")
    if not (0 <= self.minute <= 59):
        raise ValueError(f"Minute must be 0-59, got {self.minute}")
    if self.weekday is not None and not (0 <= self.weekday <= 6):
        raise ValueError(f"Weekday must be 0-6, got {self.weekday}")

to_calendar_interval

to_calendar_interval()

Convert to StartCalendarInterval dictionary.

Returns:

Type	Description
dict[str, int]	Dictionary with Hour, Minute, and optionally Weekday.

Source code in src/video_converter/automation/launchd.py

def to_calendar_interval(self) -> dict[str, int]:
    """Convert to StartCalendarInterval dictionary.

    Returns:
        Dictionary with Hour, Minute, and optionally Weekday.
    """
    result: dict[str, int] = {"Minute": self.minute}
    if self.hour is not None:
        result["Hour"] = self.hour
    if self.weekday is not None:
        result["Weekday"] = self.weekday
    return result

LaunchdConfig dataclass

LaunchdConfig(label=SERVICE_LABEL, program_args=list(), schedule=None, watch_paths=list(), working_directory=None, environment=dict(), stdout_path=None, stderr_path=None, run_at_load=False, keep_alive=False)

Configuration for launchd plist generation.

Attributes:

Name	Type	Description
label	str	Service identifier (e.g., "com.videoconverter.daily").
program_args	list[str]	Command line arguments for the service.
schedule	LaunchdSchedule | None	Time-based schedule (StartCalendarInterval).
watch_paths	list[Path]	Folder paths to watch for changes (WatchPaths).
working_directory	Path | None	Working directory for the service.
environment	dict[str, str]	Environment variables to set.
stdout_path	Path | None	Path for stdout log file.
stderr_path	Path | None	Path for stderr log file.
run_at_load	bool	Whether to run immediately when loaded.
keep_alive	bool	Whether to restart if process exits.

__post_init__

__post_init__()

Validate configuration.

Source code in src/video_converter/automation/launchd.py

def __post_init__(self) -> None:
    """Validate configuration."""
    if not self.schedule and not self.watch_paths:
        raise ValueError("Either schedule or watch_paths must be provided")

LaunchdPlistGenerator

LaunchdPlistGenerator(python_path=None, module_name='video_converter', command='run', log_dir=None)

Generator for launchd plist files.

This class creates valid launchd plist XML files for scheduling automatic video conversion on macOS.

Attributes:

Name	Type	Description
python_path		Path to Python interpreter.
module_name		Python module to execute.
command		CLI command to run (e.g., "run").
log_dir		Directory for log files.

Initialize the plist generator.

Parameters:

Name	Type	Description	Default
python_path	Path | None	Path to Python interpreter. Auto-detected if None.	None
module_name	str	Python module to execute.	'video_converter'
command	str	CLI command to run.	'run'
log_dir	Path | None	Directory for log files.	None

Source code in src/video_converter/automation/launchd.py

def __init__(
    self,
    python_path: Path | None = None,
    module_name: str = "video_converter",
    command: str = "run",
    log_dir: Path | None = None,
) -> None:
    """Initialize the plist generator.

    Args:
        python_path: Path to Python interpreter. Auto-detected if None.
        module_name: Python module to execute.
        command: CLI command to run.
        log_dir: Directory for log files.
    """
    self.python_path = python_path or self._detect_python_path()
    self.module_name = module_name
    self.command = command
    self.log_dir = log_dir or DEFAULT_LOG_DIR

generate_config

generate_config(hour=3, minute=0, weekday=None, watch_paths=None, run_at_load=False, extra_args=None)

Generate launchd configuration.

Parameters:

Name	Type	Description	Default
hour	int | None	Hour to run (0-23). None for watch-only mode.	3
minute	int	Minute to run (0-59).	0
weekday	int | None	Day of week (0=Sunday). None for daily.	None
watch_paths	list[Path] | None	Folders to watch for changes.	None
run_at_load	bool	Whether to run immediately when loaded.	False
extra_args	list[str] | None	Additional CLI arguments.	None

Returns:

Type	Description
LaunchdConfig	LaunchdConfig with all settings.

Source code in src/video_converter/automation/launchd.py

def generate_config(
    self,
    hour: int | None = 3,
    minute: int = 0,
    weekday: int | None = None,
    watch_paths: list[Path] | None = None,
    run_at_load: bool = False,
    extra_args: list[str] | None = None,
) -> LaunchdConfig:
    """Generate launchd configuration.

    Args:
        hour: Hour to run (0-23). None for watch-only mode.
        minute: Minute to run (0-59).
        weekday: Day of week (0=Sunday). None for daily.
        watch_paths: Folders to watch for changes.
        run_at_load: Whether to run immediately when loaded.
        extra_args: Additional CLI arguments.

    Returns:
        LaunchdConfig with all settings.
    """
    schedule = None
    if hour is not None:
        schedule = LaunchdSchedule(hour=hour, minute=minute, weekday=weekday)

    expanded_watch_paths = []
    if watch_paths:
        expanded_watch_paths = [p.expanduser().resolve() for p in watch_paths]

    # Ensure log directory exists
    self.log_dir.mkdir(parents=True, exist_ok=True)

    return LaunchdConfig(
        label=SERVICE_LABEL,
        program_args=self._build_program_args(extra_args),
        schedule=schedule,
        watch_paths=expanded_watch_paths,
        environment=self._build_environment(),
        stdout_path=self.log_dir / "stdout.log",
        stderr_path=self.log_dir / "stderr.log",
        run_at_load=run_at_load,
    )

generate_plist

generate_plist(config=None, hour=3, minute=0, weekday=None, watch_paths=None, run_at_load=False, extra_args=None)

Generate launchd plist dictionary.

Parameters:

Name	Type	Description	Default
config	LaunchdConfig | None	Pre-built configuration. If None, built from other args.	None
hour	int | None	Hour to run (0-23). None for watch-only mode.	3
minute	int	Minute to run (0-59).	0
weekday	int | None	Day of week (0=Sunday). None for daily.	None
watch_paths	list[Path] | None	Folders to watch for changes.	None
run_at_load	bool	Whether to run immediately when loaded.	False
extra_args	list[str] | None	Additional CLI arguments.	None

Returns:

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

Example

generator = LaunchdPlistGenerator() plist = generator.generate_plist(hour=3, minute=0)

plist is ready for writing with plistlib

Source code in src/video_converter/automation/launchd.py

def generate_plist(
    self,
    config: LaunchdConfig | None = None,
    hour: int | None = 3,
    minute: int = 0,
    weekday: int | None = None,
    watch_paths: list[Path] | None = None,
    run_at_load: bool = False,
    extra_args: list[str] | None = None,
) -> dict[str, Any]:
    """Generate launchd plist dictionary.

    Args:
        config: Pre-built configuration. If None, built from other args.
        hour: Hour to run (0-23). None for watch-only mode.
        minute: Minute to run (0-59).
        weekday: Day of week (0=Sunday). None for daily.
        watch_paths: Folders to watch for changes.
        run_at_load: Whether to run immediately when loaded.
        extra_args: Additional CLI arguments.

    Returns:
        Dictionary representation of plist.

    Example:
        >>> generator = LaunchdPlistGenerator()
        >>> plist = generator.generate_plist(hour=3, minute=0)
        >>> # plist is ready for writing with plistlib
    """
    if config is None:
        config = self.generate_config(
            hour=hour,
            minute=minute,
            weekday=weekday,
            watch_paths=watch_paths,
            run_at_load=run_at_load,
            extra_args=extra_args,
        )

    plist: dict[str, Any] = {
        "Label": config.label,
        "ProgramArguments": config.program_args,
    }

    # Add schedule (StartCalendarInterval)
    if config.schedule:
        plist["StartCalendarInterval"] = config.schedule.to_calendar_interval()

    # Add watch paths (WatchPaths)
    if config.watch_paths:
        plist["WatchPaths"] = [str(p) for p in config.watch_paths]

    # Add working directory
    if config.working_directory:
        plist["WorkingDirectory"] = str(config.working_directory)

    # Add environment variables
    if config.environment:
        plist["EnvironmentVariables"] = config.environment

    # Add logging
    if config.stdout_path:
        plist["StandardOutPath"] = str(config.stdout_path)
    if config.stderr_path:
        plist["StandardErrorPath"] = str(config.stderr_path)

    # Add run at load
    if config.run_at_load:
        plist["RunAtLoad"] = True

    # Add keep alive
    if config.keep_alive:
        plist["KeepAlive"] = True

    return plist

plist_to_xml

plist_to_xml(plist)

Convert plist dictionary to XML bytes.

Parameters:

Name	Type	Description	Default
plist	dict[str, Any]	Plist dictionary.	required

Returns:

Type	Description
bytes	XML representation as bytes.

Source code in src/video_converter/automation/launchd.py

def plist_to_xml(self, plist: dict[str, Any]) -> bytes:
    """Convert plist dictionary to XML bytes.

    Args:
        plist: Plist dictionary.

    Returns:
        XML representation as bytes.
    """
    return plistlib.dumps(plist, fmt=plistlib.FMT_XML)

write_plist

write_plist(plist, output_path=None)

Write plist to file.

Parameters:

Name	Type	Description	Default
plist	dict[str, Any]	Plist dictionary.	required
output_path	Path | None	Path to write plist. Defaults to LaunchAgents dir.	None

Returns:

Type	Description
Path	Path to the written plist file.

Raises:

Type	Description
IOError	If file cannot be written.

Source code in src/video_converter/automation/launchd.py

def write_plist(
    self,
    plist: dict[str, Any],
    output_path: Path | None = None,
) -> Path:
    """Write plist to file.

    Args:
        plist: Plist dictionary.
        output_path: Path to write plist. Defaults to LaunchAgents dir.

    Returns:
        Path to the written plist file.

    Raises:
        IOError: If file cannot be written.
    """
    if output_path is None:
        output_path = DEFAULT_LAUNCH_AGENTS_DIR / DEFAULT_PLIST_NAME

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

    with output_path.open("wb") as f:
        plistlib.dump(plist, f, fmt=plistlib.FMT_XML)

    logger.info(f"Wrote plist to {output_path}")
    return output_path

get_plist_path

get_plist_path()

Get the default plist file path.

Returns:

Type	Description
Path	Path to the plist file.

Source code in src/video_converter/automation/launchd.py

def get_plist_path(self) -> Path:
    """Get the default plist file path.

    Returns:
        Path to the plist file.
    """
    return DEFAULT_LAUNCH_AGENTS_DIR / DEFAULT_PLIST_NAME

validate_plist_syntax

validate_plist_syntax(plist_path)

Validate plist file syntax using plutil.

Parameters:

Name	Type	Description	Default
plist_path	Path	Path to plist file.	required

Returns:

Type	Description
bool	True if plist is valid, False otherwise.

Source code in src/video_converter/automation/launchd.py

def validate_plist_syntax(plist_path: Path) -> bool:
    """Validate plist file syntax using plutil.

    Args:
        plist_path: Path to plist file.

    Returns:
        True if plist is valid, False otherwise.
    """
    try:
        result = subprocess.run(
            ["plutil", "-lint", str(plist_path)],
            capture_output=True,
            text=True,
            check=False,
        )
        return result.returncode == 0
    except FileNotFoundError:
        logger.warning("plutil not found, skipping validation")
        return True

generate_daily_plist

generate_daily_plist(hour=3, minute=0, python_path=None)

Convenience function to generate a daily schedule plist.

Parameters:

Name	Type	Description	Default
hour	int	Hour to run (0-23).	3
minute	int	Minute to run (0-59).	0
python_path	Path | None	Path to Python interpreter.	None

Returns:

Type	Description
dict[str, Any]	Plist dictionary.

Example

plist = generate_daily_plist(hour=3, minute=0)

Write to file

generator = LaunchdPlistGenerator() generator.write_plist(plist)

Source code in src/video_converter/automation/launchd.py

def generate_daily_plist(
    hour: int = 3,
    minute: int = 0,
    python_path: Path | None = None,
) -> dict[str, Any]:
    """Convenience function to generate a daily schedule plist.

    Args:
        hour: Hour to run (0-23).
        minute: Minute to run (0-59).
        python_path: Path to Python interpreter.

    Returns:
        Plist dictionary.

    Example:
        >>> plist = generate_daily_plist(hour=3, minute=0)
        >>> # Write to file
        >>> generator = LaunchdPlistGenerator()
        >>> generator.write_plist(plist)
    """
    generator = LaunchdPlistGenerator(python_path=python_path)
    return generator.generate_plist(hour=hour, minute=minute)

generate_watch_plist

generate_watch_plist(watch_paths, python_path=None)

Convenience function to generate a folder-watching plist.

Parameters:

Name	Type	Description	Default
watch_paths	list[Path]	List of folders to watch.	required
python_path	Path | None	Path to Python interpreter.	None

Returns:

Type	Description
dict[str, Any]	Plist dictionary.

Example

plist = generate_watch_plist([Path("~/Videos/Import")]) generator = LaunchdPlistGenerator() generator.write_plist(plist)

Source code in src/video_converter/automation/launchd.py

def generate_watch_plist(
    watch_paths: list[Path],
    python_path: Path | None = None,
) -> dict[str, Any]:
    """Convenience function to generate a folder-watching plist.

    Args:
        watch_paths: List of folders to watch.
        python_path: Path to Python interpreter.

    Returns:
        Plist dictionary.

    Example:
        >>> plist = generate_watch_plist([Path("~/Videos/Import")])
        >>> generator = LaunchdPlistGenerator()
        >>> generator.write_plist(plist)
    """
    generator = LaunchdPlistGenerator(python_path=python_path)
    return generator.generate_plist(hour=None, watch_paths=watch_paths)

Notification

notification

macOS Notification Center integration for video conversion alerts.

This module provides integration with macOS Notification Center to send alerts for conversion completion, errors, and summary statistics.

SDS Reference: SDS-A01-002 SRS Reference: SRS-801 (macOS Notification Integration)

Example

from video_converter.automation.notification import NotificationManager from video_converter.core.types import ConversionReport

manager = NotificationManager()

Send a simple notification

manager.send_notification( ... title="Conversion Complete", ... body="12 videos converted, 8.5 GB saved", ... )

Send notification from a report

manager.send_batch_notification(report)

NotificationType

Bases: Enum

Type of notification to send.

Attributes:

Name	Type	Description
SUCCESS		All conversions completed successfully.
PARTIAL		Some conversions completed with errors.
FAILURE		All conversions failed or critical error occurred.
INFO		Informational notification.

NotificationConfig dataclass

NotificationConfig(enabled=True, sound=True, sound_name='Glass', group_id='com.videoconverter')

Configuration for notification behavior.

Attributes:

Name	Type	Description
enabled	bool	Whether notifications are enabled.
sound	bool	Whether to play sound with notification.
sound_name	str	Name of the sound to play (macOS system sound).
group_id	str	Optional group identifier for notification grouping.

NotificationResult dataclass

NotificationResult(success, error_message=None)

Result of a notification send attempt.

Attributes:

Name	Type	Description
success	bool	Whether the notification was sent successfully.
error_message	str | None	Error message if sending failed.

NotificationManager

NotificationManager(config=None)

Manager for sending macOS Notification Center notifications.

This class provides methods to send notifications for various video conversion events including batch completion, errors, and progress milestones.

Attributes:

Name	Type	Description
config		Notification configuration settings.

Initialize the notification manager.

Parameters:

Name	Type	Description	Default
config	NotificationConfig | None	Optional configuration. Uses defaults if not provided.	None

Source code in src/video_converter/automation/notification.py

def __init__(self, config: NotificationConfig | None = None) -> None:
    """Initialize the notification manager.

    Args:
        config: Optional configuration. Uses defaults if not provided.
    """
    self.config = config or NotificationConfig()
    self._is_macos = platform.system() == "Darwin"

is_available

is_available()

Check if notifications are available on this system.

Returns:

Type	Description
bool	True if notifications can be sent, False otherwise.

Source code in src/video_converter/automation/notification.py

def is_available(self) -> bool:
    """Check if notifications are available on this system.

    Returns:
        True if notifications can be sent, False otherwise.
    """
    if not self._is_macos:
        return False

    if not self.config.enabled:
        return False

    try:
        result = subprocess.run(
            ["which", "osascript"],
            capture_output=True,
            check=False,
        )
        return result.returncode == 0
    except (OSError, subprocess.SubprocessError):
        return False

send_notification

send_notification(title, body, sound=None, subtitle=None)

Send a notification to macOS Notification Center.

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
subtitle	str | None	Optional subtitle for the notification.	None

Returns:

Type	Description
NotificationResult	NotificationResult indicating success or failure.

Source code in src/video_converter/automation/notification.py

def send_notification(
    self,
    title: str,
    body: str,
    sound: bool | None = None,
    subtitle: str | None = None,
) -> NotificationResult:
    """Send a notification to macOS Notification Center.

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

    Returns:
        NotificationResult indicating success or failure.
    """
    if not self.config.enabled:
        return NotificationResult(success=True)

    if not self._is_macos:
        logger.debug("Notifications not supported on this platform")
        return NotificationResult(
            success=False,
            error_message="Notifications only supported on macOS",
        )

    use_sound = sound if sound is not None else self.config.sound

    try:
        script = self._build_applescript(title, body, use_sound, subtitle)
        result = subprocess.run(
            ["osascript", "-e", script],
            capture_output=True,
            text=True,
            check=False,
            timeout=10,
        )

        if result.returncode != 0:
            error_msg = result.stderr.strip() or "Unknown error"
            logger.warning(f"Failed to send notification: {error_msg}")
            return NotificationResult(success=False, error_message=error_msg)

        logger.debug(f"Notification sent: {title}")
        return NotificationResult(success=True)

    except subprocess.TimeoutExpired:
        logger.warning("Notification send timed out")
        return NotificationResult(
            success=False,
            error_message="Notification send timed out",
        )
    except (OSError, subprocess.SubprocessError) as e:
        logger.warning(f"Failed to send notification: {e}")
        return NotificationResult(success=False, error_message=str(e))

send_batch_notification

send_batch_notification(report)

Send notification based on batch conversion report.

Automatically determines notification type and formats the message based on conversion results.

Parameters:

Name	Type	Description	Default
report	ConversionReport	The conversion report to notify about.	required

Returns:

Type	Description
NotificationResult	NotificationResult indicating success or failure.

Source code in src/video_converter/automation/notification.py

def send_batch_notification(
    self,
    report: ConversionReport,
) -> NotificationResult:
    """Send notification based on batch conversion report.

    Automatically determines notification type and formats
    the message based on conversion results.

    Args:
        report: The conversion report to notify about.

    Returns:
        NotificationResult indicating success or failure.
    """
    notification_type = self._determine_notification_type(report)
    title = self._get_title_for_type(notification_type)
    body = self._format_report_body(report, notification_type)

    return self.send_notification(title=title, body=body)

send_success_notification

send_success_notification(videos_converted, size_saved)

Send a success notification with statistics.

Parameters:

Name	Type	Description	Default
videos_converted	int	Number of videos converted.	required
size_saved	int	Total bytes saved.	required

Returns:

Type	Description
NotificationResult	NotificationResult indicating success or failure.

Source code in src/video_converter/automation/notification.py

def send_success_notification(
    self,
    videos_converted: int,
    size_saved: int,
) -> NotificationResult:
    """Send a success notification with statistics.

    Args:
        videos_converted: Number of videos converted.
        size_saved: Total bytes saved.

    Returns:
        NotificationResult indicating success or failure.
    """
    title = "Conversion Complete"
    size_str = _format_size(size_saved)
    if videos_converted == 1:
        body = f"1 video converted, {size_str} saved"
    else:
        body = f"{videos_converted} videos converted, {size_str} saved"

    return self.send_notification(title=title, body=body)

send_error_notification

send_error_notification(error_message, file_name=None)

Send an error notification.

Parameters:

Name	Type	Description	Default
error_message	str	The error message to display.	required
file_name	str | None	Optional file name that caused the error.	None

Returns:

Type	Description
NotificationResult	NotificationResult indicating success or failure.

Source code in src/video_converter/automation/notification.py

def send_error_notification(
    self,
    error_message: str,
    file_name: str | None = None,
) -> NotificationResult:
    """Send an error notification.

    Args:
        error_message: The error message to display.
        file_name: Optional file name that caused the error.

    Returns:
        NotificationResult indicating success or failure.
    """
    title = "Conversion Failed"
    body = f"{file_name}: {error_message}" if file_name else error_message

    return self.send_notification(title=title, body=body)

send_partial_notification

send_partial_notification(succeeded, failed)

Send a partial success notification.

Parameters:

Name	Type	Description	Default
succeeded	int	Number of successful conversions.	required
failed	int	Number of failed conversions.	required

Returns:

Type	Description
NotificationResult	NotificationResult indicating success or failure.

Source code in src/video_converter/automation/notification.py

def send_partial_notification(
    self,
    succeeded: int,
    failed: int,
) -> NotificationResult:
    """Send a partial success notification.

    Args:
        succeeded: Number of successful conversions.
        failed: Number of failed conversions.

    Returns:
        NotificationResult indicating success or failure.
    """
    title = "Conversion Completed with Errors"
    total = succeeded + failed
    body = f"{succeeded}/{total} succeeded, {failed} failed"

    return self.send_notification(title=title, body=body)

send_notification

send_notification(title, body, sound=True)

Convenience function to send a simple notification.

Parameters:

Name	Type	Description	Default
title	str	The notification title.	required
body	str	The notification body text.	required
sound	bool	Whether to play sound.	True

Returns:

Type	Description
NotificationResult	NotificationResult indicating success or failure.

Example

result = send_notification( ... title="Conversion Complete", ... body="12 videos converted", ... )

Source code in src/video_converter/automation/notification.py

def send_notification(
    title: str,
    body: str,
    sound: bool = True,
) -> NotificationResult:
    """Convenience function to send a simple notification.

    Args:
        title: The notification title.
        body: The notification body text.
        sound: Whether to play sound.

    Returns:
        NotificationResult indicating success or failure.

    Example:
        >>> result = send_notification(
        ...     title="Conversion Complete",
        ...     body="12 videos converted",
        ... )
    """
    manager = NotificationManager()
    return manager.send_notification(title=title, body=body, sound=sound)