gnommo/gnommo/cache.py

"""GnommoCache - External storage extension for large media files.

Provides transparent fallback to external storage when files are not found locally.
Configure via ~/.gnommo.conf:

    [cache]
    path = /Volumes/GnommoDisk/gnommo

Files are looked up first locally, then in the cache at:
    {cache_path}/{project_name}/{relative_path}
"""

import configparser
import os
from pathlib import Path
from typing import Optional, Tuple

_cache_config: Optional[dict] = None
_perf_config: Optional[dict] = None


def get_ffmpeg_thread_count() -> int:
    """Return FFmpeg thread count based on [performance] cpu_limit in ~/.gnommo.conf.

    cpu_limit is a fraction of logical CPUs (e.g. 0.8 = 80%).
    Defaults to 1 when not configured, which is safe on memory-constrained machines.

    Example ~/.gnommo.conf:
        [performance]
        cpu_limit = 0.8
    """
    global _perf_config
    if _perf_config is None:
        config_path = Path.home() / ".gnommo.conf"
        _perf_config = {}
        if config_path.exists():
            cfg = configparser.ConfigParser()
            cfg.read(config_path)
            if cfg.has_option("performance", "cpu_limit"):
                try:
                    _perf_config["cpu_limit"] = float(
                        cfg.get("performance", "cpu_limit")
                    )
                except ValueError:
                    pass

    cpu_limit = _perf_config.get("cpu_limit")
    if cpu_limit is None:
        return 1
    cpu_count = os.cpu_count() or 1
    return max(1, int(cpu_count * cpu_limit))


def get_render_chunk_size() -> Optional[int]:
    """Return slides-per-chunk for auto-chunked rendering, or None if not configured.

    When set, cmd_render splits the filter graph into chunks of this many slides
    to avoid OOM from allocating filter buffers for the entire video at once.

    Example ~/.gnommo.conf:
        [performance]
        render_chunk_slides = 15
    """
    global _perf_config
    if _perf_config is None:
        get_ffmpeg_thread_count()  # populates _perf_config
    val = _perf_config.get("render_chunk_slides")
    if val is None:
        return None
    try:
        return max(1, int(val))
    except (ValueError, TypeError):
        return None


def load_cache_config() -> Optional[Path]:
    """Load gnommo.conf and return cache path if configured.

    Configuration file location: ~/.gnommo.conf

    Returns:
        Path to the cache root directory, or None if not configured.
    """
    global _cache_config
    if _cache_config is not None:
        return _cache_config.get("path")

    config_path = Path.home() / ".gnommo.conf"
    if not config_path.exists():
        _cache_config = {}
        return None

    config = configparser.ConfigParser()
    config.read(config_path)

    if config.has_option("cache", "path"):
        cache_path = Path(config.get("cache", "path"))
        _cache_config = {"path": cache_path}
        return cache_path

    _cache_config = {}
    return None


def resolve_with_cache(
    local_path: Path,
    project_path: Path,
) -> Tuple[Path, bool]:
    """
    Resolve a file path with cache fallback (read-only).

    Checks the local path first. If not found and cache is configured,
    checks the cache directory which mirrors the project structure.

    Args:
        local_path: The expected local path to the file
        project_path: The project root directory

    Returns:
        Tuple of (resolved_path, is_cached) where is_cached=True if
        the file was found in the external cache instead of locally.
    """
    # Check local path first
    if local_path.exists():
        return local_path, False

    # Check cache
    cache_base = load_cache_config()
    if cache_base is None:
        return local_path, False  # No cache configured

    # Try 1: path inside the project  →  cache_base / project_name / relative
    try:
        relative = local_path.relative_to(project_path)
        cache_path = cache_base / project_path.name / relative
        if cache_path.exists():
            return cache_path, True
    except ValueError:
        pass  # local_path is not under project_path

    # Try 2: path relative to gnommo root (sibling dirs like shared_assets)
    # e.g. shared_assets/pexels/file.mp4  →  cache_base / shared_assets / pexels / file.mp4
    try:
        relative = local_path.relative_to(project_path.parent)
        cache_path = cache_base / relative
        if cache_path.exists():
            return cache_path, True
    except ValueError:
        pass  # local_path is not under project_path.parent either

    return local_path, False


def load_server_config() -> Optional[dict]:
    """Load server rsync config from ~/.gnommo.conf.

    Expected config:
        [server]
        host = 76.13.144.52
        user = root
        path = /gnommo/project

    Returns:
        Dict with keys host, user, path (and optionally port), or None.
    """
    config_path = Path.home() / ".gnommo.conf"
    if not config_path.exists():
        return None

    config = configparser.ConfigParser()
    config.read(config_path)

    if not config.has_section("server"):
        return None

    host = config.get("server", "host", fallback=None)
    user = config.get("server", "user", fallback="root")
    path = config.get("server", "path", fallback="/gnommo/project")
    port = config.get("server", "port", fallback="22")

    if not host:
        return None

    return {"host": host, "user": user, "path": path, "port": port}


def is_cache_configured() -> bool:
    """Check if cache is configured (for status messages)."""
    return load_cache_config() is not None


def get_cache_info() -> Optional[str]:
    """Get a human-readable cache configuration string."""
    cache_path = load_cache_config()
    if cache_path is None:
        return None
    if cache_path.exists():
        return f"{cache_path} (connected)"
    return f"{cache_path} (not connected)"