altametris.sara.core.http_weight_manager

HTTP downloader for model weights via APIM with OAuth2 JWT authentication.

Ce module fournit un gestionnaire pour télécharger automatiquement les poids de modèles depuis APIM (Azure API Management) avec authentification OAuth2 JWT, avec support de cache local et retry logic.

L’authentification utilise DsCredentials pour générer des tokens JWT qui sont ensuite utilisés dans les headers HTTP des requêtes vers l’APIM.

Example

>>> from altametris.sara.core.auth import DsCredentials
>>> from altametris.sara.core import HTTPWeightDownloader
>>> # Initialiser avec authentification
>>> creds = DsCredentials()
>>> downloader = HTTPWeightDownloader(credentials=creds)
>>> # Télécharger un modèle via APIM
>>> local_path = downloader.download_weight("weights/yolo11x.pt")
>>> print(f"Model downloaded to: {local_path}")
>>> # Lister les modèles disponibles
>>> weights = downloader.list_weights(prefix="yolo11")
>>> for weight in weights:
...     print(f"- {weight['name']} ({weight['size_mb']:.2f} MB)")

Attributes

requests

logger

Classes

HTTPWeightDownloader

Gestionnaire de téléchargement de poids via APIM avec OAuth2 JWT.

Module Contents

altametris.sara.core.http_weight_manager.requests = None
altametris.sara.core.http_weight_manager.logger
class altametris.sara.core.http_weight_manager.HTTPWeightDownloader(credentials: altametris.sara.core.auth.DsCredentials, cache_manager: altametris.sara.core.cache_manager.CacheManager | None = None, apim_base_url: str | None = None, show_progress: bool = True, max_retries: int = 3, retry_delay: float = 1.0, timeout: int = 300)

Gestionnaire de téléchargement de poids via APIM avec OAuth2 JWT.

Ce gestionnaire utilise des requêtes HTTP avec authentification JWT pour télécharger les poids de modèles depuis Azure API Management (APIM). Il intègre le système de cache local et fournit une retry logic robuste.

Parameters:

  • credentials – Instance de DsCredentials pour l’authentification JWT.

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

  • apim_base_url – URL de base de l’APIM (lu depuis env 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).

  • timeout – Timeout HTTP en secondes pour les requêtes.

Raises:

  • ConfigurationError – Si la configuration est invalide

  • ImportError – Si requests n’est pas installé

Example

>>> from altametris.sara.core.auth import DsCredentials
>>> from altametris.sara.core import HTTPWeightDownloader
>>> # Utiliser avec authentification JWT
>>> creds = DsCredentials()
>>> downloader = HTTPWeightDownloader(credentials=creds)
>>> # Télécharger un modèle
>>> path = downloader.download_weight("weights/yolo11x.pt")
>>> # Forcer le re-téléchargement
>>> path = downloader.download_weight("weights/yolo11x.pt", force_download=True)
credentials
show_progress = True
max_retries = 3
retry_delay = 1.0
timeout = 300
apim_base_url
cache = None
download_weight(weight_path: str, 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 APIM.

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 (ex: “weights/yolo11x.pt”)

  • 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

>>> downloader = HTTPWeightDownloader(credentials=DsCredentials())
>>> # Première fois: télécharge depuis APIM
>>> path = downloader.download_weight("weights/yolo11x.pt")
>>> print(f"Downloaded to: {path}")
>>> # Deuxième fois: retourne depuis le cache
>>> path = downloader.download_weight("weights/yolo11x.pt")
>>> print(f"Loaded from cache: {path}")
>>> # Forcer le téléchargement
>>> path = downloader.download_weight("weights/yolo11x.pt", force_download=True)
list_weights(prefix: str | None = None) → list[dict[str, Any]]

Liste les poids disponibles via l’API APIM.

Parameters:

prefix – Préfixe optionnel pour filtrer (ex: “yolo11”)

Returns:

  • name: Nom du fichier

  • size_bytes: Taille en octets

  • size_mb: Taille en MB

  • last_modified: Timestamp de dernière modification (si disponible)

Return type:

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

Raises:

Example

>>> downloader = HTTPWeightDownloader(credentials=DsCredentials())
>>> # Lister tous les modèles
>>> weights = downloader.list_weights()
>>> for weight in weights:
...     print(f"{weight['name']}: {weight['size_mb']:.2f} MB")
>>> # Lister seulement les modèles yolo11
>>> yolo_weights = downloader.list_weights(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: apim/*.pt). Si None, supprime tout le cache APIM.

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)