



import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


UpperCAmelCase_									=							logging.get_logger(__name__)





class 		UpperCAmelCase						:
   SCREAMING_SNAKE_CASE__							=     None






@experimental
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Tuple     ,				_SCREAMING_SNAKE_CASE   :	List[str]     ,				_SCREAMING_SNAKE_CASE   :	Optional[int]     ,				_SCREAMING_SNAKE_CASE   :	List[Any]     ,				_SCREAMING_SNAKE_CASE   :	str     ,				_SCREAMING_SNAKE_CASE   :	int     ,				_SCREAMING_SNAKE_CASE   :	List[str]				)->str:
      if ParallelBackendConfig.backend_name is None:
            return _map_with_multiprocessing_pool(
                _SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE				)

      return _map_with_joblib(_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE				)
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Any     ,				_SCREAMING_SNAKE_CASE   :	Union[str, Any]     ,				_SCREAMING_SNAKE_CASE   :	Dict     ,				_SCREAMING_SNAKE_CASE   :	Tuple     ,				_SCREAMING_SNAKE_CASE   :	Tuple     ,				_SCREAMING_SNAKE_CASE   :	str     ,				_SCREAMING_SNAKE_CASE   :	str				)->Union[str, Any]:
      _lowerCAmelCase 			=  num_proc if num_proc <= len(_SCREAMING_SNAKE_CASE				) else len(_SCREAMING_SNAKE_CASE				)
      _lowerCAmelCase 			=  []  # We organize the splits ourselve (contiguous splits)
      for index in range(_SCREAMING_SNAKE_CASE				):
            _lowerCAmelCase 			=  len(_SCREAMING_SNAKE_CASE				) // num_proc
            _lowerCAmelCase 			=  len(_SCREAMING_SNAKE_CASE				) % num_proc
            _lowerCAmelCase 			=  div * index + min(_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE				)
            _lowerCAmelCase 			=  start + div + (1 if index < mod else 0)
            split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)				)

      if len(_SCREAMING_SNAKE_CASE				) != sum(len(i[1]				) for i in split_kwds				):
            raise ValueError(
                f'''Error dividing inputs iterable among processes. '''
                f'''Total number of objects {len(_SCREAMING_SNAKE_CASE				)}, '''
                f'''length: {sum(len(i[1]				) for i in split_kwds				)}'''				)

      logger.info(
          f'''Spawning {num_proc} processes for {len(_SCREAMING_SNAKE_CASE				)} objects in slices of {[len(i[1]				) for i in split_kwds]}'''				)
      _lowerCAmelCase			,   _lowerCAmelCase 			=  None, None
      if not disable_tqdm:
            _lowerCAmelCase			,   _lowerCAmelCase 			=  (RLock(),), tqdm.set_lock
      with Pool(_SCREAMING_SNAKE_CASE     ,				initargs=_SCREAMING_SNAKE_CASE     ,				initializer=_SCREAMING_SNAKE_CASE				) as pool:
            _lowerCAmelCase 			=  pool.map(_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE				)
      logger.info(f'''Finished {num_proc} processes'''				)
      _lowerCAmelCase 			=  [obj for proc_res in mapped for obj in proc_res]
      logger.info(f'''Unpacked {len(_SCREAMING_SNAKE_CASE				)} objects'''				)

      return mapped
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	List[Any]     ,				_SCREAMING_SNAKE_CASE   :	str     ,				_SCREAMING_SNAKE_CASE   :	Optional[int]     ,				_SCREAMING_SNAKE_CASE   :	int     ,				_SCREAMING_SNAKE_CASE   :	Optional[Any]     ,				_SCREAMING_SNAKE_CASE   :	Any     ,				_SCREAMING_SNAKE_CASE   :	List[Any]				)->Optional[Any]:
      # progress bar is not yet supported for _map_with_joblib, because tqdm couldn't accurately be applied to joblib,
      # and it requires monkey-patching joblib internal classes which is subject to change
      import joblib

      with joblib.parallel_backend(ParallelBackendConfig.backend_name     ,				n_jobs=_SCREAMING_SNAKE_CASE				):
            return joblib.Parallel()(
                joblib.delayed(_SCREAMING_SNAKE_CASE				)((function, obj, types, None, True, None)				) for obj in iterable				)


@experimental
@contextlib.contextmanager
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	str				)->List[Any]:
      _lowerCAmelCase 			=  backend_name

      if backend_name == "spark":
            from joblibspark import register_spark

            register_spark()

            # TODO: call create_cache_and_write_probe if "download" in steps
            # TODO: raise NotImplementedError when Dataset.map etc is called

      try:
            yield
      finally:
            _lowerCAmelCase 			=  None




import gc
import unittest

import numpy as np
import torch

from diffusers import AutoencoderKL, DDIMScheduler, DiTPipeline, DPMSolverMultistepScheduler, TransformeraDModel
from diffusers.utils import is_xformers_available, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS,
    CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()





class 		UpperCAmelCase						(						snake_case_       ,unittest.TestCase				):
   SCREAMING_SNAKE_CASE__							=     DiTPipeline
   SCREAMING_SNAKE_CASE__							=     CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS
   SCREAMING_SNAKE_CASE__							=     PipelineTesterMixin.required_optional_params - {
       '''latents''',
       '''num_images_per_prompt''',
       '''callback''',
       '''callback_steps''',
   }
   SCREAMING_SNAKE_CASE__							=     CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS
   SCREAMING_SNAKE_CASE__							=     False





   def 	__lowerCAmelCase							(	self			):
         torch.manual_seed(0			)
         _lowerCAmelCase 			=  TransformeraDModel(
             sample_size=16       ,     num_layers=2       ,     patch_size=4       ,     attention_head_dim=8       ,     num_attention_heads=2       ,     in_channels=4       ,     out_channels=8       ,     attention_bias=_lowerCAmelCase       ,     activation_fn='''gelu-approximate'''       ,     num_embeds_ada_norm=1_000       ,     norm_type='''ada_norm_zero'''       ,     norm_elementwise_affine=_lowerCAmelCase       ,     )
         _lowerCAmelCase 			=  AutoencoderKL()
         _lowerCAmelCase 			=  DDIMScheduler()
         _lowerCAmelCase 			=  {'''transformer''': transformer.eval(), '''vae''': vae.eval(), '''scheduler''': scheduler}
         return components





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase=0			):
         if str(_lowerCAmelCase			).startswith('''mps'''			):
               _lowerCAmelCase 			=  torch.manual_seed(_lowerCAmelCase			)
         else:
               _lowerCAmelCase 			=  torch.Generator(device=_lowerCAmelCase			).manual_seed(_lowerCAmelCase			)
         _lowerCAmelCase 			=  {
             '''class_labels''': [1],
             '''generator''': generator,
             '''num_inference_steps''': 2,
             '''output_type''': '''numpy''',
         }
         return inputs





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  '''cpu'''

         _lowerCAmelCase 			=  self.get_dummy_components()
         _lowerCAmelCase 			=  self.pipeline_class(**_lowerCAmelCase			)
         pipe.to(_lowerCAmelCase			)
         pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

         _lowerCAmelCase 			=  self.get_dummy_inputs(_lowerCAmelCase			)
         _lowerCAmelCase 			=  pipe(**_lowerCAmelCase			).images
         _lowerCAmelCase 			=  image[0, -3:, -3:, -1]

         self.assertEqual(image.shape       ,     (1, 16, 16, 3)			)
         _lowerCAmelCase 			=  np.array([0.2_946, 0.6_601, 0.4_329, 0.3_296, 0.4_144, 0.5_319, 0.7_273, 0.5_013, 0.4_457]			)
         _lowerCAmelCase 			=  np.abs(image_slice.flatten() - expected_slice			).max()
         self.assertLessEqual(_lowerCAmelCase       ,     1E-3			)





   def 	__lowerCAmelCase							(	self			):
         self._test_inference_batch_single_identical(relax_max_difference=_lowerCAmelCase       ,     expected_max_diff=1E-3			)





   @unittest.skipIf(
       torch_device != '''cuda''' or not is_xformers_available()       ,     reason='''XFormers attention is only available with CUDA and `xformers` installed'''       ,     )
   def 	__lowerCAmelCase							(	self			):
         self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3			)






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





   def 	__lowerCAmelCase							(	self			):
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  torch.manual_seed(0			)

         _lowerCAmelCase 			=  DiTPipeline.from_pretrained('''facebook/DiT-XL-2-256'''			)
         pipe.to('''cuda'''			)

         _lowerCAmelCase 			=  ['''vase''', '''umbrella''', '''white shark''', '''white wolf''']
         _lowerCAmelCase 			=  pipe.get_label_ids(_lowerCAmelCase			)

         _lowerCAmelCase 			=  pipe(_lowerCAmelCase       ,     generator=_lowerCAmelCase       ,     num_inference_steps=40       ,     output_type='''np'''			).images

         for word, image in zip(_lowerCAmelCase       ,     _lowerCAmelCase			):
               _lowerCAmelCase 			=  load_numpy(
                   F'''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/dit/{word}.npy'''			)
               assert np.abs((expected_image - image).max()			) < 1E-2





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  DiTPipeline.from_pretrained('''facebook/DiT-XL-2-512'''			)
         _lowerCAmelCase 			=  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config			)
         pipe.to('''cuda'''			)

         _lowerCAmelCase 			=  ['''vase''', '''umbrella''']
         _lowerCAmelCase 			=  pipe.get_label_ids(_lowerCAmelCase			)

         _lowerCAmelCase 			=  torch.manual_seed(0			)
         _lowerCAmelCase 			=  pipe(_lowerCAmelCase       ,     generator=_lowerCAmelCase       ,     num_inference_steps=25       ,     output_type='''np'''			).images

         for word, image in zip(_lowerCAmelCase       ,     _lowerCAmelCase			):
               _lowerCAmelCase 			=  load_numpy(
                   '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
                   F'''/dit/{word}_512.npy'''			)

               assert np.abs((expected_image - image).max()			) < 1E-1