


import importlib
import json
import os
from collections import OrderedDict
from typing import Dict, Optional, Union

# Build the list of all feature extractors
from ...configuration_utils import PretrainedConfig
from ...dynamic_module_utils import get_class_from_dynamic_module, resolve_trust_remote_code
from ...feature_extraction_utils import FeatureExtractionMixin
from ...utils import CONFIG_NAME, FEATURE_EXTRACTOR_NAME, get_file_from_repo, logging
from .auto_factory import _LazyAutoMapping
from .configuration_auto import (
    CONFIG_MAPPING_NAMES,
    AutoConfig,
    model_type_to_module_name,
    replace_list_option_in_docstrings,
)


lowerCAmelCase__      		= logging.get_logger(__name__)

lowerCAmelCase__      		= OrderedDict(
    [
        ('audio-spectrogram-transformer', 'ASTFeatureExtractor'),
        ('beit', 'BeitFeatureExtractor'),
        ('chinese_clip', 'ChineseCLIPFeatureExtractor'),
        ('clap', 'ClapFeatureExtractor'),
        ('clip', 'CLIPFeatureExtractor'),
        ('clipseg', 'ViTFeatureExtractor'),
        ('conditional_detr', 'ConditionalDetrFeatureExtractor'),
        ('convnext', 'ConvNextFeatureExtractor'),
        ('cvt', 'ConvNextFeatureExtractor'),
        ('data2vec-audio', 'Wav2Vec2FeatureExtractor'),
        ('data2vec-vision', 'BeitFeatureExtractor'),
        ('deformable_detr', 'DeformableDetrFeatureExtractor'),
        ('deit', 'DeiTFeatureExtractor'),
        ('detr', 'DetrFeatureExtractor'),
        ('dinat', 'ViTFeatureExtractor'),
        ('donut-swin', 'DonutFeatureExtractor'),
        ('dpt', 'DPTFeatureExtractor'),
        ('encodec', 'EncodecFeatureExtractor'),
        ('flava', 'FlavaFeatureExtractor'),
        ('glpn', 'GLPNFeatureExtractor'),
        ('groupvit', 'CLIPFeatureExtractor'),
        ('hubert', 'Wav2Vec2FeatureExtractor'),
        ('imagegpt', 'ImageGPTFeatureExtractor'),
        ('layoutlmv2', 'LayoutLMv2FeatureExtractor'),
        ('layoutlmv3', 'LayoutLMv3FeatureExtractor'),
        ('levit', 'LevitFeatureExtractor'),
        ('maskformer', 'MaskFormerFeatureExtractor'),
        ('mctct', 'MCTCTFeatureExtractor'),
        ('mobilenet_v1', 'MobileNetV1FeatureExtractor'),
        ('mobilenet_v2', 'MobileNetV2FeatureExtractor'),
        ('mobilevit', 'MobileViTFeatureExtractor'),
        ('nat', 'ViTFeatureExtractor'),
        ('owlvit', 'OwlViTFeatureExtractor'),
        ('perceiver', 'PerceiverFeatureExtractor'),
        ('poolformer', 'PoolFormerFeatureExtractor'),
        ('regnet', 'ConvNextFeatureExtractor'),
        ('resnet', 'ConvNextFeatureExtractor'),
        ('segformer', 'SegformerFeatureExtractor'),
        ('sew', 'Wav2Vec2FeatureExtractor'),
        ('sew-d', 'Wav2Vec2FeatureExtractor'),
        ('speech_to_text', 'Speech2TextFeatureExtractor'),
        ('speecht5', 'SpeechT5FeatureExtractor'),
        ('swiftformer', 'ViTFeatureExtractor'),
        ('swin', 'ViTFeatureExtractor'),
        ('swinv2', 'ViTFeatureExtractor'),
        ('table-transformer', 'DetrFeatureExtractor'),
        ('timesformer', 'VideoMAEFeatureExtractor'),
        ('tvlt', 'TvltFeatureExtractor'),
        ('unispeech', 'Wav2Vec2FeatureExtractor'),
        ('unispeech-sat', 'Wav2Vec2FeatureExtractor'),
        ('van', 'ConvNextFeatureExtractor'),
        ('videomae', 'VideoMAEFeatureExtractor'),
        ('vilt', 'ViltFeatureExtractor'),
        ('vit', 'ViTFeatureExtractor'),
        ('vit_mae', 'ViTFeatureExtractor'),
        ('vit_msn', 'ViTFeatureExtractor'),
        ('wav2vec2', 'Wav2Vec2FeatureExtractor'),
        ('wav2vec2-conformer', 'Wav2Vec2FeatureExtractor'),
        ('wavlm', 'Wav2Vec2FeatureExtractor'),
        ('whisper', 'WhisperFeatureExtractor'),
        ('xclip', 'CLIPFeatureExtractor'),
        ('yolos', 'YolosFeatureExtractor'),
    ]
)

lowerCAmelCase__      		= _LazyAutoMapping(CONFIG_MAPPING_NAMES, FEATURE_EXTRACTOR_MAPPING_NAMES)
def 							_UpperCAmelCase				(UpperCamelCase__     :  str						):
       for module_name, extractors in FEATURE_EXTRACTOR_MAPPING_NAMES.items():
              if class_name in extractors:
                     _A      :       Dict				    = model_type_to_module_name(UpperCamelCase__						)

                     _A      :       Tuple				    = importlib.import_module(f".{module_name}"				,							"transformers.models"						)
                     try:
                            return getattr(UpperCamelCase__				,							UpperCamelCase__						)
                     except AttributeError:
                            continue

       for _, extractor in FEATURE_EXTRACTOR_MAPPING._extra_content.items():
              if getattr(UpperCamelCase__				,							"__name__"				,							UpperCamelCase__						) == class_name:
                     return extractor

    # We did not fine the class, but maybe it's because a dep is missing. In that case, the class will be in the main
    # init and we return the proper dummy to get an appropriate error message.
       _A      :       Any				    = importlib.import_module("transformers"						)
       if hasattr(UpperCamelCase__				,							UpperCamelCase__						):
              return getattr(UpperCamelCase__				,							UpperCamelCase__						)

       return None
def 							_UpperCAmelCase				(UpperCamelCase__     :  Union[str, os.PathLike]				,							UpperCamelCase__     :  Optional[Union[str, os.PathLike]] = None				,							UpperCamelCase__     :  bool = False				,							UpperCamelCase__     :  bool = False				,							UpperCamelCase__     :  Optional[Dict[str, str]] = None				,							UpperCamelCase__     :  Optional[Union[bool, str]] = None				,							UpperCamelCase__     :  Optional[str] = None				,							UpperCamelCase__     :  bool = False				,							**UpperCamelCase__     :  Optional[int]				,							):
       _A      :       Tuple				    = get_file_from_repo(
           UpperCamelCase__				,							UpperCamelCase__				,							cache_dir=UpperCamelCase__				,							force_download=UpperCamelCase__				,							resume_download=UpperCamelCase__				,							proxies=UpperCamelCase__				,							use_auth_token=UpperCamelCase__				,							revision=UpperCamelCase__				,							local_files_only=UpperCamelCase__				,							)
       if resolved_config_file is None:
              logger.info(
                  "Could not locate the feature extractor configuration file, will try to use the model config instead."						)
              return {}

       with open(UpperCamelCase__				,							encoding="utf-8"						) as reader:
              return json.load(UpperCamelCase__						)







class      lowerCAmelCase__       :




     '''simple docstring'''


     def __init__(						self)			->				Any:
            raise EnvironmentError(
                "AutoFeatureExtractor is designed to be instantiated "
                "using the `AutoFeatureExtractor.from_pretrained(pretrained_model_name_or_path)` method.")


     @classmethod
     @replace_list_option_in_docstrings(__lowerCamelCase)
     def  _lowerCamelCase (						cls ,						__lowerCamelCase ,						**__lowerCamelCase)			->				Optional[int]:
            _A      :       Optional[int]				    = kwargs.pop("config" ,						__lowerCamelCase)
            _A      :       Tuple				    = kwargs.pop("trust_remote_code" ,						__lowerCamelCase)
            _A      :       List[Any]				    = True

            _A    ,			_A      :       Optional[int]				    = FeatureExtractionMixin.get_feature_extractor_dict(__lowerCamelCase ,						**__lowerCamelCase)
            _A      :       List[Any]				    = config_dict.get("feature_extractor_type" ,						__lowerCamelCase)
            _A      :       int				    = None
            if "AutoFeatureExtractor" in config_dict.get("auto_map" ,						{}):
                   _A      :       Any				    = config_dict["auto_map"]["AutoFeatureExtractor"]

            # If we don't find the feature extractor class in the feature extractor config, let's try the model config.
            if feature_extractor_class is None and feature_extractor_auto_map is None:
                   if not isinstance(__lowerCamelCase ,						__lowerCamelCase):
                          _A      :       str				    = AutoConfig.from_pretrained(__lowerCamelCase ,						**__lowerCamelCase)
                   # It could be in `config.feature_extractor_type``
                   _A      :       List[Any]				    = getattr(__lowerCamelCase ,						"feature_extractor_type" ,						__lowerCamelCase)
                   if hasattr(__lowerCamelCase ,						"auto_map") and "AutoFeatureExtractor" in config.auto_map:
                          _A      :       Optional[int]				    = config.auto_map["AutoFeatureExtractor"]

            if feature_extractor_class is not None:
                   _A      :       List[Any]				    = feature_extractor_class_from_name(__lowerCamelCase)

            _A      :       Any				    = feature_extractor_auto_map is not None
            _A      :       Optional[int]				    = feature_extractor_class is not None or type(__lowerCamelCase) in FEATURE_EXTRACTOR_MAPPING
            _A      :       int				    = resolve_trust_remote_code(
                __lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase)

            if has_remote_code and trust_remote_code:
                   _A      :       List[Any]				    = get_class_from_dynamic_module(
                       __lowerCamelCase ,						__lowerCamelCase ,						**__lowerCamelCase)
                   _A      :       List[str]				    = kwargs.pop("code_revision" ,						__lowerCamelCase)
                   if os.path.isdir(__lowerCamelCase):
                          feature_extractor_class.register_for_auto_class()
                   return feature_extractor_class.from_dict(__lowerCamelCase ,						**__lowerCamelCase)
            elif feature_extractor_class is not None:
                   return feature_extractor_class.from_dict(__lowerCamelCase ,						**__lowerCamelCase)
            # Last try: we use the FEATURE_EXTRACTOR_MAPPING.
            elif type(__lowerCamelCase) in FEATURE_EXTRACTOR_MAPPING:
                   _A      :       Dict				    = FEATURE_EXTRACTOR_MAPPING[type(__lowerCamelCase)]
                   return feature_extractor_class.from_dict(__lowerCamelCase ,						**__lowerCamelCase)

            raise ValueError(
                F"Unrecognized feature extractor in {pretrained_model_name_or_path}. Should have a "
                F"`feature_extractor_type` key in its {FEATURE_EXTRACTOR_NAME} of {CONFIG_NAME}, or one of the following "
                F"`model_type` keys in its {CONFIG_NAME}: {', '.join(c for c in FEATURE_EXTRACTOR_MAPPING_NAMES.keys())}")


     @staticmethod
     def  _lowerCamelCase (						__lowerCamelCase ,						__lowerCamelCase)			->				Union[str, Any]:
            FEATURE_EXTRACTOR_MAPPING.register(__lowerCamelCase ,						__lowerCamelCase)



import os
import tempfile
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
						import torch
						from torch import nn

						from transformers import (
						    Adafactor,
						    AdamW,
						    get_constant_schedule,
						    get_constant_schedule_with_warmup,
						    get_cosine_schedule_with_warmup,
						    get_cosine_with_hard_restarts_schedule_with_warmup,
						    get_inverse_sqrt_schedule,
						    get_linear_schedule_with_warmup,
						    get_polynomial_decay_schedule_with_warmup,
						)







def 			A_      (    _UpperCAmelCase				,	_UpperCAmelCase=10    ):
						SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       []
						for _ in range(_UpperCAmelCase    ):
												lrs.append(scheduler.get_lr()[0]    )
												scheduler.step()
						return lrs







def 			A_      (    _UpperCAmelCase				,	_UpperCAmelCase=10    ):
						SCREAMING_SNAKE_CASE_: List[str]  		=       []
						for step in range(_UpperCAmelCase    ):
												lrs.append(scheduler.get_lr()[0]    )
												scheduler.step()
												if step == num_steps // 2:
																		with tempfile.TemporaryDirectory() as tmpdirname:
																								SCREAMING_SNAKE_CASE_: Optional[int]  		=       os.path.join(_UpperCAmelCase				,	"schedule.bin"    )
																								torch.save(scheduler.state_dict()				,	_UpperCAmelCase    )

																								SCREAMING_SNAKE_CASE_: Optional[Any]  		=       torch.load(_UpperCAmelCase    )
																								scheduler.load_state_dict(_UpperCAmelCase    )
						return lrs



@require_torch
class   __lowercase				( unittest.TestCase	):




			"""simple docstring"""







			def        _SCREAMING_SNAKE_CASE			( self  :     Dict   ,						lowerCAmelCase__  :     int   ,						lowerCAmelCase__  :     Dict   ,						lowerCAmelCase__  :     Tuple):
									self.assertEqual(len(lowerCAmelCase__)   ,						len(lowerCAmelCase__))
									for a, b in zip(lowerCAmelCase__   ,						lowerCAmelCase__):
															self.assertAlmostEqual(lowerCAmelCase__   ,						lowerCAmelCase__   ,						delta=lowerCAmelCase__)



			def        _SCREAMING_SNAKE_CASE			( self  :     Dict):
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       torch.tensor([0.1, -0.2, -0.1]   ,						requires_grad=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: str  		=       torch.tensor([0.4, 0.2, -0.5])
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       nn.MSELoss()
									# No warmup, constant schedule, no gradient clipping
									SCREAMING_SNAKE_CASE_: int  		=       AdamW(params=[w]   ,						lr=2E-1   ,						weight_decay=0.0)
									for _ in range(100):
															SCREAMING_SNAKE_CASE_: Dict  		=       criterion(lowerCAmelCase__   ,						lowerCAmelCase__)
															loss.backward()
															optimizer.step()
															w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
															w.grad.zero_()
									self.assertListAlmostEqual(w.tolist()   ,						[0.4, 0.2, -0.5]   ,						tol=1E-2)




			def        _SCREAMING_SNAKE_CASE			( self  :     Dict):
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       torch.tensor([0.1, -0.2, -0.1]   ,						requires_grad=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: int  		=       torch.tensor([0.4, 0.2, -0.5])
									SCREAMING_SNAKE_CASE_: Any  		=       nn.MSELoss()
									# No warmup, constant schedule, no gradient clipping
									SCREAMING_SNAKE_CASE_: int  		=       Adafactor(
									    params=[w]   ,						lr=1E-2   ,						eps=(1E-30, 1E-3)   ,						clip_threshold=1.0   ,						decay_rate=-0.8   ,						betaa=lowerCAmelCase__   ,						weight_decay=0.0   ,						relative_step=lowerCAmelCase__   ,						scale_parameter=lowerCAmelCase__   ,						warmup_init=lowerCAmelCase__   ,						)
									for _ in range(1000):
															SCREAMING_SNAKE_CASE_: List[Any]  		=       criterion(lowerCAmelCase__   ,						lowerCAmelCase__)
															loss.backward()
															optimizer.step()
															w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
															w.grad.zero_()
									self.assertListAlmostEqual(w.tolist()   ,						[0.4, 0.2, -0.5]   ,						tol=1E-2)







@require_torch
class   __lowercase				( unittest.TestCase	):




			"""simple docstring"""







			_UpperCAmelCase   :							Union[str, Any]										=     nn.Linear(50		,       50	) if is_torch_available() else None
			_UpperCAmelCase   :							List[Any]										=     AdamW(m.parameters()		,       lr=10.0	) if is_torch_available() else None
			_UpperCAmelCase   :							Optional[Any]										=     10



			def        _SCREAMING_SNAKE_CASE			( self  :     Optional[int]   ,						lowerCAmelCase__  :     List[Any]   ,						lowerCAmelCase__  :     int   ,						lowerCAmelCase__  :     int   ,						lowerCAmelCase__  :     Optional[Any]=None):
									self.assertEqual(len(lowerCAmelCase__)   ,						len(lowerCAmelCase__))
									for a, b in zip(lowerCAmelCase__   ,						lowerCAmelCase__):
															self.assertAlmostEqual(lowerCAmelCase__   ,						lowerCAmelCase__   ,						delta=lowerCAmelCase__   ,						msg=lowerCAmelCase__)




			def        _SCREAMING_SNAKE_CASE			( self  :     int):
									SCREAMING_SNAKE_CASE_: Dict  		=       {"num_warmup_steps": 2, "num_training_steps": 10}
									# schedulers doct format
									# function: (sched_args_dict, expected_learning_rates)
									SCREAMING_SNAKE_CASE_: Dict  		=       {
									    get_constant_schedule: ({}, [10.0] * self.num_steps),
									    get_constant_schedule_with_warmup: (
									        {"num_warmup_steps": 4},
									        [0.0, 2.5, 5.0, 7.5, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0],
									    ),
									    get_linear_schedule_with_warmup: (
									        {**common_kwargs},
									        [0.0, 5.0, 10.0, 8.75, 7.5, 6.25, 5.0, 3.75, 2.5, 1.25],
									    ),
									    get_cosine_schedule_with_warmup: (
									        {**common_kwargs},
									        [0.0, 5.0, 10.0, 9.61, 8.53, 6.91, 5.0, 3.08, 1.46, 0.38],
									    ),
									    get_cosine_with_hard_restarts_schedule_with_warmup: (
									        {**common_kwargs, "num_cycles": 2},
									        [0.0, 5.0, 10.0, 8.53, 5.0, 1.46, 10.0, 8.53, 5.0, 1.46],
									    ),
									    get_polynomial_decay_schedule_with_warmup: (
									        {**common_kwargs, "power": 2.0, "lr_end": 1E-7},
									        [0.0, 5.0, 10.0, 7.656, 5.625, 3.906, 2.5, 1.406, 0.625, 0.156],
									    ),
									    get_inverse_sqrt_schedule: (
									        {"num_warmup_steps": 2},
									        [0.0, 5.0, 10.0, 8.165, 7.071, 6.325, 5.774, 5.345, 5.0, 4.714],
									    ),
									}

									for scheduler_func, data in scheds.items():
															SCREAMING_SNAKE_CASE_		,    SCREAMING_SNAKE_CASE_: Tuple  		=       data

															SCREAMING_SNAKE_CASE_: List[Any]  		=       scheduler_func(self.optimizer   ,						**lowerCAmelCase__)
															self.assertEqual(len([scheduler.get_lr()[0]])   ,						1)
															SCREAMING_SNAKE_CASE_: int  		=       unwrap_schedule(lowerCAmelCase__   ,						self.num_steps)
															self.assertListAlmostEqual(
															    lowerCAmelCase__   ,						lowerCAmelCase__   ,						tol=1E-2   ,						msg=F"failed for {scheduler_func} in normal scheduler"   ,						)

															SCREAMING_SNAKE_CASE_: List[str]  		=       scheduler_func(self.optimizer   ,						**lowerCAmelCase__)
															if scheduler_func.__name__ != "get_constant_schedule":
																					LambdaScheduleWrapper.wrap_scheduler(lowerCAmelCase__)  # wrap to test picklability of the schedule
															SCREAMING_SNAKE_CASE_: Tuple  		=       unwrap_and_save_reload_schedule(lowerCAmelCase__   ,						self.num_steps)
															self.assertListEqual(lowerCAmelCase__   ,						lowerCAmelCase__   ,						msg=F"failed for {scheduler_func} in save and reload")







class   __lowercase				:




			"""simple docstring"""







			def __init__( self  :     str   ,						lowerCAmelCase__  :     List[str]):
									SCREAMING_SNAKE_CASE_: List[Any]  		=       fn



			def __call__( self  :     Optional[int]   ,						*lowerCAmelCase__  :     List[Any]   ,						**lowerCAmelCase__  :     Tuple):
									return self.fn(*lowerCAmelCase__   ,						**lowerCAmelCase__)




			@classmethod
			def        _SCREAMING_SNAKE_CASE			( self  :     Optional[int]   ,						lowerCAmelCase__  :     str):
									SCREAMING_SNAKE_CASE_: str  		=       list(map(self   ,						scheduler.lr_lambdas))






