


import collections
import importlib.util
import os
import re
from pathlib import Path


UpperCAmelCase_      : int  					=       'src/transformers'


# Matches is_xxx_available()
UpperCAmelCase_      : Any  					=       re.compile(R'is\_([a-z_]*)_available()')
# Catches a one-line _import_struct = {xxx}
UpperCAmelCase_      : Any  					=       re.compile(R'^_import_structure\s+=\s+\{([^\}]+)\}')
# Catches a line with a key-values pattern: "bla": ["foo", "bar"]
UpperCAmelCase_      : Dict  					=       re.compile(R'\s+"\S*":\s+\[([^\]]*)\]')
# Catches a line if not is_foo_available
UpperCAmelCase_      : int  					=       re.compile(R'^\s*if\s+not\s+is\_[a-z_]*\_available\(\)')
# Catches a line _import_struct["bla"].append("foo")
UpperCAmelCase_      : Optional[Any]  					=       re.compile(R'^\s*_import_structure\["\S*"\]\.append\("(\S*)"\)')
# Catches a line _import_struct["bla"].extend(["foo", "bar"]) or _import_struct["bla"] = ["foo", "bar"]
UpperCAmelCase_      : List[Any]  					=       re.compile(R'^\s*_import_structure\[\S*\](?:\.extend\(|\s*=\s+)\[([^\]]*)\]')
# Catches a line with an object between quotes and a comma:     "MyModel",
UpperCAmelCase_      : int  					=       re.compile('^\s+"([^"]+)",')
# Catches a line with objects between brackets only:    ["foo", "bar"],
UpperCAmelCase_      : Optional[Any]  					=       re.compile('^\s+\[([^\]]+)\]')
# Catches a line with from foo import bar, bla, boo
UpperCAmelCase_      : Any  					=       re.compile(R'\s+from\s+\S*\s+import\s+([^\(\s].*)\n')
# Catches a line with try:
UpperCAmelCase_      : Tuple  					=       re.compile(R'^\s*try:')
# Catches a line with else:
UpperCAmelCase_      : Any  					=       re.compile(R'^\s*else:')







def        SCREAMING_SNAKE_CASE_						(			__A :			str				)	->		List[str]:

       """simple docstring"""

       if _re_test_backend.search(__A				) is None:
              return None
       a_					:		Dict							=  [b[0] for b in _re_backend.findall(__A				)]
       backends.sort()
       return "_and_".join(__A				)







def        SCREAMING_SNAKE_CASE_						(			__A :			Optional[Any]				)	->		str:

       """simple docstring"""

       with open(__A  ,    'r'  ,    encoding='utf-8'  ,    newline='\n'				) as f:
              a_					:		Optional[Any]							=  f.readlines()

       a_					:		Optional[int]							=  0
       while line_index < len(__A				) and not lines[line_index].startswith('_import_structure = {'				):
              line_index += 1

       # If this is a traditional init, just return.
       if line_index >= len(__A				):
              return None

       # First grab the objects without a specific backend in _import_structure
       a_					:		Optional[Any]							=  []
       while not lines[line_index].startswith('if TYPE_CHECKING'				) and find_backend(lines[line_index]				) is None:
              a_					:		Optional[int]							=  lines[line_index]
              # If we have everything on a single line, let's deal with it.
              if _re_one_line_import_struct.search(__A				):
                     a_					:		Union[str, Any]							=  _re_one_line_import_struct.search(__A				).groups()[0]
                     a_					:		int							=  re.findall('\[([^\]]+)\]'  ,    __A				)
                     for imp in imports:
                            objects.extend([obj[1:-1] for obj in imp.split(', '				)]				)
                     line_index += 1
                     continue
              a_					:		Any							=  _re_import_struct_key_value.search(__A				)
              if single_line_import_search is not None:
                     a_					:		Union[str, Any]							=  [obj[1:-1] for obj in single_line_import_search.groups()[0].split(', '				) if len(__A				) > 0]
                     objects.extend(__A				)
              elif line.startswith(' ' * 8 + '"'				):
                     objects.append(line[9:-3]				)
              line_index += 1

       a_					:		Dict							=  {'none': objects}
       # Let's continue with backend-specific objects in _import_structure
       while not lines[line_index].startswith('if TYPE_CHECKING'				):
              # If the line is an if not is_backend_available, we grab all objects associated.
              a_					:		str							=  find_backend(lines[line_index]				)
              # Check if the backend declaration is inside a try block:
              if _re_try.search(lines[line_index - 1]				) is None:
                     a_					:		Dict							=  None

              if backend is not None:
                     line_index += 1

                     # Scroll until we hit the else block of try-except-else
                     while _re_else.search(lines[line_index]				) is None:
                            line_index += 1

                     line_index += 1

                     a_					:		Tuple							=  []
                     # Until we unindent, add backend objects to the list
                     while len(lines[line_index]				) <= 1 or lines[line_index].startswith(' ' * 4				):
                            a_					:		Optional[Any]							=  lines[line_index]
                            if _re_import_struct_add_one.search(__A				) is not None:
                                   objects.append(_re_import_struct_add_one.search(__A				).groups()[0]				)
                            elif _re_import_struct_add_many.search(__A				) is not None:
                                   a_					:		Tuple							=  _re_import_struct_add_many.search(__A				).groups()[0].split(', '				)
                                   a_					:		Tuple							=  [obj[1:-1] for obj in imports if len(__A				) > 0]
                                   objects.extend(__A				)
                            elif _re_between_brackets.search(__A				) is not None:
                                   a_					:		Tuple							=  _re_between_brackets.search(__A				).groups()[0].split(', '				)
                                   a_					:		Tuple							=  [obj[1:-1] for obj in imports if len(__A				) > 0]
                                   objects.extend(__A				)
                            elif _re_quote_object.search(__A				) is not None:
                                   objects.append(_re_quote_object.search(__A				).groups()[0]				)
                            elif line.startswith(' ' * 8 + '"'				):
                                   objects.append(line[9:-3]				)
                            elif line.startswith(' ' * 12 + '"'				):
                                   objects.append(line[13:-3]				)
                            line_index += 1

                     a_					:		str							=  objects
              else:
                     line_index += 1

    # At this stage we are in the TYPE_CHECKING part, first grab the objects without a specific backend
       a_					:		str							=  []
       while (
           line_index < len(__A				)
           and find_backend(lines[line_index]				) is None
           and not lines[line_index].startswith('else'				)
       ):
              a_					:		Optional[Any]							=  lines[line_index]
              a_					:		Any							=  _re_import.search(__A				)
              if single_line_import_search is not None:
                     objects.extend(single_line_import_search.groups()[0].split(', '				)				)
              elif line.startswith(' ' * 8				):
                     objects.append(line[8:-2]				)
              line_index += 1

       a_					:		Any							=  {'none': objects}
       # Let's continue with backend-specific objects
       while line_index < len(__A				):
              # If the line is an if is_backend_available, we grab all objects associated.
              a_					:		str							=  find_backend(lines[line_index]				)
              # Check if the backend declaration is inside a try block:
              if _re_try.search(lines[line_index - 1]				) is None:
                     a_					:		List[Any]							=  None

              if backend is not None:
                     line_index += 1

                     # Scroll until we hit the else block of try-except-else
                     while _re_else.search(lines[line_index]				) is None:
                            line_index += 1

                     line_index += 1

                     a_					:		Optional[Any]							=  []
                     # Until we unindent, add backend objects to the list
                     while len(lines[line_index]				) <= 1 or lines[line_index].startswith(' ' * 8				):
                            a_					:		Optional[Any]							=  lines[line_index]
                            a_					:		List[Any]							=  _re_import.search(__A				)
                            if single_line_import_search is not None:
                                   objects.extend(single_line_import_search.groups()[0].split(', '				)				)
                            elif line.startswith(' ' * 12				):
                                   objects.append(line[12:-2]				)
                            line_index += 1

                     a_					:		int							=  objects
              else:
                     line_index += 1

       return import_dict_objects, type_hint_objects







def        SCREAMING_SNAKE_CASE_						(			__A :			List[str]  ,    __A :			Any				)	->		int:

       """simple docstring"""

       def find_duplicates(__A :			List[Any]				):
              return [k for k, v in collections.Counter(__A				).items() if v > 1]

       if list(import_dict_objects.keys()				) != list(type_hint_objects.keys()				):
              return ["Both sides of the init do not have the same backends!"]

       a_					:		Tuple							=  []
       for key in import_dict_objects.keys():
              a_					:		Dict							=  find_duplicates(import_dict_objects[key]				)
              if duplicate_imports:
                     errors.append(F"""Duplicate _import_structure definitions for: {duplicate_imports}"""				)
              a_					:		List[str]							=  find_duplicates(type_hint_objects[key]				)
              if duplicate_type_hints:
                     errors.append(F"""Duplicate TYPE_CHECKING objects for: {duplicate_type_hints}"""				)

              if sorted(set(import_dict_objects[key]				)				) != sorted(set(type_hint_objects[key]				)				):
                     a_					:		Optional[Any]							=  'base imports' if key == 'none' else F"""{key} backend"""
                     errors.append(F"""Differences for {name}:"""				)
                     for a in type_hint_objects[key]:
                            if a not in import_dict_objects[key]:
                                   errors.append(F"""  {a} in TYPE_HINT but not in _import_structure."""				)
                     for a in import_dict_objects[key]:
                            if a not in type_hint_objects[key]:
                                   errors.append(F"""  {a} in _import_structure but not in TYPE_HINT."""				)
       return errors







def        SCREAMING_SNAKE_CASE_						(			)	->		Union[str, Any]:

       """simple docstring"""

       a_					:		Dict							=  []
       for root, _, files in os.walk(__A				):
              if "__init__.py" in files:
                     a_					:		List[Any]							=  os.path.join(__A  ,    '__init__.py'				)
                     a_					:		Optional[Any]							=  parse_init(__A				)
                     if objects is not None:
                            a_					:		List[Any]							=  analyze_results(*__A				)
                            if len(__A				) > 0:
                                   a_					:		Any							=  F"""Problem in {fname}, both halves do not define the same objects.\n{errors[0]}"""
                                   failures.append('\n'.join(__A				)				)
       if len(__A				) > 0:
              raise ValueError('\n\n'.join(__A				)				)







def        SCREAMING_SNAKE_CASE_						(			)	->		str:

       """simple docstring"""

       a_					:		str							=  []
       for path, directories, files in os.walk(__A				):
              for folder in directories:
                     # Ignore private modules
                     if folder.startswith('_'				):
                            directories.remove(__A				)
                            continue
                     # Ignore leftovers from branches (empty folders apart from pycache)
                     if len(list((Path(__A				) / folder).glob('*.py'				)				)				) == 0:
                            continue
                     a_					:		Optional[Any]							=  str((Path(__A				) / folder).relative_to(__A				)				)
                     a_					:		List[Any]							=  short_path.replace(os.path.sep  ,    '.'				)
                     submodules.append(__A				)
              for fname in files:
                     if fname == "__init__.py":
                            continue
                     a_					:		int							=  str((Path(__A				) / fname).relative_to(__A				)				)
                     a_					:		Optional[Any]							=  short_path.replace('.py'  ,    ''				).replace(os.path.sep  ,    '.'				)
                     if len(submodule.split('.'				)				) == 1:
                            submodules.append(__A				)
       return submodules


UpperCAmelCase_      : Dict  					=       [
    'convert_pytorch_checkpoint_to_tf2',
    'modeling_flax_pytorch_utils',
]







def        SCREAMING_SNAKE_CASE_						(			)	->		List[str]:

       """simple docstring"""

       a_					:		int							=  importlib.util.spec_from_file_location(
           'transformers'  ,    os.path.join(__A  ,    '__init__.py'				)  ,    submodule_search_locations=[PATH_TO_TRANSFORMERS]  ,    )
       a_					:		Dict							=  spec.loader.load_module()

       a_					:		List[str]							=  [
           module
           for module in get_transformers_submodules()
           if module not in IGNORE_SUBMODULES and module not in transformers._import_structure.keys()
       ]
       if len(__A				) > 0:
              a_					:		Union[str, Any]							=  '\n'.join(F"""- {module}""" for module in module_not_registered				)
              raise ValueError(
                  'The following submodules are not properly registered in the main init of Transformers:\n'
                  F"""{list_of_modules}\n"""
                  'Make sure they appear somewhere in the keys of `_import_structure` with an empty list as value.'				)


if __name__ == "__main__":
   check_all_inits()
   check_submodules()






import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
@skip_mps
class  SCREAMING_SNAKE_CASE__  (				lowercase__						,    unittest.TestCase			):
      snake_case__				:     Optional[Any]        =						TextToVideoSDPipeline
      snake_case__				:     Optional[int]        =						TEXT_TO_IMAGE_PARAMS
      snake_case__				:     str        =						TEXT_TO_IMAGE_BATCH_PARAMS
      # No `output_type`.
      snake_case__				:     Optional[Any]        =						frozenset(
          [
              '''num_inference_steps''',
              '''generator''',
              '''latents''',
              '''return_dict''',
              '''callback''',
              '''callback_steps''',
          ]			)
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			Union[str, Any]:
             torch.manual_seed(0     )
             a_					:		Optional[int]							=  UNetaDConditionModel(
                 block_out_channels=(3_2, 6_4, 6_4, 6_4)     ,							layers_per_block=2     ,							sample_size=3_2     ,							in_channels=4     ,							out_channels=4     ,							down_block_types=('CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'DownBlock3D')     ,							up_block_types=('UpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D')     ,							cross_attention_dim=3_2     ,							attention_head_dim=4     ,							)
             a_					:		int							=  DDIMScheduler(
                 beta_start=0.00085     ,							beta_end=0.012     ,							beta_schedule='scaled_linear'     ,							clip_sample=SCREAMING_SNAKE_CASE__     ,							set_alpha_to_one=SCREAMING_SNAKE_CASE__     ,							)
             torch.manual_seed(0     )
             a_					:		int							=  AutoencoderKL(
                 block_out_channels=[3_2, 6_4]     ,							in_channels=3     ,							out_channels=3     ,							down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']     ,							up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']     ,							latent_channels=4     ,							sample_size=1_2_8     ,							)
             torch.manual_seed(0     )
             a_					:		Optional[int]							=  CLIPTextConfig(
                 bos_token_id=0     ,							eos_token_id=2     ,							hidden_size=3_2     ,							intermediate_size=3_7     ,							layer_norm_eps=1E-05     ,							num_attention_heads=4     ,							num_hidden_layers=5     ,							pad_token_id=1     ,							vocab_size=1_0_0_0     ,							hidden_act='gelu'     ,							projection_dim=5_1_2     ,							)
             a_					:		Dict							=  CLIPTextModel(SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

             a_					:		Union[str, Any]							=  {
                 'unet': unet,
                 'scheduler': scheduler,
                 'vae': vae,
                 'text_encoder': text_encoder,
                 'tokenizer': tokenizer,
             }
             return components
      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     ,							SCREAMING_SNAKE_CASE__				:       str     ,							SCREAMING_SNAKE_CASE__				:       List[Any]=0     )		->			List[str]:
             if str(SCREAMING_SNAKE_CASE__     ).startswith('mps'     ):
                    a_					:		Dict							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             else:
                    a_					:		Optional[Any]							=  torch.Generator(device=SCREAMING_SNAKE_CASE__     ).manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		int							=  {
                 'prompt': 'A painting of a squirrel eating a burger',
                 'generator': generator,
                 'num_inference_steps': 2,
                 'guidance_scale': 6.0,
                 'output_type': 'pt',
             }
             return inputs
      def        SCREAMING_SNAKE_CASE							(       self				:       Union[str, Any]     )		->			Tuple:
             a_					:		List[Any]							=  'cpu'  # ensure determinism for the device-dependent torch.Generator
             a_					:		Dict							=  self.get_dummy_components()
             a_					:		str							=  TextToVideoSDPipeline(**SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Tuple							=  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__     )
             a_					:		Dict							=  'np'
             a_					:		Dict							=  sd_pipe(**SCREAMING_SNAKE_CASE__     ).frames
             a_					:		int							=  frames[0][-3:, -3:, -1]

             assert frames[0].shape == (6_4, 6_4, 3)
             a_					:		Union[str, Any]							=  np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0]     )

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-2
      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			List[Any]:
             self._test_attention_slicing_forward_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__     ,							expected_max_diff=3E-3     )
      @unittest.skipIf(
          torch_device != 'cuda' or not is_xformers_available()     ,							reason='XFormers attention is only available with CUDA and `xformers` installed'     ,							)
      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			str:
             self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__     ,							expected_max_diff=1E-2     )
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Union[str, Any]:
             pass
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			List[str]:
             pass
      @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       int     )		->			List[Any]:
             pass

      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Union[str, Any]:
             return super().test_progress_bar()
@slow
@skip_mps
class  SCREAMING_SNAKE_CASE__  (				unittest.TestCase			):
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[Any]     )		->			List[Any]:
             a_					:		str							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'     )

             a_					:		Any							=  TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'     )
             a_					:		Optional[Any]							=  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config     )
             a_					:		Optional[Any]							=  pipe.to('cuda'     )

             a_					:		Any							=  'Spiderman is surfing'
             a_					:		List[Any]							=  torch.Generator(device='cpu'     ).manual_seed(0     )

             a_					:		Optional[Any]							=  pipe(SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=2_5     ,							output_type='pt'     ).frames
             a_					:		str							=  video_frames.cpu().numpy()

             assert np.abs(expected_video - video     ).mean() < 5E-2

      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			Any:
             a_					:		Dict							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'     )

             a_					:		Tuple							=  TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'     )
             a_					:		Tuple							=  pipe.to('cuda'     )

             a_					:		Any							=  'Spiderman is surfing'
             a_					:		List[str]							=  torch.Generator(device='cpu'     ).manual_seed(0     )

             a_					:		List[Any]							=  pipe(SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=2     ,							output_type='pt'     ).frames
             a_					:		List[str]							=  video_frames.cpu().numpy()

             assert np.abs(expected_video - video     ).mean() < 5E-2


