






import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    AutoConfig,
    AutoImageProcessor,
    CLIPConfig,
    CLIPImageProcessor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER


sys.path.append(str(Path(__file__).parent.parent.parent.parent / "utils"))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_image_processing import CustomImageProcessor  # noqa E402
class        a       (					unittest.TestCase  ):



      """simple docstring"""


      def       UpperCAmelCase ( self   :     Dict   )				-> Optional[int]:
        __UpperCAmelCase				:      Union[str, Any]	 =    0
      def       UpperCAmelCase ( self   :     Optional[int]   )				-> Any:
        __UpperCAmelCase				:      str	 =    AutoImageProcessor.from_pretrained("""openai/clip-vit-base-patch32"""   )
        self.assertIsInstance(__lowercase						, __lowercase   )
      def       UpperCAmelCase ( self   :     Any   )				-> Tuple:
        with tempfile.TemporaryDirectory() as tmpdirname:
          __UpperCAmelCase				:      List[Any]	 =    Path(__lowercase   ) / """preprocessor_config.json"""
          __UpperCAmelCase				:      Dict	 =    Path(__lowercase   ) / """config.json"""
          json.dump(
              {"""image_processor_type""": """CLIPImageProcessor""", """processor_class""": """CLIPProcessor"""}						, open(__lowercase						, """w"""   )						, )
          json.dump({"""model_type""": """clip"""}						, open(__lowercase						, """w"""   )   )

          __UpperCAmelCase				:      Any	 =    AutoImageProcessor.from_pretrained(__lowercase   )
          self.assertIsInstance(__lowercase						, __lowercase   )
      def       UpperCAmelCase ( self   :     Any   )				-> Any:
        # Ensure we can load the image processor from the feature extractor config
        with tempfile.TemporaryDirectory() as tmpdirname:
          __UpperCAmelCase				:      Any	 =    Path(__lowercase   ) / """preprocessor_config.json"""
          __UpperCAmelCase				:      Tuple	 =    Path(__lowercase   ) / """config.json"""
          json.dump(
              {"""feature_extractor_type""": """CLIPFeatureExtractor""", """processor_class""": """CLIPProcessor"""}						, open(__lowercase						, """w"""   )						, )
          json.dump({"""model_type""": """clip"""}						, open(__lowercase						, """w"""   )   )

          __UpperCAmelCase				:      str	 =    AutoImageProcessor.from_pretrained(__lowercase   )
          self.assertIsInstance(__lowercase						, __lowercase   )
      def       UpperCAmelCase ( self   :     Union[str, Any]   )				-> Any:
        with tempfile.TemporaryDirectory() as tmpdirname:
          __UpperCAmelCase				:      Any	 =    CLIPConfig()

          # Create a dummy config file with image_proceesor_type
          __UpperCAmelCase				:      Union[str, Any]	 =    Path(__lowercase   ) / """preprocessor_config.json"""
          __UpperCAmelCase				:      Optional[Any]	 =    Path(__lowercase   ) / """config.json"""
          json.dump(
              {"""image_processor_type""": """CLIPImageProcessor""", """processor_class""": """CLIPProcessor"""}						, open(__lowercase						, """w"""   )						, )
          json.dump({"""model_type""": """clip"""}						, open(__lowercase						, """w"""   )   )

          # remove image_processor_type to make sure config.json alone is enough to load image processor locally
          __UpperCAmelCase				:      Dict	 =    AutoImageProcessor.from_pretrained(__lowercase   ).to_dict()

          config_dict.pop("""image_processor_type"""   )
          __UpperCAmelCase				:      List[Any]	 =    CLIPImageProcessor(**__lowercase   )

          # save in new folder
          model_config.save_pretrained(__lowercase   )
          config.save_pretrained(__lowercase   )

          __UpperCAmelCase				:      List[Any]	 =    AutoImageProcessor.from_pretrained(__lowercase   )

          # make sure private variable is not incorrectly saved
          __UpperCAmelCase				:      List[str]	 =    json.loads(config.to_json_string()   )
          self.assertTrue("""_processor_class""" not in dict_as_saved   )

        self.assertIsInstance(__lowercase						, __lowercase   )
      def       UpperCAmelCase ( self   :     Optional[int]   )				-> List[str]:
        with tempfile.TemporaryDirectory() as tmpdirname:
          __UpperCAmelCase				:      Optional[int]	 =    Path(__lowercase   ) / """preprocessor_config.json"""
          json.dump(
              {"""image_processor_type""": """CLIPImageProcessor""", """processor_class""": """CLIPProcessor"""}						, open(__lowercase						, """w"""   )						, )

          __UpperCAmelCase				:      Dict	 =    AutoImageProcessor.from_pretrained(__lowercase   )
          self.assertIsInstance(__lowercase						, __lowercase   )
      def       UpperCAmelCase ( self   :     Union[str, Any]   )				-> List[Any]:
        with self.assertRaisesRegex(
            __lowercase						, """clip-base is not a local folder and is not a valid model identifier"""   ):
          __UpperCAmelCase				:      Tuple	 =    AutoImageProcessor.from_pretrained("""clip-base"""   )
      def       UpperCAmelCase ( self   :     Union[str, Any]   )				-> Tuple:
        with self.assertRaisesRegex(
            __lowercase						, r"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""   ):
          __UpperCAmelCase				:      Tuple	 =    AutoImageProcessor.from_pretrained(__lowercase						, revision="""aaaaaa"""   )
      def       UpperCAmelCase ( self   :     Dict   )				-> Any:
        with self.assertRaisesRegex(
            __lowercase						, """hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."""						, ):
          __UpperCAmelCase				:      int	 =    AutoImageProcessor.from_pretrained("""hf-internal-testing/config-no-model"""   )
      def       UpperCAmelCase ( self   :     Optional[int]   )				-> List[str]:
        # If remote code is not set, we will time out when asking whether to load the model.
        with self.assertRaises(__lowercase   ):
          __UpperCAmelCase				:      Tuple	 =    AutoImageProcessor.from_pretrained("""hf-internal-testing/test_dynamic_image_processor"""   )
        # If remote code is disabled, we can't load this config.
        with self.assertRaises(__lowercase   ):
          __UpperCAmelCase				:      Optional[Any]	 =    AutoImageProcessor.from_pretrained(
              """hf-internal-testing/test_dynamic_image_processor"""						, trust_remote_code=__lowercase   )

        __UpperCAmelCase				:      Union[str, Any]	 =    AutoImageProcessor.from_pretrained(
            """hf-internal-testing/test_dynamic_image_processor"""						, trust_remote_code=__lowercase   )
        self.assertEqual(image_processor.__class__.__name__						, """NewImageProcessor"""   )

        # Test image processor can be reloaded.
        with tempfile.TemporaryDirectory() as tmp_dir:
          image_processor.save_pretrained(__lowercase   )
          __UpperCAmelCase				:      Optional[Any]	 =    AutoImageProcessor.from_pretrained(__lowercase						, trust_remote_code=__lowercase   )
        self.assertEqual(reloaded_image_processor.__class__.__name__						, """NewImageProcessor"""   )
      def       UpperCAmelCase ( self   :     Optional[int]   )				-> Optional[Any]:
        try:
          AutoConfig.register("""custom"""						, __lowercase   )
          AutoImageProcessor.register(__lowercase						, __lowercase   )
          # Trying to register something existing in the Transformers library will raise an error
          with self.assertRaises(__lowercase   ):
            AutoImageProcessor.register(__lowercase						, __lowercase   )

          with tempfile.TemporaryDirectory() as tmpdirname:
            __UpperCAmelCase				:      int	 =    Path(__lowercase   ) / """preprocessor_config.json"""
            __UpperCAmelCase				:      Any	 =    Path(__lowercase   ) / """config.json"""
            json.dump(
                {"""feature_extractor_type""": """CLIPFeatureExtractor""", """processor_class""": """CLIPProcessor"""}						, open(__lowercase						, """w"""   )						, )
            json.dump({"""model_type""": """clip"""}						, open(__lowercase						, """w"""   )   )

            __UpperCAmelCase				:      List[Any]	 =    CustomImageProcessor.from_pretrained(__lowercase   )

          # Now that the config is registered, it can be used as any other config with the auto-API
          with tempfile.TemporaryDirectory() as tmp_dir:
            image_processor.save_pretrained(__lowercase   )
            __UpperCAmelCase				:      Any	 =    AutoImageProcessor.from_pretrained(__lowercase   )
            self.assertIsInstance(__lowercase						, __lowercase   )

        finally:
          if "custom" in CONFIG_MAPPING._extra_content:
            del CONFIG_MAPPING._extra_content["custom"]
          if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
            del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]




      def       UpperCAmelCase ( self   :     Optional[Any]   )				-> str:
        class        a       (					lowercase__  ):



              """simple docstring"""


              a    :					Any							=      True

        try:
          AutoConfig.register("""custom"""						, __lowercase   )
          AutoImageProcessor.register(__lowercase						, __lowercase   )
          # If remote code is not set, the default is to use local
          __UpperCAmelCase				:      Dict	 =    AutoImageProcessor.from_pretrained("""hf-internal-testing/test_dynamic_image_processor"""   )
          self.assertEqual(image_processor.__class__.__name__						, """NewImageProcessor"""   )
          self.assertTrue(image_processor.is_local   )

          # If remote code is disabled, we load the local one.
          __UpperCAmelCase				:      Union[str, Any]	 =    AutoImageProcessor.from_pretrained(
              """hf-internal-testing/test_dynamic_image_processor"""						, trust_remote_code=__lowercase   )
          self.assertEqual(image_processor.__class__.__name__						, """NewImageProcessor"""   )
          self.assertTrue(image_processor.is_local   )

          # If remote is enabled, we load from the Hub
          __UpperCAmelCase				:      Optional[Any]	 =    AutoImageProcessor.from_pretrained(
              """hf-internal-testing/test_dynamic_image_processor"""						, trust_remote_code=__lowercase   )
          self.assertEqual(image_processor.__class__.__name__						, """NewImageProcessor"""   )
          self.assertTrue(not hasattr(__lowercase						, """is_local"""   )   )

        finally:
          if "custom" in CONFIG_MAPPING._extra_content:
            del CONFIG_MAPPING._extra_content["custom"]
          if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
            del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]








"""simple docstring"""






import gc
import unittest

import numpy as np
import torch

from diffusers import (
    AudioDiffusionPipeline,
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    DiffusionPipeline,
    Mel,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()






class lowerCAmelCase__      (	unittest.TestCase		):






			'''simple docstring'''



			def    A_       (    self    ):
							# clean up the VRAM after each test
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()
			@property
			def    A_       (    self    ):
							torch.manual_seed(0    )
							_lowerCamelCase	:     Union[str, Any]       =       UNetaDModel(
							    sample_size=(32, 64)    , in_channels=1    , out_channels=1    , layers_per_block=2    , block_out_channels=(128, 128)    , down_block_types=('AttnDownBlock2D', 'DownBlock2D')    , up_block_types=('UpBlock2D', 'AttnUpBlock2D')    , )
							return model
			@property
			def    A_       (    self    ):
							torch.manual_seed(0    )
							_lowerCamelCase	:     Dict       =       UNetaDConditionModel(
							    sample_size=(64, 32)    , in_channels=1    , out_channels=1    , layers_per_block=2    , block_out_channels=(128, 128)    , down_block_types=('CrossAttnDownBlock2D', 'DownBlock2D')    , up_block_types=('UpBlock2D', 'CrossAttnUpBlock2D')    , cross_attention_dim=10    , )
							return model
			@property
			def    A_       (    self    ):
							torch.manual_seed(0    )
							_lowerCamelCase	:     List[str]       =       AutoencoderKL(
							    sample_size=(128, 64)    , in_channels=1    , out_channels=1    , latent_channels=1    , layers_per_block=2    , block_out_channels=(128, 128)    , down_block_types=('DownEncoderBlock2D', 'DownEncoderBlock2D')    , up_block_types=('UpDecoderBlock2D', 'UpDecoderBlock2D')    , )
							_lowerCamelCase	:     int       =       UNetaDModel(
							    sample_size=(64, 32)    , in_channels=1    , out_channels=1    , layers_per_block=2    , block_out_channels=(128, 128)    , down_block_types=('AttnDownBlock2D', 'DownBlock2D')    , up_block_types=('UpBlock2D', 'AttnUpBlock2D')    , )
							return vqvae, unet






			@slow
			def    A_       (    self    ):
							_lowerCamelCase	:     str       =       'cpu'  # ensure determinism for the device-dependent torch.Generator
							_lowerCamelCase	:     Optional[int]       =       Mel(
							    x_res=self.dummy_unet.config.sample_size[1]    , y_res=self.dummy_unet.config.sample_size[0]    , )

							_lowerCamelCase	:     List[str]       =       DDPMScheduler()
							_lowerCamelCase	:     List[Any]       =       AudioDiffusionPipeline(vqvae=lowercase    , unet=self.dummy_unet    , mel=lowercase    , scheduler=lowercase    )
							_lowerCamelCase	:     List[Any]       =       pipe.to(lowercase    )
							pipe.set_progress_bar_config(disable=lowercase    )

							_lowerCamelCase	:     Union[str, Any]       =       torch.Generator(device=lowercase    ).manual_seed(42    )
							_lowerCamelCase	:     Union[str, Any]       =       pipe(generator=lowercase    , steps=4    )
							_lowerCamelCase	:     Optional[Any]       =       output.audios[0]
							_lowerCamelCase	:     int       =       output.images[0]

							_lowerCamelCase	:     Dict       =       torch.Generator(device=lowercase    ).manual_seed(42    )
							_lowerCamelCase	:     Dict       =       pipe(generator=lowercase    , steps=4    , return_dict=lowercase    )
							_lowerCamelCase	:     List[str]       =       output[0][0]

							assert audio.shape == (1, (self.dummy_unet.config.sample_size[1] - 1) * mel.hop_length)
							assert (
							    image.height == self.dummy_unet.config.sample_size[0]
							    and image.width == self.dummy_unet.config.sample_size[1]
							)
							_lowerCamelCase	:     Optional[Any]       =       np.frombuffer(image.tobytes()    , dtype='uint8'    )[:10]
							_lowerCamelCase	:     Dict       =       np.frombuffer(image_from_tuple.tobytes()    , dtype='uint8'    )[:10]
							_lowerCamelCase	:     List[str]       =       np.array([69, 255, 255, 255, 0, 0, 77, 181, 12, 127]    )

							assert np.abs(image_slice.flatten() - expected_slice    ).max() == 0
							assert np.abs(image_from_tuple_slice.flatten() - expected_slice    ).max() == 0

							_lowerCamelCase	:     List[Any]       =       Mel(
							    x_res=self.dummy_vqvae_and_unet[0].config.sample_size[1]    , y_res=self.dummy_vqvae_and_unet[0].config.sample_size[0]    , )

							_lowerCamelCase	:     Dict       =       DDIMScheduler()
							_lowerCamelCase	:     Tuple       =       self.dummy_vqvae_and_unet
							_lowerCamelCase	:     List[str]       =       AudioDiffusionPipeline(
							    vqvae=self.dummy_vqvae_and_unet[0]    , unet=dummy_vqvae_and_unet[1]    , mel=lowercase    , scheduler=lowercase    )
							_lowerCamelCase	:     Tuple       =       pipe.to(lowercase    )
							pipe.set_progress_bar_config(disable=lowercase    )

							np.random.seed(0    )
							_lowerCamelCase	:     Any       =       np.random.uniform(-1    , 1    , ((dummy_vqvae_and_unet[0].config.sample_size[1] - 1) * mel.hop_length,)    )
							_lowerCamelCase	:     Optional[int]       =       torch.Generator(device=lowercase    ).manual_seed(42    )
							_lowerCamelCase	:     Dict       =       pipe(raw_audio=lowercase    , generator=lowercase    , start_step=5    , steps=10    )
							_lowerCamelCase	:     str       =       output.images[0]

							assert (
							    image.height == self.dummy_vqvae_and_unet[0].config.sample_size[0]
							    and image.width == self.dummy_vqvae_and_unet[0].config.sample_size[1]
							)
							_lowerCamelCase	:     Dict       =       np.frombuffer(image.tobytes()    , dtype='uint8'    )[:10]
							_lowerCamelCase	:     Union[str, Any]       =       np.array([120, 117, 110, 109, 138, 167, 138, 148, 132, 121]    )

							assert np.abs(image_slice.flatten() - expected_slice    ).max() == 0

							_lowerCamelCase	:     Dict       =       self.dummy_unet_condition
							_lowerCamelCase	:     Optional[int]       =       AudioDiffusionPipeline(
							    vqvae=self.dummy_vqvae_and_unet[0]    , unet=lowercase    , mel=lowercase    , scheduler=lowercase    )
							_lowerCamelCase	:     Dict       =       pipe.to(lowercase    )
							pipe.set_progress_bar_config(disable=lowercase    )

							np.random.seed(0    )
							_lowerCamelCase	:     Optional[int]       =       torch.rand((1, 1, 10)    )
							_lowerCamelCase	:     Optional[Any]       =       pipe(generator=lowercase    , encoding=lowercase    )
							_lowerCamelCase	:     Dict       =       output.images[0]
							_lowerCamelCase	:     Tuple       =       np.frombuffer(image.tobytes()    , dtype='uint8'    )[:10]
							_lowerCamelCase	:     List[str]       =       np.array([107, 103, 120, 127, 142, 122, 113, 122, 97, 111]    )

							assert np.abs(image_slice.flatten() - expected_slice    ).max() == 0






@slow
@require_torch_gpu
class lowerCAmelCase__      (	unittest.TestCase		):






			'''simple docstring'''



			def    A_       (    self    ):
							# clean up the VRAM after each test
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()






			def    A_       (    self    ):
							_lowerCamelCase	:     Optional[Any]       =       torch_device

							_lowerCamelCase	:     Dict       =       DiffusionPipeline.from_pretrained('teticio/audio-diffusion-ddim-256'    )
							_lowerCamelCase	:     List[str]       =       pipe.to(lowercase    )
							pipe.set_progress_bar_config(disable=lowercase    )

							_lowerCamelCase	:     int       =       torch.Generator(device=lowercase    ).manual_seed(42    )
							_lowerCamelCase	:     Tuple       =       pipe(generator=lowercase    )
							_lowerCamelCase	:     Dict       =       output.audios[0]
							_lowerCamelCase	:     Dict       =       output.images[0]

							assert audio.shape == (1, (pipe.unet.config.sample_size[1] - 1) * pipe.mel.hop_length)
							assert image.height == pipe.unet.config.sample_size[0] and image.width == pipe.unet.config.sample_size[1]
							_lowerCamelCase	:     Optional[int]       =       np.frombuffer(image.tobytes()    , dtype='uint8'    )[:10]
							_lowerCamelCase	:     int       =       np.array([151, 167, 154, 144, 122, 134, 121, 105, 70, 26]    )

							assert np.abs(image_slice.flatten() - expected_slice    ).max() == 0