altametris.sara.core.weight_resolver ==================================== .. py:module:: altametris.sara.core.weight_resolver .. autoapi-nested-parse:: 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' .. rubric:: 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 ---------- .. autoapisummary:: altametris.sara.core.weight_resolver.HAS_AZURE altametris.sara.core.weight_resolver.HAS_HTTP_DOWNLOADER altametris.sara.core.weight_resolver.HAS_DS_CREDENTIALS altametris.sara.core.weight_resolver.logger Classes ------- .. autoapisummary:: altametris.sara.core.weight_resolver.WeightResolver Module Contents --------------- .. py:data:: HAS_AZURE :value: True .. py:data:: HAS_HTTP_DOWNLOADER :value: True .. py:data:: HAS_DS_CREDENTIALS :value: True .. py:data:: logger .. py:class:: WeightResolver(enable_azure: bool = True, cache_manager: Optional[altametris.sara.core.cache_manager.CacheManager] = None, azure_weight_manager: Optional[Any] = None, credentials: Optional[Any] = None, **azure_kwargs: Any) Résolveur intelligent de chemins de poids avec support multi-protocoles. :param enable_azure: Active le support Azure (défaut: True). Si False, les chemins azure:// et les URLs Azure retourneront une erreur. :param cache_manager: Instance de CacheManager (créée automatiquement si None). :param azure_weight_manager: Instance de AzureWeightManager (créée automatiquement si None et enable_azure=True). :param credentials: Instance de DsCredentials pour l'authentification APIM OAuth2 (créée automatiquement si None). Utilisé uniquement pour les URLs APIM (HTTPWeightDownloader). :param \*\*azure_kwargs: Arguments pour initialiser AzureWeightManager si non fourni (storage_account_name, auth_method, connection_string, etc.) :raises ConfigurationError: Si la configuration est invalide. :raises ImportError: Si azure-storage-blob n'est pas installé et enable_azure=True. .. rubric:: 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") .. py:attribute:: enable_azure :value: True .. py:attribute:: cache :value: None .. py:attribute:: azure_manager :type: Optional[Any] :value: None .. py:attribute:: http_downloader :type: Optional[Any] :value: None .. py:attribute:: credentials :value: None .. py:method:: 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 :param model_path: Chemin du modèle (azure://, https://, ou chemin local) :param 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 :raises ModelError: Si le fichier local n'existe pas ou si le téléchargement échoue :raises WeightsNotFoundError: Si le fichier de poids Azure n'existe pas :raises AzureDownloadError: Si le téléchargement Azure échoue :raises AzureAuthenticationError: Si l'authentification APIM échoue .. rubric:: 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) .. py:method:: clear_cache(pattern: Optional[str] = None) -> int Vide le cache local des poids téléchargés. Proxy vers CacheManager.clear_cache() pour faciliter l'utilisation. :param pattern: Pattern optionnel (glob style, ex: ``"models/*.pt"``) Si None, supprime tout le cache. :returns: Nombre de fichiers supprimés .. rubric:: 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() .. py:method:: 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 .. rubric:: Example >>> resolver = WeightResolver() >>> stats = resolver.get_cache_stats() >>> print(f"Cache: {stats['total_files']} files, {stats['total_size_mb']:.2f} MB") .. py:method:: list_weights(container: str = 'models', prefix: Optional[str] = None) -> list[dict[str, Any]] Liste les poids disponibles dans un conteneur Azure. Proxy vers AzureWeightManager.list_weights() pour faciliter l'utilisation. :param container: Nom du conteneur (défaut: "models") :param 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é :raises AzureError: Si le listing échoue .. rubric:: 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")