
"""simple docstring"""




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 			__magic_name__       (       UpperCamelCase_   ,					unittest.TestCase ):
       UpperCamelCase_   =		VideoToVideoSDPipeline
       UpperCamelCase_   =		TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'''video'''} ) - {'''image''', '''width''', '''height'''}
       UpperCamelCase_   =		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''video'''} ) - {'''image'''}
       UpperCamelCase_   =		PipelineTesterMixin.required_optional_params - {'''latents'''}
       UpperCamelCase_   =		False

       # No `output_type`.
       UpperCamelCase_   =		frozenset(
           [
               '''num_inference_steps''',
               '''generator''',
               '''latents''',
               '''return_dict''',
               '''callback''',
               '''callback_steps''',
           ] )
       def 							lowercase_							(     self       )     ->	Tuple:





         """simple docstring"""




         torch.manual_seed(0       )
         _lowercase:			int       =     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 ,				)
         _lowercase:			Optional[Any]       =     DDIMScheduler(
             beta_start=0.0_00_85 ,				beta_end=0.0_12 ,				beta_schedule='''scaled_linear''' ,				clip_sample=__a ,				set_alpha_to_one=__a ,				)
         torch.manual_seed(0       )
         _lowercase:			Dict       =     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       )
         _lowercase:			Union[str, Any]       =     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 ,				)
         _lowercase:			int       =     CLIPTextModel(__a       )
         _lowercase:			List[str]       =     CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''       )

         _lowercase:			Optional[Any]       =     {
             'unet': unet,
             'scheduler': scheduler,
             'vae': vae,
             'text_encoder': text_encoder,
             'tokenizer': tokenizer,
         }
         return components
       def 							lowercase_							(     self ,				A_ ,				A_=0       )     ->	Tuple:





         """simple docstring"""




         _lowercase:			Optional[int]       =     floats_tensor((1, 3, 3, 32, 32) ,				rng=random.Random(__a       )       ).to(__a       )

         if str(__a       ).startswith('''mps'''       ):
           _lowercase:			int       =     torch.manual_seed(__a       )
         else:
           _lowercase:			List[Any]       =     torch.Generator(device=__a       ).manual_seed(__a       )
         _lowercase:			List[Any]       =     {
             '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 							lowercase_							(     self       )     ->	Any:





         """simple docstring"""




         _lowercase:			Optional[int]       =     'cpu'  # ensure determinism for the device-dependent torch.Generator
         _lowercase:			str       =     self.get_dummy_components()
         _lowercase:			List[Any]       =     VideoToVideoSDPipeline(**__a       )
         _lowercase:			Tuple       =     sd_pipe.to(__a       )
         sd_pipe.set_progress_bar_config(disable=__a       )

         _lowercase:			Dict       =     self.get_dummy_inputs(__a       )
         _lowercase:			Dict       =     'np'
         _lowercase:			Tuple       =     sd_pipe(**__a       ).frames
         _lowercase:			Any       =     frames[0][-3:, -3:, -1]

         assert frames[0].shape == (32, 32, 3)
         _lowercase:			List[str]       =     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 							lowercase_							(     self       )     ->	Optional[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 							lowercase_							(     self       )     ->	Union[str, Any]:





         """simple docstring"""




         pass
       @unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''       )
       def 							lowercase_							(     self       )     ->	List[Any]:





         """simple docstring"""




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





         """simple docstring"""




         pass



       def 							lowercase_							(     self       )     ->	List[Any]:





         """simple docstring"""




         return super().test_progress_bar()


@slow
@skip_mps
class 			__magic_name__       (       unittest.TestCase ):
       def 							lowercase_							(     self       )     ->	Union[str, Any]:





         """simple docstring"""




         _lowercase:			Optional[int]       =     VideoToVideoSDPipeline.from_pretrained('''cerspense/zeroscope_v2_XL''' ,				torch_dtype=torch.floataa       )
         pipe.enable_model_cpu_offload()

         # 10 frames
         _lowercase:			List[Any]       =     torch.Generator(device='''cpu'''       ).manual_seed(0       )
         _lowercase:			List[str]       =     torch.randn((1, 10, 3, 1024, 576) ,				generator=__a       )
         _lowercase:			Dict       =     video.to('''cuda'''       )

         _lowercase:			Optional[int]       =     'Spiderman is surfing'

         _lowercase:			List[str]       =     pipe(__a ,				video=__a ,				generator=__a ,				num_inference_steps=3 ,				output_type='''pt'''       ).frames

         _lowercase:			Tuple       =     np.array([-1.0_45_89_84, -1.1_27_92_97, -0.9_66_30_86, -0.91_50_39_06, -0.75_09_76_56]       )
         assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array       ).sum() < 1E-2


"""simple docstring"""




import os
import re
import shutil
from argparse import ArgumentParser, Namespace

from datasets.commands import BaseDatasetsCLICommand
from datasets.utils.logging import get_logger


A__   :					Optional[Any]							=   '<<<<<<< This should probably be modified because it mentions: '

A__   :					Any							=   '=======\n>>>>>>>\n'

A__   :					int							=   [
    'TextEncoderConfig',
    'ByteTextEncoder',
    'SubwordTextEncoder',
    'encoder_config',
    'maybe_build_from_corpus',
    'manual_dir',
]

A__   :					Union[str, Any]							=   [
    # (pattern, replacement)
    # Order is important here for some replacements
    (r'tfds\.core', r'datasets'),
    (r'tf\.io\.gfile\.GFile', r'open'),
    (r'tf\.([\w\d]+)', r'datasets.Value(\'\1\')'),
    (r'tfds\.features\.Text\(\)', r'datasets.Value(\'string\')'),
    (r'tfds\.features\.Text\(', r'datasets.Value(\'string\'),'),
    (r'features\s*=\s*tfds.features.FeaturesDict\(', r'features=datasets.Features('),
    (r'tfds\.features\.FeaturesDict\(', r'dict('),
    (r'The TensorFlow Datasets Authors', r'The TensorFlow Datasets Authors and the HuggingFace Datasets Authors'),
    (r'tfds\.', r'datasets.'),
    (r'dl_manager\.manual_dir', r'self.config.data_dir'),
    (r'self\.builder_config', r'self.config'),
]




def  _lowerCAmelCase		(		_UpperCamelCase						):






  """simple docstring"""



  return ConvertCommand(args.tfds_path ,				args.datasets_directory						)







class 			__magic_name__       (       SCREAMING_SNAKE_CASE__ ):
       @staticmethod
       def 							lowercase_							(     A_       )     ->	Union[str, Any]:





         """simple docstring"""




         _lowercase:			int       =     parser.add_parser(
             '''convert''' ,				help='''Convert a TensorFlow Datasets dataset to a HuggingFace Datasets dataset.''' ,				)
         train_parser.add_argument(
             '''--tfds_path''' ,				type=A_ ,				required=A_ ,				help='''Path to a TensorFlow Datasets folder to convert or a single tfds file to convert.''' ,				)
         train_parser.add_argument(
             '''--datasets_directory''' ,				type=A_ ,				required=A_ ,				help='''Path to the HuggingFace Datasets folder.'''       )
         train_parser.set_defaults(func=A_       )
       def __init__(     self ,				A_ ,				A_ ,				*A_       )     ->	Tuple:





         """simple docstring"""




         _lowercase:			Optional[Any]       =     get_logger('''datasets-cli/converting'''       )

         _lowercase:			Optional[Any]       =     tfds_path
         _lowercase:			Dict       =     datasets_directory



       def 							lowercase_							(     self       )     ->	Any:





         """simple docstring"""




         if os.path.isdir(self._tfds_path       ):
           _lowercase:			Optional[Any]       =     os.path.abspath(self._tfds_path       )
         elif os.path.isfile(self._tfds_path       ):
           _lowercase:			Any       =     os.path.dirname(self._tfds_path       )
         else:
           raise ValueError('''--tfds_path is neither a directory nor a file. Please check path.'''       )

         _lowercase:			Dict       =     os.path.abspath(self._datasets_directory       )

         self._logger.info(f'''Converting datasets from {abs_tfds_path} to {abs_datasets_path}'''       )

         _lowercase:			Tuple       =     []
         _lowercase:			List[str]       =     []
         _lowercase:			int       =     {}

         if os.path.isdir(self._tfds_path       ):
           _lowercase:			str       =     os.listdir(A_       )
         else:
           _lowercase:			Tuple       =     [os.path.basename(self._tfds_path       )]

         for f_name in file_names:
           self._logger.info(f'''Looking at file {f_name}'''       )
           _lowercase:			Optional[int]       =     os.path.join(A_ ,				A_       )
           _lowercase:			Optional[Any]       =     os.path.join(A_ ,				A_       )

           if not os.path.isfile(A_       ) or "__init__" in f_name or "_test" in f_name or ".py" not in f_name:
             self._logger.info('''Skipping file'''       )
             continue

           with open(A_ ,				encoding='''utf-8'''       ) as f:
             _lowercase:			Any       =     f.readlines()

           _lowercase:			str       =     []
           _lowercase:			Dict       =     False
           _lowercase:			Any       =     False
           _lowercase:			Tuple       =     []
           for line in lines:
             _lowercase:			str       =     line

             # Convert imports
             if "import tensorflow.compat.v2 as tf" in out_line:
               continue
             elif "@tfds.core" in out_line:
               continue
             elif "builder=self" in out_line:
               continue
             elif "import tensorflow_datasets.public_api as tfds" in out_line:
               _lowercase:			Optional[Any]       =     '''import datasets\n'''
             elif "import tensorflow" in out_line:
               # order is important here
               _lowercase:			Any       =     ''''''
               continue
             elif "from absl import logging" in out_line:
               _lowercase:			Dict       =     '''from datasets import logging\n'''
             elif "getLogger" in out_line:
               _lowercase:			Optional[Any]       =     out_line.replace('''getLogger''' ,				'''get_logger'''       )
             elif any(expression in out_line for expression in TO_HIGHLIGHT       ):
               _lowercase:			int       =     True
               _lowercase:			List[str]       =     list(filter(lambda A_       : e in out_line ,				A_       )       )
               out_lines.append(HIGHLIGHT_MESSAGE_PRE + str(A_       ) + '''\n'''       )
               out_lines.append(A_       )
               out_lines.append(A_       )
               continue
             else:
               for pattern, replacement in TO_CONVERT:
                 _lowercase:			Tuple       =     re.sub(A_ ,				A_ ,				A_       )

                # Take care of saving utilities (to later move them together with main script)
             if "tensorflow_datasets" in out_line:
               _lowercase:			Optional[int]       =     re.match(R'''from\stensorflow_datasets.*import\s([^\.\r\n]+)''' ,				A_       )
               tfds_imports.extend(imp.strip() for imp in match.group(1       ).split(''','''       )       )
               _lowercase:			Dict       =     '''from . import ''' + match.group(1       )

             # Check we have not forget anything
             if "tf." in out_line or "tfds." in out_line or "tensorflow_datasets" in out_line:
               raise ValueError(f'''Error converting {out_line.strip()}'''       )

             if "GeneratorBasedBuilder" in out_line or "BeamBasedBuilder" in out_line:
               _lowercase:			Tuple       =     True
             out_lines.append(A_       )

           if is_builder or "wmt" in f_name:
             # We create a new directory for each dataset
             _lowercase:			List[str]       =     f_name.replace('''.py''' ,				''''''       )
             _lowercase:			Dict       =     os.path.join(A_ ,				A_       )
             _lowercase:			Dict       =     os.path.join(A_ ,				A_       )
             os.makedirs(A_ ,				exist_ok=A_       )
             self._logger.info(f'''Adding directory {output_dir}'''       )
             imports_to_builder_map.update({imp: output_dir for imp in tfds_imports}       )
           else:
             # Utilities will be moved at the end
             utils_files.append(A_       )

           if needs_manual_update:
             with_manual_update.append(A_       )

           with open(A_ ,				'''w''' ,				encoding='''utf-8'''       ) as f:
             f.writelines(A_       )
           self._logger.info(f'''Converted in {output_file}'''       )

         for utils_file in utils_files:
           try:
             _lowercase:			Optional[int]       =     os.path.basename(A_       )
             _lowercase:			List[Any]       =     imports_to_builder_map[f_name.replace('''.py''' ,				''''''       )]
             self._logger.info(f'''Moving {dest_folder} to {utils_file}'''       )
             shutil.copy(A_ ,				A_       )
           except KeyError:
             self._logger.error(f'''Cannot find destination folder for {utils_file}. Please copy manually.'''       )

         if with_manual_update:
           for file_path in with_manual_update:
             self._logger.warning(
                 f'''You need to manually update file {file_path} to remove configurations using \'TextEncoderConfig\'.'''       )
