"""simple docstring"""







import torch
from torch import nn

class 		UpperCamelCase      (   nn.Module						):



				def __init__(self    :				List[str]     ,  _A    :				List[Any]     ,  _A    :				Tuple     ,  _A    :				Any     ,  _A    :				Any     ,  _A    :				str=1     ,  _A    :				Union[str, Any]=False)   ->     List[Any]:
					super().__init__()

					__snake_case      :							List[str]						=			n_token
					__snake_case      :							List[Any]						=			d_embed
					__snake_case      :							str						=			d_proj

					__snake_case      :							Optional[Any]						=			cutoffs + [n_token]
					__snake_case      :							Tuple						=			[0] + self.cutoffs
					__snake_case      :							int						=			div_val

					__snake_case      :							Any						=			self.cutoffs[0]
					__snake_case      :							List[str]						=			len(self.cutoffs) - 1
					__snake_case      :							Dict						=			self.shortlist_size + self.n_clusters

					if self.n_clusters > 0:
						__snake_case      :							Dict						=			nn.Parameter(torch.zeros(self.n_clusters     ,  self.d_embed))
						__snake_case      :							Dict						=			nn.Parameter(torch.zeros(self.n_clusters))

					__snake_case      :							Union[str, Any]						=			nn.ModuleList()
					__snake_case      :							str						=			nn.ParameterList()

					if div_val == 1:
						for i in range(len(self.cutoffs)):
							if d_proj != d_embed:
								self.out_projs.append(nn.Parameter(torch.FloatTensor(_A     ,  _A)))
							else:
								self.out_projs.append(_A)

						self.out_layers.append(nn.Linear(_A     ,  _A))
					else:
						for i in range(len(self.cutoffs)):
							__snake_case		,    __snake_case      :							List[str]						=			self.cutoff_ends[i], self.cutoff_ends[i + 1]
							__snake_case      :							List[Any]						=			d_embed // (div_val**i)

							self.out_projs.append(nn.Parameter(torch.FloatTensor(_A     ,  _A)))

							self.out_layers.append(nn.Linear(_A     ,  r_idx - l_idx))

					__snake_case      :							int						=			keep_order



				def  _lowercase					(self    :				Dict     ,  _A    :				Dict     ,  _A    :				Dict     ,  _A    :				Union[str, Any]     ,  _A    :				int)   ->     List[Any]:
					if proj is None:
						__snake_case      :							List[str]						=			nn.functional.linear(_A     ,  _A     ,  bias=_A)
					else:
						# if CUDA_MAJOR <= 9 and CUDA_MINOR <= 1:
						__snake_case      :							List[str]						=			nn.functional.linear(_A     ,  proj.t().contiguous())
						__snake_case      :							int						=			nn.functional.linear(_A     ,  _A     ,  bias=_A)
						# else:
						#     logit = torch.einsum('bd,de,ev->bv', (hidden, proj, weight.t()))
						#     if bias is not None:
						#         logit = logit + bias

					return logit



				def  _lowercase					(self    :				str     ,  _A    :				Any     ,  _A    :				Tuple=None     ,  _A    :				Optional[Any]=False)   ->     Optional[Any]:

					if labels is not None:
						# Shift so that tokens < n predict n
						__snake_case      :							int						=			hidden[..., :-1, :].contiguous()
						__snake_case      :							Union[str, Any]						=			labels[..., 1:].contiguous()
						__snake_case      :							Tuple						=			hidden.view(-1     ,  hidden.size(-1))
						__snake_case      :							Optional[int]						=			labels.view(-1)
						if hidden.size(0) != labels.size(0):
							raise RuntimeError('Input and labels should have the same size in the batch dimension.')
					else:
						__snake_case      :							Union[str, Any]						=			hidden.view(-1     ,  hidden.size(-1))

					if self.n_clusters == 0:
						__snake_case      :							Optional[Any]						=			self._compute_logit(_A     ,  self.out_layers[0].weight     ,  self.out_layers[0].bias     ,  self.out_projs[0])
						if labels is not None:
							__snake_case      :							Any						=			labels != -1_00
							__snake_case      :							Any						=			torch.zeros_like(_A     ,  dtype=hidden.dtype     ,  device=hidden.device)
							__snake_case      :							str						=			(
							    -nn.functional.log_softmax(_A     ,  dim=-1)[mask].gather(1     ,  labels[mask].unsqueeze(1)).squeeze(1)
							)
						else:
							__snake_case      :							Optional[Any]						=			nn.functional.log_softmax(_A     ,  dim=-1)
					else:
						# construct weights and biases
						__snake_case		,    __snake_case      :							Any						=			[], []
						for i in range(len(self.cutoffs)):
							if self.div_val == 1:
								__snake_case		,    __snake_case      :							str						=			self.cutoff_ends[i], self.cutoff_ends[i + 1]
								__snake_case      :							str						=			self.out_layers[0].weight[l_idx:r_idx]
								__snake_case      :							List[Any]						=			self.out_layers[0].bias[l_idx:r_idx]
							else:
								__snake_case      :							int						=			self.out_layers[i].weight
								__snake_case      :							List[Any]						=			self.out_layers[i].bias

							if i == 0:
								__snake_case      :							List[Any]						=			torch.cat([weight_i, self.cluster_weight]     ,  dim=0)
								__snake_case      :							Optional[Any]						=			torch.cat([bias_i, self.cluster_bias]     ,  dim=0)

							weights.append(_A)
							biases.append(_A)

						__snake_case		,    __snake_case		,    __snake_case      :							Tuple						=			weights[0], biases[0], self.out_projs[0]

						__snake_case      :							Optional[int]						=			self._compute_logit(_A     ,  _A     ,  _A     ,  _A)
						__snake_case      :							Tuple						=			nn.functional.log_softmax(_A     ,  dim=1)

						if labels is None:
							__snake_case      :							int						=			hidden.new_empty((head_logit.size(0), self.n_token))
						else:
							__snake_case      :							int						=			torch.zeros_like(_A     ,  dtype=hidden.dtype     ,  device=hidden.device)

						__snake_case      :							List[str]						=			0
						__snake_case      :							List[Any]						=			[0] + self.cutoffs
						for i in range(len(_A) - 1):
							__snake_case		,    __snake_case      :							Any						=			cutoff_values[i], cutoff_values[i + 1]

							if labels is not None:
								__snake_case      :							Tuple						=			(labels >= l_idx) & (labels < r_idx)
								__snake_case      :							Any						=			mask_i.nonzero().squeeze()

								if indices_i.numel() == 0:
									continue

								__snake_case      :							List[Any]						=			labels.index_select(0     ,  _A) - l_idx
								__snake_case      :							Optional[Any]						=			head_logprob.index_select(0     ,  _A)
								__snake_case      :							int						=			hidden.index_select(0     ,  _A)
							else:
								__snake_case      :							List[Any]						=			hidden

							if i == 0:
								if labels is not None:
									__snake_case      :							Optional[Any]						=			head_logprob_i.gather(1     ,  target_i[:, None]).squeeze(1)
								else:
									__snake_case      :							List[Any]						=			head_logprob[:, : self.cutoffs[0]]
							else:
								__snake_case		,    __snake_case		,    __snake_case      :							Dict						=			weights[i], biases[i], self.out_projs[i]

								__snake_case      :							Any						=			self._compute_logit(_A     ,  _A     ,  _A     ,  _A)
								__snake_case      :							List[Any]						=			nn.functional.log_softmax(_A     ,  dim=1)
								__snake_case      :							Optional[int]						=			self.cutoffs[0] + i - 1  # No probability for the head cluster
								if labels is not None:
									__snake_case      :							Any						=			head_logprob_i[:, cluster_prob_idx] + tail_logprob_i.gather(
									    1     ,  target_i[:, None]).squeeze(1)
								else:
									__snake_case      :							Tuple						=			head_logprob[:, cluster_prob_idx, None] + tail_logprob_i
									__snake_case      :							List[Any]						=			logprob_i

							if labels is not None:
								if (hasattr(self     ,  'keep_order') and self.keep_order) or keep_order:
									out.index_copy_(0     ,  _A     ,  -logprob_i)
								else:
									out[offset : offset + logprob_i.size(0)].copy_(-logprob_i)
								offset += logprob_i.size(0)

					return out






				def  _lowercase					(self    :				Optional[Any]     ,  _A    :				Dict)   ->     List[Any]:
					if self.n_clusters == 0:
						__snake_case      :							Dict						=			self._compute_logit(_A     ,  self.out_layers[0].weight     ,  self.out_layers[0].bias     ,  self.out_projs[0])
						return nn.functional.log_softmax(_A     ,  dim=-1)
					else:
						# construct weights and biases
						__snake_case		,    __snake_case      :							Any						=			[], []
						for i in range(len(self.cutoffs)):
							if self.div_val == 1:
								__snake_case		,    __snake_case      :							Any						=			self.cutoff_ends[i], self.cutoff_ends[i + 1]
								__snake_case      :							str						=			self.out_layers[0].weight[l_idx:r_idx]
								__snake_case      :							int						=			self.out_layers[0].bias[l_idx:r_idx]
							else:
								__snake_case      :							Any						=			self.out_layers[i].weight
								__snake_case      :							List[Any]						=			self.out_layers[i].bias

							if i == 0:
								__snake_case      :							int						=			torch.cat([weight_i, self.cluster_weight]     ,  dim=0)
								__snake_case      :							List[Any]						=			torch.cat([bias_i, self.cluster_bias]     ,  dim=0)

							weights.append(_A)
							biases.append(_A)

						__snake_case		,    __snake_case		,    __snake_case      :							Union[str, Any]						=			weights[0], biases[0], self.out_projs[0]
						__snake_case      :							Optional[Any]						=			self._compute_logit(_A     ,  _A     ,  _A     ,  _A)

						__snake_case      :							Union[str, Any]						=			hidden.new_empty((head_logit.size(0), self.n_token))
						__snake_case      :							int						=			nn.functional.log_softmax(_A     ,  dim=1)

						__snake_case      :							List[str]						=			[0] + self.cutoffs
						for i in range(len(_A) - 1):
							__snake_case		,    __snake_case      :							List[Any]						=			cutoff_values[i], cutoff_values[i + 1]

							if i == 0:
								__snake_case      :							List[Any]						=			head_logprob[:, : self.cutoffs[0]]
							else:
								__snake_case		,    __snake_case		,    __snake_case      :							Dict						=			weights[i], biases[i], self.out_projs[i]

								__snake_case      :							Optional[int]						=			self._compute_logit(_A     ,  _A     ,  _A     ,  _A)
								__snake_case      :							Optional[Any]						=			nn.functional.log_softmax(_A     ,  dim=1)

								__snake_case      :							List[str]						=			head_logprob[:, -i] + tail_logprob_i
								__snake_case      :							str						=			logprob_i

						return out






"""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


_a					:					List[str]=			False

class 		UpperCamelCase      (   unittest.TestCase						):



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



				@property
				def  _lowercase					(self    :				Optional[Any])   ->     Union[str, Any]:
					return 12



				@property
				def  _lowercase					(self    :				Dict)   ->     Union[str, Any]:
					return 12



				@property
				def  _lowercase					(self    :				int)   ->     Tuple:
					return 32



				@property
				def  _lowercase					(self    :				Optional[int])   ->     Dict:
					torch.manual_seed(0)
					__snake_case      :							Any						=			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  _lowercase					(self    :				List[Any])   ->     Optional[int]:
					__snake_case      :							Dict						=			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')
					return tokenizer



				@property
				def  _lowercase					(self    :				Union[str, Any])   ->     Optional[int]:
					torch.manual_seed(0)
					__snake_case      :							Any						=			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=10_00     ,  )
					return CLIPTextModel(_A)



				@property
				def  _lowercase					(self    :				Union[str, Any])   ->     Dict:
					torch.manual_seed(0)

					__snake_case      :							Any						=			12
					__snake_case      :							int						=			12

					__snake_case      :							List[Any]						=			{
					    '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',
					}

					__snake_case      :							Union[str, Any]						=			TransformeraDModel(**_A)
					return model



				def  _lowercase					(self    :				Union[str, Any])   ->     Dict:
					__snake_case      :							Tuple						=			'cpu'

					__snake_case      :							List[str]						=			self.dummy_vqvae
					__snake_case      :							str						=			self.dummy_text_encoder
					__snake_case      :							Optional[Any]						=			self.dummy_tokenizer
					__snake_case      :							Dict						=			self.dummy_transformer
					__snake_case      :							Optional[int]						=			VQDiffusionScheduler(self.num_embed)
					__snake_case      :							List[Any]						=			LearnedClassifierFreeSamplingEmbeddings(learnable=_A)

					__snake_case      :							List[Any]						=			VQDiffusionPipeline(
					    vqvae=_A     ,  text_encoder=_A     ,  tokenizer=_A     ,  transformer=_A     ,  scheduler=_A     ,  learned_classifier_free_sampling_embeddings=_A     ,  )
					__snake_case      :							List[Any]						=			pipe.to(_A)
					pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							Optional[Any]						=			'teddy bear playing in the pool'

					__snake_case      :							str						=			torch.Generator(device=_A).manual_seed(0)
					__snake_case      :							Union[str, Any]						=			pipe([prompt]     ,  generator=_A     ,  num_inference_steps=2     ,  output_type='np')
					__snake_case      :							Optional[int]						=			output.images

					__snake_case      :							int						=			torch.Generator(device=_A).manual_seed(0)
					__snake_case      :							Tuple						=			pipe(
					    [prompt]     ,  generator=_A     ,  output_type='np'     ,  return_dict=_A     ,  num_inference_steps=2)[0]

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

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

					__snake_case      :							str						=			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  _lowercase					(self    :				Tuple)   ->     Optional[int]:
					__snake_case      :							Optional[Any]						=			'cpu'

					__snake_case      :							Optional[int]						=			self.dummy_vqvae
					__snake_case      :							List[str]						=			self.dummy_text_encoder
					__snake_case      :							Optional[int]						=			self.dummy_tokenizer
					__snake_case      :							Optional[Any]						=			self.dummy_transformer
					__snake_case      :							Union[str, Any]						=			VQDiffusionScheduler(self.num_embed)
					__snake_case      :							Optional[int]						=			LearnedClassifierFreeSamplingEmbeddings(
					    learnable=_A     ,  hidden_size=self.text_embedder_hidden_size     ,  length=tokenizer.model_max_length)

					__snake_case      :							Union[str, Any]						=			VQDiffusionPipeline(
					    vqvae=_A     ,  text_encoder=_A     ,  tokenizer=_A     ,  transformer=_A     ,  scheduler=_A     ,  learned_classifier_free_sampling_embeddings=_A     ,  )
					__snake_case      :							Union[str, Any]						=			pipe.to(_A)
					pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							Union[str, Any]						=			'teddy bear playing in the pool'

					__snake_case      :							Optional[int]						=			torch.Generator(device=_A).manual_seed(0)
					__snake_case      :							Tuple						=			pipe([prompt]     ,  generator=_A     ,  num_inference_steps=2     ,  output_type='np')
					__snake_case      :							Optional[Any]						=			output.images

					__snake_case      :							str						=			torch.Generator(device=_A).manual_seed(0)
					__snake_case      :							Dict						=			pipe(
					    [prompt]     ,  generator=_A     ,  output_type='np'     ,  return_dict=_A     ,  num_inference_steps=2)[0]

					__snake_case      :							int						=			image[0, -3:, -3:, -1]
					__snake_case      :							Tuple						=			image_from_tuple[0, -3:, -3:, -1]

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

					__snake_case      :							Optional[Any]						=			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 		UpperCamelCase      (   unittest.TestCase						):



				def  _lowercase					(self    :				Any)   ->     Union[str, Any]:
					# clean up the VRAM after each test
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()






				def  _lowercase					(self    :				Tuple)   ->     Optional[int]:
					__snake_case      :							List[Any]						=			load_numpy(
					    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
					    '/vq_diffusion/teddy_bear_pool_classifier_free_sampling.npy')

					__snake_case      :							Union[str, Any]						=			VQDiffusionPipeline.from_pretrained('microsoft/vq-diffusion-ithq')
					__snake_case      :							Tuple						=			pipeline.to(_A)
					pipeline.set_progress_bar_config(disable=_A)

					# requires GPU generator for gumbel softmax
					# don't use GPU generator in tests though
					__snake_case      :							Optional[int]						=			torch.Generator(device=_A).manual_seed(0)
					__snake_case      :							int						=			pipeline(
					    'teddy bear playing in the pool'     ,  num_images_per_prompt=1     ,  generator=_A     ,  output_type='np'     ,  )

					__snake_case      :							int						=			output.images[0]

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





