


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,
)


__snake_case   							=						logging.get_logger(__name__)

__snake_case   							=						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'''),
    ]
)

__snake_case   							=						_LazyAutoMapping(CONFIG_MAPPING_NAMES, FEATURE_EXTRACTOR_MAPPING_NAMES)






def 		_A		(		_lowercase )					-> str:


     """simple docstring"""


     for module_name, extractors in FEATURE_EXTRACTOR_MAPPING_NAMES.items():
          if class_name in extractors:
               __UpperCamelCase =			model_type_to_module_name(_lowercase )

               __UpperCamelCase =			importlib.import_module(f'''.{module_name}'''					,  'transformers.models' )
               try:
                    return getattr(_lowercase					,  _lowercase )
               except AttributeError:
                    continue

     for _, extractor in FEATURE_EXTRACTOR_MAPPING._extra_content.items():
          if getattr(_lowercase					,  '__name__'					,  _lowercase ) == 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.
     __UpperCamelCase =			importlib.import_module('transformers' )
     if hasattr(_lowercase					,  _lowercase ):
          return getattr(_lowercase					,  _lowercase )

     return None






def 		_A		(		_lowercase					,  _lowercase = None					,  _lowercase = False					,  _lowercase = False					,  _lowercase = None					,  _lowercase = None					,  _lowercase = None					,  _lowercase = False					,  **_lowercase					,  )					-> Dict:


     """simple docstring"""


     __UpperCamelCase =			get_file_from_repo(
         _lowercase					,  _lowercase					,  cache_dir=_lowercase					,  force_download=_lowercase					,  resume_download=_lowercase					,  proxies=_lowercase					,  use_auth_token=_lowercase					,  revision=_lowercase					,  local_files_only=_lowercase					,  )
     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(_lowercase					,  encoding='utf-8' ) as reader:
          return json.load(_lowercase )



class 	__lowerCamelCase			:







       def __init__(			self: Union[str, Any]					):






            '''simple docstring'''




            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(A_					)
       def 		snake_case_	(			cls: List[Any],A_: Dict,**A_: int					):






            '''simple docstring'''




            __UpperCamelCase =			kwargs.pop('config',A_					)
            __UpperCamelCase =			kwargs.pop('trust_remote_code',A_					)
            __UpperCamelCase =			True

            __UpperCamelCase,    __UpperCamelCase =			FeatureExtractionMixin.get_feature_extractor_dict(A_,**A_					)
            __UpperCamelCase =			config_dict.get('feature_extractor_type',A_					)
            __UpperCamelCase =			None
            if "AutoFeatureExtractor" in config_dict.get('auto_map',{}					):
                 __UpperCamelCase =			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(A_,A_					):
                      __UpperCamelCase =			AutoConfig.from_pretrained(A_,**A_					)
                 # It could be in `config.feature_extractor_type``
                 __UpperCamelCase =			getattr(A_,'feature_extractor_type',A_					)
                 if hasattr(A_,'auto_map'					) and "AutoFeatureExtractor" in config.auto_map:
                      __UpperCamelCase =			config.auto_map['AutoFeatureExtractor']

            if feature_extractor_class is not None:
                 __UpperCamelCase =			feature_extractor_class_from_name(A_					)

            __UpperCamelCase =			feature_extractor_auto_map is not None
            __UpperCamelCase =			feature_extractor_class is not None or type(A_					) in FEATURE_EXTRACTOR_MAPPING
            __UpperCamelCase =			resolve_trust_remote_code(
                A_,A_,A_,A_					)

            if has_remote_code and trust_remote_code:
                 __UpperCamelCase =			get_class_from_dynamic_module(
                     A_,A_,**A_					)
                 __UpperCamelCase =			kwargs.pop('code_revision',A_					)
                 if os.path.isdir(A_					):
                      feature_extractor_class.register_for_auto_class()
                 return feature_extractor_class.from_dict(A_,**A_					)
            elif feature_extractor_class is not None:
                 return feature_extractor_class.from_dict(A_,**A_					)
            # Last try: we use the FEATURE_EXTRACTOR_MAPPING.
            elif type(A_					) in FEATURE_EXTRACTOR_MAPPING:
                 __UpperCamelCase =			FEATURE_EXTRACTOR_MAPPING[type(A_					)]
                 return feature_extractor_class.from_dict(A_,**A_					)

            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 		snake_case_	(			A_: Any,A_: Tuple					):






            '''simple docstring'''




            FEATURE_EXTRACTOR_MAPPING.register(A_,A_					)






import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()



@skip_mps
class 	__lowerCamelCase			(_a				,      unittest.TestCase						):
       _lowercase											=     VideoToVideoSDPipeline
       _lowercase											=     TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"""video"""}						) - {"""image""", """width""", """height"""}
       _lowercase											=     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""video"""}						) - {"""image"""}
       _lowercase											=     PipelineTesterMixin.required_optional_params - {"""latents"""}
       _lowercase											=     False

       # No `output_type`.
       _lowercase											=     frozenset(
           [
               """num_inference_steps""",
               """generator""",
               """latents""",
               """return_dict""",
               """callback""",
               """callback_steps""",
           ]						)







       def 		snake_case_	(			self: List[str]					):






            '''simple docstring'''




            torch.manual_seed(0					)
            __UpperCamelCase =			UNetaDConditionModel(
                block_out_channels=(32, 64, 64, 64),layers_per_block=2,sample_size=32,in_channels=4,out_channels=4,down_block_types=('CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'DownBlock3D'),up_block_types=('UpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D'),cross_attention_dim=32,attention_head_dim=4,)
            __UpperCamelCase =			DDIMScheduler(
                beta_start=0.0_0_0_8_5,beta_end=0.0_1_2,beta_schedule='scaled_linear',clip_sample=A_,set_alpha_to_one=A_,)
            torch.manual_seed(0					)
            __UpperCamelCase =			AutoencoderKL(
                block_out_channels=[32, 64],in_channels=3,out_channels=3,down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D'],up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D'],latent_channels=4,sample_size=128,)
            torch.manual_seed(0					)
            __UpperCamelCase =			CLIPTextConfig(
                bos_token_id=0,eos_token_id=2,hidden_size=32,intermediate_size=37,layer_norm_eps=1E-05,num_attention_heads=4,num_hidden_layers=5,pad_token_id=1,vocab_size=1000,hidden_act='gelu',projection_dim=512,)
            __UpperCamelCase =			CLIPTextModel(A_					)
            __UpperCamelCase =			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					)

            __UpperCamelCase =			{
                'unet': unet,
                'scheduler': scheduler,
                'vae': vae,
                'text_encoder': text_encoder,
                'tokenizer': tokenizer,
            }
            return components







       def 		snake_case_	(			self: Union[str, Any],A_: Any,A_: Any=0					):






            '''simple docstring'''




            __UpperCamelCase =			floats_tensor((1, 3, 3, 32, 32),rng=random.Random(A_					)					).to(A_					)

            if str(A_					).startswith('mps'					):
                 __UpperCamelCase =			torch.manual_seed(A_					)
            else:
                 __UpperCamelCase =			torch.Generator(device=A_					).manual_seed(A_					)
            __UpperCamelCase =			{
                'prompt': 'A painting of a squirrel eating a burger',
                'video': video,
                'generator': generator,
                'num_inference_steps': 2,
                'guidance_scale': 6.0,
                'output_type': 'pt',
            }
            return inputs







       def 		snake_case_	(			self: Union[str, Any]					):






            '''simple docstring'''




            __UpperCamelCase =			'cpu'  # ensure determinism for the device-dependent torch.Generator
            __UpperCamelCase =			self.get_dummy_components()
            __UpperCamelCase =			VideoToVideoSDPipeline(**A_					)
            __UpperCamelCase =			sd_pipe.to(A_					)
            sd_pipe.set_progress_bar_config(disable=A_					)

            __UpperCamelCase =			self.get_dummy_inputs(A_					)
            __UpperCamelCase =			'np'
            __UpperCamelCase =			sd_pipe(**A_					).frames
            __UpperCamelCase =			frames[0][-3:, -3:, -1]

            assert frames[0].shape == (32, 32, 3)
            __UpperCamelCase =			np.array([106, 117, 113, 174, 137, 112, 148, 151, 131]					)

            assert np.abs(image_slice.flatten() - expected_slice					).max() < 1E-2







       @unittest.skipIf(
           torch_device != 'cuda' or not is_xformers_available(),reason='XFormers attention is only available with CUDA and `xformers` installed',)
       def 		snake_case_	(			self: Any					):






            '''simple docstring'''




            self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=A_,expected_max_diff=5E-3					)







       @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'					)
       def 		snake_case_	(			self: str					):






            '''simple docstring'''




            pass







       @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'					)
       def 		snake_case_	(			self: Optional[Any]					):






            '''simple docstring'''




            pass







       @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'					)
       def 		snake_case_	(			self: int					):






            '''simple docstring'''




            pass







       def 		snake_case_	(			self: Any					):






            '''simple docstring'''




            return super().test_progress_bar()



@slow
@skip_mps
class 	__lowerCamelCase			(unittest.TestCase						):







       def 		snake_case_	(			self: Tuple					):






            '''simple docstring'''




            __UpperCamelCase =			VideoToVideoSDPipeline.from_pretrained('cerspense/zeroscope_v2_XL',torch_dtype=torch.floataa					)
            pipe.enable_model_cpu_offload()

            # 10 frames
            __UpperCamelCase =			torch.Generator(device='cpu'					).manual_seed(0					)
            __UpperCamelCase =			torch.randn((1, 10, 3, 1024, 576),generator=A_					)
            __UpperCamelCase =			video.to('cuda'					)

            __UpperCamelCase =			'Spiderman is surfing'

            __UpperCamelCase =			pipe(A_,video=A_,generator=A_,num_inference_steps=3,output_type='pt'					).frames

            __UpperCamelCase =			np.array([-1.0_4_5_8_9_8_4, -1.1_2_7_9_2_9_7, -0.9_6_6_3_0_8_6, -0.9_1_5_0_3_9_0_6, -0.7_5_0_9_7_6_5_6]					)
            assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array					).sum() < 1E-2


