altametris.sara.core.azure_weight_manager

Gestionnaire de téléchargement de poids depuis Azure Blob Storage.

Ce module fournit un gestionnaire pour télécharger automatiquement les poids de modèles depuis Azure Blob Storage, avec support de plusieurs méthodes d’authentification et intégration avec le système de cache local.

Méthodes d’authentification supportées: - Connection String (dev local) - Managed Identity (production - RECOMMANDÉ) - SAS Token (accès temporaire)

Example

>>> from altametris.sara.core import AzureWeightManager
>>> # Utiliser avec Managed Identity
>>> manager = AzureWeightManager(
...     storage_account_name="myaccount",
...     auth_method="managed_identity"
... )
>>> # Télécharger un modèle
>>> local_path = manager.get_weights("models/yolo11x.pt")
>>> print(f"Model downloaded to: {local_path}")
>>> # Lister les modèles disponibles
>>> weights = manager.list_weights("models", prefix="yolo11")
>>> for weight in weights:
...     print(f"- {weight['name']} ({weight['size_mb']:.2f} MB)")

Attributes

logger

AuthMethod

Classes

AzureWeightManager

Gestionnaire de téléchargement de poids depuis Azure Blob Storage avec support de cache local via CacheManager.

Module Contents

altametris.sara.core.azure_weight_manager.logger
altametris.sara.core.azure_weight_manager.AuthMethod
class altametris.sara.core.azure_weight_manager.AzureWeightManager(storage_account_name: str | None = None, auth_method: AuthMethod = 'connection_string', connection_string: str | None = None, sas_token: str | None = None, cache_manager: altametris.sara.core.cache_manager.CacheManager | None = None, show_progress: bool = True, max_retries: int = 3, retry_delay: float = 1.0)

Gestionnaire de téléchargement de poids depuis Azure Blob Storage avec support de cache local via CacheManager.

Parameters:

  • storage_account_name – Nom du compte de stockage Azure. Requis pour managed_identity et sas_token.

  • auth_method – Méthode d’authentification à utiliser: - “connection_string”: Utilise AZURE_STORAGE_CONNECTION_STRING - “managed_identity”: Utilise Azure Managed Identity - “sas_token”: Utilise AZURE_STORAGE_SAS_TOKEN

  • connection_string – Connection string Azure (optionnel, lu depuis env si None).

  • sas_token – SAS token Azure (optionnel, lu depuis env si None).

  • cache_manager – Instance de CacheManager (créée automatiquement si None).

  • show_progress – Afficher une barre de progression pour les téléchargements.

  • max_retries – Nombre maximum de tentatives en cas d’échec.

  • retry_delay – Délai initial en secondes entre les tentatives (backoff exponentiel).

Raises:

  • ConfigurationError – Si la configuration est invalide

  • ImportError – Si azure-storage-blob n’est pas installé

auth_method = 'connection_string'
storage_account_name = None
show_progress = True
max_retries = 3
retry_delay = 1.0
cache = None
get_weights(weight_path: str, container: str = 'models', force_download: bool = False) → pathlib.Path

Télécharge ou retourne depuis le cache les poids d’un modèle.

Cette méthode vérifie d’abord si les poids sont en cache et valides. Si oui, retourne le chemin local. Sinon, télécharge depuis Azure.

En cas d’échec du téléchargement, si un cache expiré existe, le retourne avec un warning.

Parameters:

  • weight_path – Chemin du fichier des poids dans le conteneur (ex: “yolo11x.pt” ou “models/yolo11x.pt”)

  • container – Nom du conteneur Azure (défaut: “models”)

  • force_download – Si True, force le téléchargement même si en cache

Returns:

Chemin absolu du fichier de poids en local

Raises:

Example

>>> manager = AzureWeightManager(auth_method="managed_identity", storage_account_name="myaccount")
>>> # Première fois: télécharge depuis Azure
>>> path = manager.get_weights("yolo11x.pt", container="models")
>>> print(f"Downloaded to: {path}")
>>> # Deuxième fois: retourne depuis le cache
>>> path = manager.get_weights("yolo11x.pt", container="models")
>>> print(f"Loaded from cache: {path}")
>>> # Forcer le téléchargement
>>> path = manager.get_weights("yolo11x.pt", container="models", force_download=True)
list_weights(container: str = 'models', prefix: str | None = None) → list[dict[str, Any]]

Liste les poids disponibles dans un conteneur Azure.

Parameters:

  • container – Nom du conteneur (défaut: “models”)

  • prefix – Préfixe optionnel pour filtrer (ex: “yolo11” pour lister tous les yolo11)

Returns:

  • name: Nom du blob

  • size_bytes: Taille en octets

  • size_mb: Taille en MB

  • last_modified: Timestamp de dernière modification

  • etag: ETag du blob

Return type:

Liste de dictionnaires contenant les métadonnées de chaque fichier de poids

Raises:

Example

>>> manager = AzureWeightManager(auth_method="managed_identity", storage_account_name="myaccount")
>>> # Lister tous les modèles
>>> weights = manager.list_weights("models")
>>> for weight in weights:
...     print(f"{weight['name']}: {weight['size_mb']:.2f} MB")
>>> # Lister seulement les modèles yolo11
>>> yolo_weights = manager.list_weights("models", prefix="yolo11")
clear_cache(pattern: str | None = None) → int

Vide le cache local des poids téléchargés.

Parameters:

pattern – Pattern optionnel (glob style, ex: "models/*.pt") Si None, supprime tout le cache.

Returns:

Nombre de fichiers supprimés

get_cache_stats() → dict[str, Any]

Retourne les statistiques du cache local.

Returns:

Dictionnaire avec les statistiques du cache (voir CacheManager.get_cache_stats)