Skip to content

Cache

Disk caching utilities used by compute_magnetic_field_variables to avoid re-running expensive IRBEM computations when downstream code crashes.

The default cache location is $HOME/.elpaso/joblib_cache. Stale entries older than 7 days are purged automatically at import el_paso time, and the entire cache is cleared on graceful interpreter exit via atexit.

el_paso.cache

Functions:

el_paso.cache.cleanup_stale_cache

cleanup_stale_cache

Delete cache entries older than max_age_days from the default cache directory.

This is a no-op if the cache directory does not exist.

Parameters:

Name Type Description Default
max_age_days int

Maximum age in days before an entry is removed.

_MAX_AGE_DAYS
Source code in el_paso/cache.py
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def cleanup_stale_cache(max_age_days: int = _MAX_AGE_DAYS) -> None:
    """Delete cache entries older than *max_age_days* from the default cache directory.

    This is a no-op if the cache directory does not exist.

    Args:
        max_age_days: Maximum age in days before an entry is removed.
    """
    home_path = os.getenv("HOME")
    if home_path is None:
        return

    cache_dir = Path(home_path) / ".elpaso" / _CACHE_SUBDIR
    if not cache_dir.exists():
        return

    cutoff = time.time() - max_age_days * 86400

    for entry in cache_dir.iterdir():
        if entry.is_dir() and entry.stat().st_mtime < cutoff:
            shutil.rmtree(entry)
            logger.debug("Removed stale cache entry: %s", entry)

el_paso.cache.clear_cache

clear_cache

Remove the entire joblib cache directory.

Source code in el_paso/cache.py
78
79
80
81
82
83
84
85
86
87
def clear_cache() -> None:
    """Remove the entire joblib cache directory."""
    home_path = os.getenv("HOME")
    if home_path is None:
        return

    cache_dir = Path(home_path) / ".elpaso" / _CACHE_SUBDIR
    if cache_dir.exists():
        shutil.rmtree(cache_dir)
        logger.debug("Cleared cache directory: %s", cache_dir)

el_paso.cache.clear_cache_on_success

clear_cache_on_success

Remove the cache directory only if the interpreter is exiting without an unhandled exception.

Intended as an atexit handler so the cache survives crashes but is cleaned up after a successful run.

Source code in el_paso/cache.py
90
91
92
93
94
95
96
97
98
99
def clear_cache_on_success() -> None:
    """Remove the cache directory only if the interpreter is exiting without an unhandled exception.

    Intended as an ``atexit`` handler so the cache survives crashes but is
    cleaned up after a successful run.
    """
    if _exit_with_exception:
        logger.debug("Keeping cache — interpreter is exiting due to an exception.")
        return
    clear_cache()

el_paso.cache.get_cache_dir

get_cache_dir

Return the default joblib cache directory at $HOME/.elpaso/joblib_cache.

The directory is created if it does not exist.

Returns:

Name Type Description
Path Path

The cache directory path.

Raises:

Type Description
OSError

If the HOME environment variable is not set.

Source code in el_paso/cache.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def get_cache_dir() -> Path:
    """Return the default joblib cache directory at ``$HOME/.elpaso/joblib_cache``.

    The directory is created if it does not exist.

    Returns:
        Path: The cache directory path.

    Raises:
        OSError: If the ``HOME`` environment variable is not set.
    """
    home_path = os.getenv("HOME")
    if home_path is None:
        msg = "HOME environment variable is not set!"
        raise OSError(msg)

    cache_dir = Path(home_path) / ".elpaso" / _CACHE_SUBDIR
    cache_dir.mkdir(parents=True, exist_ok=True)
    return cache_dir