




from typing import TYPE_CHECKING

from ...file_utils import _LazyModule, is_torch_available
from ...utils import OptionalDependencyNotAvailable


lowerCamelCase	:	Dict				=   {
    'configuration_gpt_neox_japanese': ['GPT_NEOX_JAPANESE_PRETRAINED_CONFIG_ARCHIVE_MAP', 'GPTNeoXJapaneseConfig'],
    'tokenization_gpt_neox_japanese': ['GPTNeoXJapaneseTokenizer'],
}

try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       lowerCamelCase	:	Optional[Any]				=   [
           'GPT_NEOX_JAPANESE_PRETRAINED_MODEL_ARCHIVE_LIST',
           'GPTNeoXJapaneseForCausalLM',
           'GPTNeoXJapaneseLayer',
           'GPTNeoXJapaneseModel',
           'GPTNeoXJapanesePreTrainedModel',
       ]


if TYPE_CHECKING:
       from .configuration_gpt_neox_japanese import GPT_NEOX_JAPANESE_PRETRAINED_CONFIG_ARCHIVE_MAP, GPTNeoXJapaneseConfig
       from .tokenization_gpt_neox_japanese import GPTNeoXJapaneseTokenizer

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_gpt_neox_japanese import (
                  GPT_NEOX_JAPANESE_PRETRAINED_MODEL_ARCHIVE_LIST,
                  GPTNeoXJapaneseForCausalLM,
                  GPTNeoXJapaneseLayer,
                  GPTNeoXJapaneseModel,
                  GPTNeoXJapanesePreTrainedModel,
              )


else:
       import sys

       lowerCamelCase	:	Any				=   _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)











import gc
import unittest

import numpy as np
import torch
from torch.backends.cuda import sdp_kernel

from diffusers import (
    CMStochasticIterativeScheduler,
    ConsistencyModelPipeline,
    UNetaDModel,
)
from diffusers.utils import randn_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_a, require_torch_gpu

from ..pipeline_params import UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS, UNCONDITIONAL_IMAGE_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class 						__lowercase		(UpperCamelCase__					,		unittest.TestCase							):




      """simple docstring"""





      _snake_case												=						ConsistencyModelPipeline
      _snake_case												=						UNCONDITIONAL_IMAGE_GENERATION_PARAMS
      _snake_case												=						UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

      # Override required_optional_params to remove num_images_per_prompt
      _snake_case												=						frozenset(
          [
              """num_inference_steps""",
              """generator""",
              """latents""",
              """output_type""",
              """return_dict""",
              """callback""",
              """callback_steps""",
          ]							)

      @property
      def      UpperCAmelCase				(				self     )		->    List[str]:
         snake_case				:    Dict           =	UNetaDModel.from_pretrained(
             """diffusers/consistency-models-test"""    ,				subfolder="""test_unet"""    ,				)
         return unet

      @property
      def      UpperCAmelCase				(				self     )		->    Any:
         snake_case				:    Optional[int]           =	UNetaDModel.from_pretrained(
             """diffusers/consistency-models-test"""    ,				subfolder="""test_unet_class_cond"""    ,				)
         return unet

      def      UpperCAmelCase				(				self    ,				A=False     )		->    Optional[Any]:
         if class_cond:
            snake_case				:    List[str]           =	self.dummy_cond_unet
         else:
            snake_case				:    Optional[Any]           =	self.dummy_uncond_unet

         # Default to CM multistep sampler
         snake_case				:    Optional[Any]           =	CMStochasticIterativeScheduler(
             num_train_timesteps=4_0    ,				sigma_min=0.0_02    ,				sigma_max=80.0    ,				)

         snake_case				:    Dict           =	{
             """unet""": unet,
             """scheduler""": scheduler,
         }

         return components

      def      UpperCAmelCase				(				self    ,				A    ,				A=0     )		->    Optional[int]:
         if str(A     ).startswith("""mps"""     ):
            snake_case				:    Union[str, Any]           =	torch.manual_seed(A     )
         else:
            snake_case				:    Dict           =	torch.Generator(device=A     ).manual_seed(A     )

         snake_case				:    Tuple           =	{
             """batch_size""": 1,
             """num_inference_steps""": None,
             """timesteps""": [2_2, 0],
             """generator""": generator,
             """output_type""": """np""",
         }

         return inputs

      def      UpperCAmelCase				(				self     )		->    List[Any]:
         snake_case				:    Union[str, Any]           =	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         snake_case				:    List[Any]           =	self.get_dummy_components()
         snake_case				:    Dict           =	ConsistencyModelPipeline(**A     )
         snake_case				:    int           =	pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )

         snake_case				:    Any           =	self.get_dummy_inputs(A     )
         snake_case				:    Tuple           =	pipe(**A     ).images
         assert image.shape == (1, 3_2, 3_2, 3)

         snake_case				:    Dict           =	image[0, -3:, -3:, -1]
         snake_case				:    Tuple           =	np.array([0.35_72, 0.62_73, 0.40_31, 0.39_61, 0.43_21, 0.57_30, 0.52_66, 0.47_80, 0.50_04]     )

         assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-3

      def      UpperCAmelCase				(				self     )		->    List[Any]:
         snake_case				:    Optional[int]           =	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         snake_case				:    Tuple           =	self.get_dummy_components(class_cond=A     )
         snake_case				:    Dict           =	ConsistencyModelPipeline(**A     )
         snake_case				:    List[Any]           =	pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )

         snake_case				:    str           =	self.get_dummy_inputs(A     )
         snake_case				:    Optional[int]           =	0
         snake_case				:    List[Any]           =	pipe(**A     ).images
         assert image.shape == (1, 3_2, 3_2, 3)

         snake_case				:    Optional[int]           =	image[0, -3:, -3:, -1]
         snake_case				:    Dict           =	np.array([0.35_72, 0.62_73, 0.40_31, 0.39_61, 0.43_21, 0.57_30, 0.52_66, 0.47_80, 0.50_04]     )

         assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-3

      def      UpperCAmelCase				(				self     )		->    Dict:
         snake_case				:    str           =	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         snake_case				:    Optional[int]           =	self.get_dummy_components()
         snake_case				:    List[str]           =	ConsistencyModelPipeline(**A     )
         snake_case				:    Dict           =	pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )

         snake_case				:    Dict           =	self.get_dummy_inputs(A     )
         snake_case				:    Tuple           =	1
         snake_case				:    Optional[Any]           =	None
         snake_case				:    List[Any]           =	pipe(**A     ).images
         assert image.shape == (1, 3_2, 3_2, 3)

         snake_case				:    Optional[Any]           =	image[0, -3:, -3:, -1]
         snake_case				:    List[str]           =	np.array([0.50_04, 0.50_04, 0.49_94, 0.50_08, 0.49_76, 0.50_18, 0.49_90, 0.49_82, 0.49_87]     )

         assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-3

      def      UpperCAmelCase				(				self     )		->    Optional[Any]:
         snake_case				:    Optional[int]           =	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         snake_case				:    Dict           =	self.get_dummy_components(class_cond=A     )
         snake_case				:    List[Any]           =	ConsistencyModelPipeline(**A     )
         snake_case				:    List[Any]           =	pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )

         snake_case				:    Optional[Any]           =	self.get_dummy_inputs(A     )
         snake_case				:    Optional[Any]           =	1
         snake_case				:    Any           =	None
         snake_case				:    Optional[int]           =	0
         snake_case				:    List[Any]           =	pipe(**A     ).images
         assert image.shape == (1, 3_2, 3_2, 3)

         snake_case				:    Optional[int]           =	image[0, -3:, -3:, -1]
         snake_case				:    List[str]           =	np.array([0.50_04, 0.50_04, 0.49_94, 0.50_08, 0.49_76, 0.50_18, 0.49_90, 0.49_82, 0.49_87]     )

         assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-3




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




      """simple docstring"""





      def      UpperCAmelCase				(				self     )		->    Dict:
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()

      def      UpperCAmelCase				(				self    ,				A=0    ,				A=False    ,				A="cpu"    ,				A=torch.floataa    ,				A=(1, 3, 6_4, 6_4)     )		->    int:
         snake_case				:    Union[str, Any]           =	torch.manual_seed(A     )

         snake_case				:    Tuple           =	{
             """num_inference_steps""": None,
             """timesteps""": [2_2, 0],
             """class_labels""": 0,
             """generator""": generator,
             """output_type""": """np""",
         }

         if get_fixed_latents:
            snake_case				:    Optional[int]           =	self.get_fixed_latents(seed=A    ,				device=A    ,				dtype=A    ,				shape=A     )
            snake_case				:    int           =	latents

         return inputs

      def      UpperCAmelCase				(				self    ,				A=0    ,				A="cpu"    ,				A=torch.floataa    ,				A=(1, 3, 6_4, 6_4)     )		->    Any:
         if type(A     ) == str:
            snake_case				:    List[str]           =	torch.device(A     )
         snake_case				:    Any           =	torch.Generator(device=A     ).manual_seed(A     )
         snake_case				:    Dict           =	randn_tensor(A    ,				generator=A    ,				device=A    ,				dtype=A     )
         return latents

      def      UpperCAmelCase				(				self     )		->    Dict:
         snake_case				:    List[Any]           =	UNetaDModel.from_pretrained("""diffusers/consistency_models"""    ,				subfolder="""diffusers_cd_imagenet64_l2"""     )
         snake_case				:    Optional[int]           =	CMStochasticIterativeScheduler(
             num_train_timesteps=4_0    ,				sigma_min=0.0_02    ,				sigma_max=80.0    ,				)
         snake_case				:    Any           =	ConsistencyModelPipeline(unet=A    ,				scheduler=A     )
         pipe.to(torch_device=A     )
         pipe.set_progress_bar_config(disable=A     )

         snake_case				:    Tuple           =	self.get_inputs()
         snake_case				:    List[str]           =	pipe(**A     ).images
         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case				:    Optional[Any]           =	image[0, -3:, -3:, -1]

         snake_case				:    Union[str, Any]           =	np.array([0.08_88, 0.08_81, 0.06_66, 0.04_79, 0.02_92, 0.01_95, 0.02_01, 0.01_63, 0.02_54]     )

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

      def      UpperCAmelCase				(				self     )		->    Tuple:
         snake_case				:    Tuple           =	UNetaDModel.from_pretrained("""diffusers/consistency_models"""    ,				subfolder="""diffusers_cd_imagenet64_l2"""     )
         snake_case				:    int           =	CMStochasticIterativeScheduler(
             num_train_timesteps=4_0    ,				sigma_min=0.0_02    ,				sigma_max=80.0    ,				)
         snake_case				:    List[str]           =	ConsistencyModelPipeline(unet=A    ,				scheduler=A     )
         pipe.to(torch_device=A     )
         pipe.set_progress_bar_config(disable=A     )

         snake_case				:    Union[str, Any]           =	self.get_inputs()
         snake_case				:    Tuple           =	1
         snake_case				:    List[str]           =	None
         snake_case				:    Any           =	pipe(**A     ).images
         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case				:    List[Any]           =	image[0, -3:, -3:, -1]

         snake_case				:    Union[str, Any]           =	np.array([0.03_40, 0.01_52, 0.00_63, 0.02_67, 0.02_21, 0.01_07, 0.04_16, 0.01_86, 0.02_17]     )

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

      @require_torch_a
      def      UpperCAmelCase				(				self     )		->    Optional[int]:
         snake_case				:    List[Any]           =	UNetaDModel.from_pretrained("""diffusers/consistency_models"""    ,				subfolder="""diffusers_cd_imagenet64_l2"""     )
         snake_case				:    Optional[Any]           =	CMStochasticIterativeScheduler(
             num_train_timesteps=4_0    ,				sigma_min=0.0_02    ,				sigma_max=80.0    ,				)
         snake_case				:    List[Any]           =	ConsistencyModelPipeline(unet=A    ,				scheduler=A     )
         pipe.to(torch_device=A    ,				torch_dtype=torch.floataa     )
         pipe.set_progress_bar_config(disable=A     )

         snake_case				:    List[str]           =	self.get_inputs(get_fixed_latents=A    ,				device=A     )
         # Ensure usage of flash attention in torch 2.0
         with sdp_kernel(enable_flash=A    ,				enable_math=A    ,				enable_mem_efficient=A     ):
            snake_case				:    Tuple           =	pipe(**A     ).images
         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case				:    Union[str, Any]           =	image[0, -3:, -3:, -1]

         snake_case				:    int           =	np.array([0.18_75, 0.14_28, 0.12_89, 0.21_51, 0.20_92, 0.14_77, 0.18_77, 0.16_41, 0.13_53]     )

         assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-3

      @require_torch_a
      def      UpperCAmelCase				(				self     )		->    Optional[Any]:
         snake_case				:    Optional[Any]           =	UNetaDModel.from_pretrained("""diffusers/consistency_models"""    ,				subfolder="""diffusers_cd_imagenet64_l2"""     )
         snake_case				:    Any           =	CMStochasticIterativeScheduler(
             num_train_timesteps=4_0    ,				sigma_min=0.0_02    ,				sigma_max=80.0    ,				)
         snake_case				:    Dict           =	ConsistencyModelPipeline(unet=A    ,				scheduler=A     )
         pipe.to(torch_device=A    ,				torch_dtype=torch.floataa     )
         pipe.set_progress_bar_config(disable=A     )

         snake_case				:    Optional[int]           =	self.get_inputs(get_fixed_latents=A    ,				device=A     )
         snake_case				:    Union[str, Any]           =	1
         snake_case				:    Dict           =	None
         # Ensure usage of flash attention in torch 2.0
         with sdp_kernel(enable_flash=A    ,				enable_math=A    ,				enable_mem_efficient=A     ):
            snake_case				:    List[Any]           =	pipe(**A     ).images
         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case				:    List[Any]           =	image[0, -3:, -3:, -1]

         snake_case				:    Dict           =	np.array([0.16_63, 0.19_48, 0.22_75, 0.16_80, 0.12_04, 0.12_45, 0.18_58, 0.13_38, 0.20_95]     )

         assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-3





