
'''simple docstring'''







import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
					import jieba

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowercase    :							Optional[int]      =					logging.get_logger(__name__)

lowercase    :							Optional[int]      =					{"""vocab_file""": """vocab.txt"""}

lowercase    :							Optional[Any]      =					{
    """vocab_file""": {
        """openbmb/cpm-ant-10b""": """https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt""",
    },
}

lowercase    :							List[Any]      =					{
    """openbmb/cpm-ant-10b""": 1024,
}







def 				SCREAMING_SNAKE_CASE__	(   __A	)       ->	Optional[Any]:
	_snake_case   						=		collections.OrderedDict()
	with open(_snake_case			,				'r'			,				encoding='utf-8'	) as reader:
		_snake_case   						=		reader.readlines()
	for index, token in enumerate(_snake_case	):
		_snake_case   						=		token.rstrip('\n'	)
		_snake_case   						=		index
	return vocab


class 						__UpperCAmelCase      (					__snake_case						):


							def __init__(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_="<unk>"    ,   lowerCAmelCase_=2_00  ):





								"""simple docstring"""






								_snake_case   						=		vocab
								_snake_case   						=		unk_token
								_snake_case   						=		max_input_chars_per_word




							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								_snake_case   						=		list(a_  )
								if len(a_  ) > self.max_input_chars_per_word:
									return [self.unk_token]

								_snake_case   						=		0
								_snake_case   						=		[]
								while start < len(a_  ):
									_snake_case   						=		len(a_  )
									_snake_case   						=		None
									while start < end:
										_snake_case   						=		''''''.join(chars[start:end]  )
										if substr in self.vocab:
											_snake_case   						=		substr
											break
										end -= 1
									if cur_substr is None:
										sub_tokens.append(self.unk_token  )
										start += 1
									else:
										sub_tokens.append(a_  )
										_snake_case   						=		end

								return sub_tokens


class 						__UpperCAmelCase      (					__snake_case						):
							__lowercase      		=       VOCAB_FILES_NAMES
							__lowercase      		=       PRETRAINED_VOCAB_FILES_MAP
							__lowercase      		=       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							__lowercase      		=       ["""input_ids""", """attention_mask"""]
							__lowercase      		=       False


							def __init__(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_="<d>"    ,   lowerCAmelCase_="</d>"    ,   lowerCAmelCase_="<s>"    ,   lowerCAmelCase_="</s>"    ,   lowerCAmelCase_="<pad>"    ,   lowerCAmelCase_="<unk>"    ,   lowerCAmelCase_="</n>"    ,   lowerCAmelCase_="</_>"    ,   lowerCAmelCase_="left"    ,   **lowerCAmelCase_    ,   ):





								"""simple docstring"""






								requires_backends(self    ,   ['jieba']  )
								super().__init__(
								    bod_token=a_    ,   eod_token=a_    ,   bos_token=a_    ,   eos_token=a_    ,   pad_token=a_    ,   unk_token=a_    ,   line_token=a_    ,   space_token=a_    ,   padding_side=a_    ,   **a_    ,   )
								_snake_case   						=		bod_token
								_snake_case   						=		eod_token
								_snake_case   						=		load_vocab(a_  )
								_snake_case   						=		self.encoder[space_token]
								_snake_case   						=		self.encoder[line_token]

								del self.encoder[space_token]
								del self.encoder[line_token]

								_snake_case   						=		collections.OrderedDict(sorted(self.encoder.items()    ,   key=lambda lowerCAmelCase_  : x[1]  )  )
								_snake_case   						=		{v: k for k, v in self.encoder.items()}

								_snake_case   						=		WordpieceTokenizer(vocab=self.encoder    ,   unk_token=self.unk_token  )


							@property
							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								return self.encoder[self.bod_token]


							@property
							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								return self.encoder[self.eod_token]


							@property
							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								return self.encoder["\n"]


							@property
							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								return len(self.encoder  )


							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								return dict(self.encoder    ,   **self.added_tokens_encoder  )


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								_snake_case   						=		[]
								for x in jieba.cut(a_    ,   cut_all=a_  ):
									output_tokens.extend(self.wordpiece_tokenizer.tokenize(a_  )  )
								return output_tokens


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   **lowerCAmelCase_  ):





								"""simple docstring"""






								_snake_case   						=		[i for i in token_ids if i >= 0]
								_snake_case   						=		[
								    x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
								]
								return super()._decode(a_    ,   **a_  )


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								return token in self.encoder


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								return "".join(a_  )


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								return self.encoder.get(a_    ,   self.encoder.get(self.unk_token  )  )


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								return self.decoder.get(a_    ,   self.unk_token  )


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_ = None  ):





								"""simple docstring"""






								if os.path.isdir(a_  ):
									_snake_case   						=		os.path.join(
									    a_    ,   (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']  )
								else:
									_snake_case   						=		(filename_prefix + '''-''' if filename_prefix else '''''') + save_directory
								_snake_case   						=		0
								if " " in self.encoder:
									_snake_case   						=		self.encoder[''' ''']
									del self.encoder[" "]
								if "\n" in self.encoder:
									_snake_case   						=		self.encoder['''\n''']
									del self.encoder["\n"]
								_snake_case   						=		collections.OrderedDict(sorted(self.encoder.items()    ,   key=lambda lowerCAmelCase_  : x[1]  )  )
								with open(a_    ,   'w'    ,   encoding='utf-8'  ) as writer:
									for token, token_index in self.encoder.items():
										if index != token_index:
											logger.warning(
											    F'Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive.'
											    ' Please check that the vocabulary is not corrupted!'  )
											_snake_case   						=		token_index
										writer.write(token + '\n'  )
										index += 1
								return (vocab_file,)


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_ = None  ):





								"""simple docstring"""






								if token_ids_a is None:
									return [self.bos_token_id] + token_ids_a
								return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a




							def  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_ = None    ,   lowerCAmelCase_ = False  ):





								"""simple docstring"""






								if already_has_special_tokens:
									return super().get_special_tokens_mask(
									    token_ids_a=a_    ,   token_ids_a=a_    ,   already_has_special_tokens=a_  )

								if token_ids_a is not None:
									return [1] + ([0] * len(a_  )) + [1] + ([0] * len(a_  ))
								return [1] + ([0] * len(a_  ))





"""simple docstring"""
import gc
import unittest

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

from diffusers import HeunDiscreteScheduler, PriorTransformer, ShapEPipeline
from diffusers.pipelines.shap_e import ShapERenderer
from diffusers.utils import load_numpy, slow
from diffusers.utils.testing_utils import require_torch_gpu, torch_device

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference



class 			_UpperCAmelCase (							__snake_case,				unittest.TestCase				):






		'''simple docstring'''

		lowerCamelCase__				=ShapEPipeline
		lowerCamelCase__				=['prompt']
		lowerCamelCase__				=['prompt']
		lowerCamelCase__				=[
		    'num_images_per_prompt',
		    'num_inference_steps',
		    'generator',
		    'latents',
		    'guidance_scale',
		    'frame_size',
		    'output_type',
		    'return_dict',
		]
		lowerCamelCase__				=False



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			return 32



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			return 32



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			return self.time_input_dim * 4



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			return 8



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			__snake_case	:						Optional[Any]						=				CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)
			return tokenizer



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			torch.manual_seed(0					)
			__snake_case	:						Optional[int]						=				CLIPTextConfig(
			    bos_token_id=0				,       eos_token_id=2				,       hidden_size=self.text_embedder_hidden_size				,       projection_dim=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 CLIPTextModelWithProjection(a_					)



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			torch.manual_seed(0					)

			__snake_case	:						Optional[int]						=				{
			    '''num_attention_heads''': 2,
			    '''attention_head_dim''': 16,
			    '''embedding_dim''': self.time_input_dim,
			    '''num_embeddings''': 32,
			    '''embedding_proj_dim''': self.text_embedder_hidden_size,
			    '''time_embed_dim''': self.time_embed_dim,
			    '''num_layers''': 1,
			    '''clip_embed_dim''': self.time_input_dim * 2,
			    '''additional_embeddings''': 0,
			    '''time_embed_act_fn''': '''gelu''',
			    '''norm_in_type''': '''layer''',
			    '''encoder_hid_proj_type''': None,
			    '''added_emb_type''': None,
			}

			__snake_case	:						Optional[int]						=				PriorTransformer(**a_					)
			return model



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			torch.manual_seed(0					)

			__snake_case	:						Optional[int]						=				{
			    '''param_shapes''': (
			        (self.renderer_dim, 93),
			        (self.renderer_dim, 8),
			        (self.renderer_dim, 8),
			        (self.renderer_dim, 8),
			    ),
			    '''d_latent''': self.time_input_dim,
			    '''d_hidden''': self.renderer_dim,
			    '''n_output''': 12,
			    '''background''': (
			        0.1,
			        0.1,
			        0.1,
			    ),
			}
			__snake_case	:						List[Any]						=				ShapERenderer(**a_					)
			return model



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			__snake_case	:						int						=				self.dummy_prior
			__snake_case	:						str						=				self.dummy_text_encoder
			__snake_case	:						str						=				self.dummy_tokenizer
			__snake_case	:						Tuple						=				self.dummy_renderer

			__snake_case	:						int						=				HeunDiscreteScheduler(
			    beta_schedule='''exp'''				,       num_train_timesteps=10_24				,       prediction_type='''sample'''				,       use_karras_sigmas=a_				,       clip_sample=a_				,       clip_sample_range=1.0				,       )
			__snake_case	:						Union[str, Any]						=				{
			    '''prior''': prior,
			    '''text_encoder''': text_encoder,
			    '''tokenizer''': tokenizer,
			    '''renderer''': renderer,
			    '''scheduler''': scheduler,
			}

			return components



		def 				SCREAMING_SNAKE_CASE  (self				,       a_				,       a_=0					):





			'''simple docstring'''

			if str(a_					).startswith('''mps'''					):
				__snake_case	:						Tuple						=				torch.manual_seed(a_					)
			else:
				__snake_case	:						Union[str, Any]						=				torch.Generator(device=a_					).manual_seed(a_					)
			__snake_case	:						Optional[int]						=				{
			    '''prompt''': '''horse''',
			    '''generator''': generator,
			    '''num_inference_steps''': 1,
			    '''frame_size''': 32,
			    '''output_type''': '''np''',
			}
			return inputs



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			__snake_case	:						Optional[int]						=				'''cpu'''

			__snake_case	:						str						=				self.get_dummy_components()

			__snake_case	:						List[Any]						=				self.pipeline_class(**a_					)
			__snake_case	:						str						=				pipe.to(a_					)

			pipe.set_progress_bar_config(disable=a_					)

			__snake_case	:						Optional[Any]						=				pipe(**self.get_dummy_inputs(a_					)					)
			__snake_case	:						List[str]						=				output.images[0]
			__snake_case	:						List[str]						=				image[0, -3:, -3:, -1]

			assert image.shape == (20, 32, 32, 3)

			__snake_case	:						List[Any]						=				np.array(
			    [
			        0.0003_9216,
			        0.0003_9216,
			        0.0003_9216,
			        0.0003_9216,
			        0.0003_9216,
			        0.0003_9216,
			        0.0003_9216,
			        0.0003_9216,
			        0.0003_9216,
			    ]					)

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



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			self._test_inference_batch_consistent(batch_sizes=[1, 2]					)



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			__snake_case	:						Any						=				torch_device == '''cpu'''
			__snake_case	:						Tuple						=				True

			self._test_inference_batch_single_identical(
			    batch_size=2				,       test_max_difference=a_				,       relax_max_difference=a_				,       )



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			__snake_case	:						int						=				self.get_dummy_components()
			__snake_case	:						int						=				self.pipeline_class(**a_					)
			__snake_case	:						int						=				pipe.to(a_					)
			pipe.set_progress_bar_config(disable=a_					)

			__snake_case	:						Optional[Any]						=				1
			__snake_case	:						List[Any]						=				2

			__snake_case	:						int						=				self.get_dummy_inputs(a_					)

			for key in inputs.keys():
				if key in self.batch_params:
					__snake_case	:						Dict						=				batch_size * [inputs[key]]

			__snake_case	:						Union[str, Any]						=				pipe(**a_				,       num_images_per_prompt=a_					)[0]

			assert images.shape[0] == batch_size * num_images_per_prompt



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






		'''simple docstring'''

		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			super().tearDown()
			gc.collect()
			torch.cuda.empty_cache()



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			__snake_case	:						List[str]						=				load_numpy(
			    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
			    '''/shap_e/test_shap_e_np_out.npy'''					)
			__snake_case	:						Optional[Any]						=				ShapEPipeline.from_pretrained('''openai/shap-e'''					)
			__snake_case	:						Optional[int]						=				pipe.to(a_					)
			pipe.set_progress_bar_config(disable=a_					)

			__snake_case	:						Optional[Any]						=				torch.Generator(device=a_					).manual_seed(0					)

			__snake_case	:						str						=				pipe(
			    '''a shark'''				,       generator=a_				,       guidance_scale=15.0				,       num_inference_steps=64				,       frame_size=64				,       output_type='''np'''				,       ).images[0]

			assert images.shape == (20, 64, 64, 3)

			assert_mean_pixel_difference(a_				,       a_					)





