altametris.sara.core.weight_resolver¶

Résolveur intelligent de chemins pour les poids de modèles.

Ce module fournit un dispatcher/router qui détecte automatiquement le protocole d’un chemin de poids et délègue le téléchargement ou la validation selon le type: - azure://container/path.pt : Télécharge depuis Azure Blob Storage via AzureWeightManager - https://account.blob.core.windows.net/… : Parse URL Azure et télécharge via AzureWeightManager - https://apim.azure.com/… : Télécharge via APIM avec authentification OAuth2 (HTTPWeightDownloader) - local/path.pt : Valide l’existence du fichier local

Le WeightResolver agit comme une interface unifiée pour BaseDetector et BaseModel, simplifiant la gestion des différentes sources de poids.

URL Structure with Prefix:

APIM URLs follow the pattern: {api_base_url}/weights/{prefix}/{filename}

The prefix is configured via DS_WEIGHTS_PREFIX environment variable (default: “yolo-sara”). This allows organizing weights by project in Azure Blob Storage.

Blob Storage structure:

{container}/{prefix}/{filename} Example: weights-dev/yolo-sara/yolo11n.pt

APIM URL structure:

{api_base_url}/weights/{prefix}/{filename} Example: https://am-ds-apiman-dev.azure-api.net/weights/yolo-sara/yolo11n.pt

Use DsAuthConfig helpers to build URLs correctly:

>>> from altametris.sara.core.auth.config import DsAuthConfig
>>> config = DsAuthConfig.from_env()
>>> url = config.get_apim_weight_url("yolo11n.pt")  # Includes prefix
'https://am-ds-apiman-dev.azure-api.net/weights/yolo-sara/yolo11n.pt'

Example

>>> from altametris.sara.core import WeightResolver
>>> from altametris.sara.core.auth import DsCredentials
>>> # Résolution d'un chemin Azure Blob Storage (SDK)
>>> resolver = WeightResolver(enable_azure=True)
>>> local_path = resolver.resolve("azure://models/yolo11x.pt")
>>> print(f"Model at: {local_path}")
>>> # Résolution d'un chemin local
>>> local_path = resolver.resolve("/path/to/model.pt")
>>> # Résolution d'une URL HTTPS Azure Blob Storage
>>> local_path = resolver.resolve("https://myaccount.blob.core.windows.net/models/yolo11x.pt")
>>> # Résolution d'une URL APIM avec OAuth2 (auto-create credentials)
>>> resolver = WeightResolver(enable_azure=True)
>>> local_path = resolver.resolve("https://apim.azure.com/weights/yolo11x.pt")
>>> # Résolution d'une URL APIM avec credentials explicites
>>> creds = DsCredentials()
>>> resolver = WeightResolver(enable_azure=True, credentials=creds)
>>> local_path = resolver.resolve("https://apim.azure.com/weights/yolo11x.pt")

Attributes¶

`HAS_AZURE`
`HAS_HTTP_DOWNLOADER`
`HAS_DS_CREDENTIALS`
`logger`

Classes¶

WeightResolver

Résolveur intelligent de chemins de poids avec support multi-protocoles.

Module Contents¶

altametris.sara.core.weight_resolver.HAS_AZURE = True¶

altametris.sara.core.weight_resolver.HAS_HTTP_DOWNLOADER = True¶

altametris.sara.core.weight_resolver.HAS_DS_CREDENTIALS = True¶

altametris.sara.core.weight_resolver.logger¶

class altametris.sara.core.weight_resolver.WeightResolver(enable_azure: bool = True, cache_manager: altametris.sara.core.cache_manager.CacheManager | None = None, azure_weight_manager: Any | None = None, credentials: Any | None = None, **azure_kwargs: Any)¶

Résolveur intelligent de chemins de poids avec support multi-protocoles.

Parameters:

enable_azure – Active le support Azure (défaut: True). Si False, les chemins azure:// et les URLs Azure retourneront une erreur.
cache_manager – Instance de CacheManager (créée automatiquement si None).
azure_weight_manager – Instance de AzureWeightManager (créée automatiquement si None et enable_azure=True).
credentials – Instance de DsCredentials pour l’authentification APIM OAuth2 (créée automatiquement si None). Utilisé uniquement pour les URLs APIM (HTTPWeightDownloader).
**azure_kwargs – Arguments pour initialiser AzureWeightManager si non fourni (storage_account_name, auth_method, connection_string, etc.)

Raises:

ConfigurationError – Si la configuration est invalide.
ImportError – Si azure-storage-blob n’est pas installé et enable_azure=True.

Example

>>> # Avec Azure activé (défaut)
>>> resolver = WeightResolver(
...     enable_azure=True,
...     storage_account_name="myaccount",
...     auth_method="managed_identity"
... )
>>> path = resolver.resolve("azure://models/yolo11n.pt")
>>> # Sans Azure (chemins locaux seulement)
>>> resolver = WeightResolver(enable_azure=False)
>>> path = resolver.resolve("/local/path/model.pt")
>>> # Avec credentials APIM explicites
>>> from altametris.sara.core.auth import DsCredentials
>>> creds = DsCredentials()
>>> resolver = WeightResolver(enable_azure=True, credentials=creds)
>>> path = resolver.resolve("https://apim.azure.com/weights/yolo11x.pt")

enable_azure = True¶

cache = None¶

azure_manager: Any | None = None¶

http_downloader: Any | None = None¶

credentials = None¶

resolve(model_path: str, force_download: bool = False) → pathlib.Path¶

Résout un chemin de modèle vers un chemin local.

Détecte automatiquement le protocole et applique la stratégie appropriée: - Chemin local existant → Valide existence et retourne (priorité haute) - azure://container/weights → Télécharge via AzureWeightManager (SDK) - https://apim.*/... → Télécharge via HTTPWeightDownloader (OAuth2) - https://*.blob.core.windows.net/... → Parse et télécharge via AzureWeightManager (SDK) - Autres chemins locaux → Valide existence et retourne

Parameters:

model_path – Chemin du modèle (azure://, https://, ou chemin local)
force_download – Force le téléchargement même si en cache (pour azure:// et APIM)

Returns:

Path absolu du fichier de poids en local

Raises:

ConfigurationError – Si model_path est invalide ou Azure désactivé mais requis
ModelError – Si le fichier local n’existe pas ou si le téléchargement échoue
WeightsNotFoundError – Si le fichier de poids Azure n’existe pas
AzureDownloadError – Si le téléchargement Azure échoue
AzureAuthenticationError – Si l’authentification APIM échoue

Example

>>> resolver = WeightResolver(enable_azure=True)
>>> # Azure protocol (SDK)
>>> path = resolver.resolve("azure://models/yolo11n.pt")
>>> # Azure Blob Storage HTTPS URL (SDK)
>>> path = resolver.resolve("https://myaccount.blob.core.windows.net/models/yolo11n.pt")
>>> # APIM URL with OAuth2 (auto-create credentials)
>>> path = resolver.resolve("https://apim.azure.com/weights/yolo11x.pt")
>>> # APIM URL with explicit credentials
>>> from altametris.sara.core.auth import DsCredentials
>>> resolver = WeightResolver(enable_azure=True, credentials=DsCredentials())
>>> path = resolver.resolve("https://apim.azure.com/weights/yolo11x.pt")
>>> # Local path
>>> path = resolver.resolve("/local/path/model.pt")
>>> # Force re-download
>>> path = resolver.resolve("azure://models/yolo11n.pt", force_download=True)

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

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

Proxy vers CacheManager.clear_cache() pour faciliter l’utilisation.

Parameters:: pattern – Pattern optionnel (glob style, ex: "models/*.pt") Si None, supprime tout le cache.
Returns:: Nombre de fichiers supprimés

Example

>>> resolver = WeightResolver()
>>> # Supprimer tous les modèles yolo11
>>> removed = resolver.clear_cache(pattern="models/yolo*.pt")
>>> # Vider complètement le cache
>>> removed = resolver.clear_cache()

get_cache_stats() → dict[str, Any]¶

Retourne les statistiques du cache local.

Proxy vers CacheManager.get_cache_stats() pour faciliter l’utilisation.

Returns:: Dictionnaire avec les statistiques du cache

Example

>>> resolver = WeightResolver()
>>> stats = resolver.get_cache_stats()
>>> print(f"Cache: {stats['total_files']} files, {stats['total_size_mb']:.2f} MB")

list_weights(container: str = 'models', prefix: str | None = None) → list[dict[str, Any]]¶

Liste les poids disponibles dans un conteneur Azure.

Proxy vers AzureWeightManager.list_weights() pour faciliter l’utilisation.

Parameters:

container – Nom du conteneur (défaut: “models”)
prefix – Préfixe optionnel pour filtrer (ex: “yolo11”)

Returns:

Liste de dictionnaires contenant les métadonnées de chaque poid

Raises:

ConfigurationError – Si Azure est désactivé
AzureError – Si le listing échoue

Example

>>> resolver = WeightResolver(enable_azure=True)
>>> weights = resolver.list_weights("models", prefix="yolo11")
>>> for weight in weights:
...     print(f"{weight['name']}: {weight['size_mb']:.2f} MB")