altametris.sara.core.cache_manager¶

Gestionnaire de cache local pour les poids téléchargés depuis Azure.

Ce module fournit un système de cache intelligent pour stocker localement les poids de modèles téléchargés depuis Azure Blob Storage, avec support de TTL et nettoyage automatique.

Example

>>> from altametris.sara.core import CacheManager
>>> cache = CacheManager(cache_dir="/tmp/model_cache", ttl_seconds=86400)
>>> # Vérifier si un weight est en cache
>>> if cache.is_cached("models/yolo11x.pt"):
...     path = cache.get_cached_path("models/yolo11x.pt")
...     print(f"Model cached at: {path}")
>>> # Mettre à jour les métadonnées après téléchargement
>>> cache.update_cache_metadata(
...     "models/yolo11x.pt",
...     source="azure://models/yolo11x.pt",
...     size_bytes=104857600
... )

Attributes¶

logger

Classes¶

CacheManager

Gestionnaire de cache local pour les poids de modèles téléchargés.

Module Contents¶

altametris.sara.core.cache_manager.logger¶

class altametris.sara.core.cache_manager.CacheManager(cache_dir: str | None = None, ttl_seconds: int | None = None)¶

Gestionnaire de cache local pour les poids de modèles téléchargés.

Le cache utilise une structure de répertoires miroir du blob storage:

cache_dir/
    models/
        yolo11x.pt
        yolo11x.pt.meta (métadonnées JSON)

Parameters:

cache_dir – Répertoire racine du cache (défaut: ~/.cache/altametris/weights). Stocké comme attribut après initialisation.
ttl_seconds – Durée de vie des fichiers en secondes (défaut: 7 jours). Après ce délai, les fichiers sont considérés comme expirés. Valeur None = pas d’expiration. Stocké comme attribut après initialisation.

Note

Les attributs principaux sont cache_dir (Path), ttl_seconds (int | None), et _meta_suffix (str, toujours “.meta”).

Example

>>> cache = CacheManager(cache_dir="/tmp/cache", ttl_seconds=3600)
>>> cache.is_cached("models/yolo11x.pt")
False
>>> cache.update_cache_metadata("models/yolo11x.pt", source="azure://models/yolo11x.pt", size_bytes=1024)
>>> cache.is_cached("models/yolo11x.pt")
True
>>> stats = cache.get_cache_stats()
>>> print(f"Cache contains {stats['total_files']} files, {stats['total_size_mb']:.2f} MB")

cache_dir¶

ttl_seconds = None¶

is_cached(cache_key: str) → bool¶

Vérifie si un élément est en cache et non expiré.

Un élément est considéré comme valide si: 1. Le fichier existe dans le cache 2. Le fichier de métadonnées existe et est valide (JSON parsable) 3. Le TTL n’est pas dépassé (si ttl_seconds est défini)

Parameters:: cache_key – Clé de cache à vérifier (ex: “models/yolo11x.pt”)
Returns:: True si l’élément est en cache et valide, False sinon

Example

>>> cache = CacheManager(ttl_seconds=3600)
>>> cache.is_cached("models/yolo11x.pt")
False
>>> # Après téléchargement et update_cache_metadata...
>>> cache.is_cached("models/yolo11x.pt")
True

get_cached_path(cache_key: str) → pathlib.Path¶

Retourne le chemin local d’un élément en cache.

Parameters:: cache_key – Clé de cache (ex: “models/yolo11x.pt”)
Returns:: Chemin absolu du fichier dans le cache

Example

>>> cache = CacheManager()
>>> if cache.is_cached("models/yolo11x.pt"):
...     path = cache.get_cached_path("models/yolo11x.pt")
...     print(f"Load model from: {path}")

update_cache_metadata(cache_key: str, source: str | None = None, size_bytes: int | None = None, **extra_metadata: Any) → None¶

Enregistre ou met à jour les métadonnées d’un fichier en cache.

Cette méthode doit être appelée après avoir téléchargé un fichier dans le cache pour enregistrer ses métadonnées (timestamp, source, taille)!!!.

Les métadonnées sont stockées dans un fichier JSON avec le suffixe .meta

Parameters:

cache_key – Clé de cache (ex: “models/yolo11x.pt”)
source – URL ou chemin source (ex: “azure://models/yolo11x.pt”)
size_bytes – Taille du fichier en octets
**extra_metadata – Métadonnées additionnelles à stocker

Raises:

ConfigurationError – Si le fichier cache n’existe pas ou si l’écriture échoue

Example

>>> cache = CacheManager()
>>> # Après avoir téléchargé le fichier dans cache.get_cached_path(...)
>>> cache.update_cache_metadata(
...     "models/yolo11x.pt",
...     source="azure://models/yolo11x.pt",
...     size_bytes=104857600,
...     version="v1.0"
... )

cleanup_expired() → int¶

Supprime tous les fichiers expirés du cache.

Parcourt récursivement le cache et supprime: 1. Les fichiers dont le TTL est dépassé 2. Les fichiers de métadonnées associés 3. Les répertoires vides résultants

Returns:: Nombre de fichiers supprimés (paires fichier + métadonnées comptent comme 1)

Example

>>> cache = CacheManager(ttl_seconds=3600)
>>> # Après 2 heures...
>>> removed = cache.cleanup_expired()
>>> print(f"Removed {removed} expired files")

clear_cache(pattern: str | None = None) → int¶

Vide le cache complètement ou selon un pattern.

Parameters:: pattern – Pattern optionnel (glob style, ex: "models/*.pt", "yolo11*"). Si None, supprime tout le cache.
Returns:: Nombre de fichiers supprimés (paires fichier + métadonnées comptent comme 1)

Example

>>> cache = CacheManager()
>>> # Supprimer tous les modèles yolo11
>>> removed = cache.clear_cache(pattern="models/yolo11*.pt")
>>> print(f"Removed {removed} files")
>>> # Vider complètement le cache
>>> removed = cache.clear_cache()
>>> print(f"Cache cleared: {removed} files removed")

get_cache_stats() → dict[str, Any]¶

Retourne des statistiques détaillées sur le cache.

Returns:

total_files: Nombre total de fichiers en cache
total_size_bytes: Taille totale en octets
total_size_mb: Taille totale en MB
expired_files: Nombre de fichiers expirés (si TTL configuré)
valid_files: Nombre de fichiers valides (si TTL configuré)
cache_dir: Chemin du répertoire de cache
ttl_seconds: TTL configuré (ou None)

Return type:

Dictionnaire contenant

Example

>>> cache = CacheManager()
>>> stats = cache.get_cache_stats()
>>> print(f"Cache: {stats['total_files']} files, {stats['total_size_mb']:.2f} MB")
>>> if stats['expired_files'] > 0:
...     print(f"Warning: {stats['expired_files']} expired files")