'''simple docstring'''


import pytest

from datasets.parallel import ParallelBackendConfig, parallel_backend
from datasets.utils.py_utils import map_nested

from .utils import require_dill_gt_0_3_2, require_joblibspark, require_not_windows


def 						SCREAMING_SNAKE_CASE       (      a_	:							List[str]			):  # picklable for multiprocessing
  return i + 1


@require_dill_gt_0_3_2
@require_joblibspark
@require_not_windows
def 						SCREAMING_SNAKE_CASE       (      ):
  with parallel_backend('spark'			):
    assert ParallelBackendConfig.backend_name == "spark"

  __a			=					[1, 2, 3]
  with pytest.raises(a_			):
    with parallel_backend('unsupported backend'			):
      map_nested(a_				,					a_				,					num_proc=2			)

  with pytest.raises(a_			):
    with parallel_backend('unsupported backend'			):
      map_nested(a_				,					a_				,					num_proc=-1			)


@require_dill_gt_0_3_2
@require_joblibspark
@require_not_windows
@pytest.mark.parametrize('num_proc'				,					[2, -1]			)
def 						SCREAMING_SNAKE_CASE       (      a_	:							Union[str, Any]			):
  __a			=					[1, 2]
  __a			=					{'a': 1, 'b': 2}
  __a			=					{'a': [1, 2], 'b': [3, 4]}
  __a			=					{'a': {'1': 1}, 'b': 2}
  __a			=					{'a': 1, 'b': 2, 'c': 3, 'd': 4}
  __a			=					[2, 3]
  __a			=					{'a': 2, 'b': 3}
  __a			=					{'a': [2, 3], 'b': [4, 5]}
  __a			=					{'a': {'1': 2}, 'b': 3}
  __a			=					{'a': 2, 'b': 3, 'c': 4, 'd': 5}

  with parallel_backend('spark'			):
    assert map_nested(a_				,					a_				,					num_proc=a_			) == expected_map_nested_sa
    assert map_nested(a_				,					a_				,					num_proc=a_			) == expected_map_nested_sa
    assert map_nested(a_				,					a_				,					num_proc=a_			) == expected_map_nested_sa
    assert map_nested(a_				,					a_				,					num_proc=a_			) == expected_map_nested_sa
    assert map_nested(a_				,					a_				,					num_proc=a_			) == expected_map_nested_sa









'''simple docstring'''


import gc
import unittest

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

from diffusers import TransformeraDModel, VQDiffusionPipeline, VQDiffusionScheduler, VQModel
from diffusers.pipelines.vq_diffusion.pipeline_vq_diffusion import LearnedClassifierFreeSamplingEmbeddings
from diffusers.utils import load_numpy, slow, torch_device
from diffusers.utils.testing_utils import require_torch_gpu


UpperCAmelCase_ 		=      False





class    __lowercase						(  unittest.TestCase ):







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







   @property
   def 		UpperCamelCase__			(	self      )				->    Tuple:
     return 12







   @property
   def 		UpperCamelCase__			(	self      )				->    Optional[int]:
     return 12







   @property
   def 		UpperCamelCase__			(	self      )				->    List[str]:
     return 32







   @property
   def 		UpperCamelCase__			(	self      )				->    Optional[int]:
     torch.manual_seed(0      )
     __a			=					VQModel(
         block_out_channels=[32, 64]      ,							in_channels=3      ,							out_channels=3      ,							down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']      ,							up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']      ,							latent_channels=3      ,							num_vq_embeddings=self.num_embed      ,							vq_embed_dim=3      ,							)
     return model







   @property
   def 		UpperCamelCase__			(	self      )				->    Optional[int]:
     __a			=					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'      )
     return tokenizer







   @property
   def 		UpperCamelCase__			(	self      )				->    Tuple:
     torch.manual_seed(0      )
     __a			=					CLIPTextConfig(
         bos_token_id=0      ,							eos_token_id=2      ,							hidden_size=self.text_embedder_hidden_size      ,							intermediate_size=37      ,							layer_norm_eps=1e-05      ,							num_attention_heads=4      ,							num_hidden_layers=5      ,							pad_token_id=1      ,							vocab_size=1000      ,							)
     return CLIPTextModel(UpperCamelCase      )







   @property
   def 		UpperCamelCase__			(	self      )				->    Tuple:
     torch.manual_seed(0      )

     __a			=					12
     __a			=					12

     __a			=					{
         'attention_bias': True,
         'cross_attention_dim': 32,
         'attention_head_dim': height * width,
         'num_attention_heads': 1,
         'num_vector_embeds': self.num_embed,
         'num_embeds_ada_norm': self.num_embeds_ada_norm,
         'norm_num_groups': 32,
         'sample_size': width,
         'activation_fn': 'geglu-approximate',
     }

     __a			=					TransformeraDModel(**UpperCamelCase      )
     return model







   def 		UpperCamelCase__			(	self      )				->    Optional[Any]:
     __a			=					'cpu'

     __a			=					self.dummy_vqvae
     __a			=					self.dummy_text_encoder
     __a			=					self.dummy_tokenizer
     __a			=					self.dummy_transformer
     __a			=					VQDiffusionScheduler(self.num_embed      )
     __a			=					LearnedClassifierFreeSamplingEmbeddings(learnable=UpperCamelCase      )

     __a			=					VQDiffusionPipeline(
         vqvae=UpperCamelCase      ,							text_encoder=UpperCamelCase      ,							tokenizer=UpperCamelCase      ,							transformer=UpperCamelCase      ,							scheduler=UpperCamelCase      ,							learned_classifier_free_sampling_embeddings=UpperCamelCase      ,							)
     __a			=					pipe.to(UpperCamelCase      )
     pipe.set_progress_bar_config(disable=UpperCamelCase      )

     __a			=					'teddy bear playing in the pool'

     __a			=					torch.Generator(device=UpperCamelCase      ).manual_seed(0      )
     __a			=					pipe([prompt]      ,							generator=UpperCamelCase      ,							num_inference_steps=2      ,							output_type='np'      )
     __a			=					output.images

     __a			=					torch.Generator(device=UpperCamelCase      ).manual_seed(0      )
     __a			=					pipe(
         [prompt]      ,							generator=UpperCamelCase      ,							output_type='np'      ,							return_dict=UpperCamelCase      ,							num_inference_steps=2      )[0]

     __a			=					image[0, -3:, -3:, -1]
     __a			=					image_from_tuple[0, -3:, -3:, -1]

     assert image.shape == (1, 24, 24, 3)

     __a			=					np.array([0.6_551, 0.6_168, 0.5_008, 0.5_676, 0.5_659, 0.4_295, 0.6_073, 0.5_599, 0.4_992]      )

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







   def 		UpperCamelCase__			(	self      )				->    int:
     __a			=					'cpu'

     __a			=					self.dummy_vqvae
     __a			=					self.dummy_text_encoder
     __a			=					self.dummy_tokenizer
     __a			=					self.dummy_transformer
     __a			=					VQDiffusionScheduler(self.num_embed      )
     __a			=					LearnedClassifierFreeSamplingEmbeddings(
         learnable=UpperCamelCase      ,							hidden_size=self.text_embedder_hidden_size      ,							length=tokenizer.model_max_length      )

     __a			=					VQDiffusionPipeline(
         vqvae=UpperCamelCase      ,							text_encoder=UpperCamelCase      ,							tokenizer=UpperCamelCase      ,							transformer=UpperCamelCase      ,							scheduler=UpperCamelCase      ,							learned_classifier_free_sampling_embeddings=UpperCamelCase      ,							)
     __a			=					pipe.to(UpperCamelCase      )
     pipe.set_progress_bar_config(disable=UpperCamelCase      )

     __a			=					'teddy bear playing in the pool'

     __a			=					torch.Generator(device=UpperCamelCase      ).manual_seed(0      )
     __a			=					pipe([prompt]      ,							generator=UpperCamelCase      ,							num_inference_steps=2      ,							output_type='np'      )
     __a			=					output.images

     __a			=					torch.Generator(device=UpperCamelCase      ).manual_seed(0      )
     __a			=					pipe(
         [prompt]      ,							generator=UpperCamelCase      ,							output_type='np'      ,							return_dict=UpperCamelCase      ,							num_inference_steps=2      )[0]

     __a			=					image[0, -3:, -3:, -1]
     __a			=					image_from_tuple[0, -3:, -3:, -1]

     assert image.shape == (1, 24, 24, 3)

     __a			=					np.array([0.6_693, 0.6_075, 0.4_959, 0.5_701, 0.5_583, 0.4_333, 0.6_171, 0.5_684, 0.4_988]      )

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





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







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







   def 		UpperCamelCase__			(	self      )				->    Union[str, Any]:
     __a			=					load_numpy(
         'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
         '/vq_diffusion/teddy_bear_pool_classifier_free_sampling.npy'      )

     __a			=					VQDiffusionPipeline.from_pretrained('microsoft/vq-diffusion-ithq'      )
     __a			=					pipeline.to(UpperCamelCase      )
     pipeline.set_progress_bar_config(disable=UpperCamelCase      )

     # requires GPU generator for gumbel softmax
     # don't use GPU generator in tests though
     __a			=					torch.Generator(device=UpperCamelCase      ).manual_seed(0      )
     __a			=					pipeline(
         'teddy bear playing in the pool'      ,							num_images_per_prompt=1      ,							generator=UpperCamelCase      ,							output_type='np'      ,							)

     __a			=					output.images[0]

     assert image.shape == (256, 256, 3)
     assert np.abs(expected_image - image      ).max() < 2.0



