


import json
import os
from typing import Optional

import numpy as np

from ...feature_extraction_utils import BatchFeature
from ...processing_utils import ProcessorMixin
from ...utils import logging
from ...utils.hub import get_file_from_repo
from ..auto import AutoTokenizer


__snake_case            =   logging.get_logger(__name__)




class    UpperCAmelCase					(       __snake_case    ):
			lowercase						= """AutoTokenizer"""
			lowercase						= ["""tokenizer"""]

			lowercase						= {
			    """semantic_prompt""": 1,
			    """coarse_prompt""": 2,
			    """fine_prompt""": 2,
			}




			def __init__(						self							:		Dict       ,       __magic_name__							:		Optional[Any]       ,       __magic_name__							:		Dict=None						):







						"""simple docstring"""







						super().__init__(__magic_name__						)

						UpperCamelCase         =					speaker_embeddings




			@classmethod
			def   lowerCamelCase_					(						cls							:		List[Any]       ,       __magic_name__							:		List[Any]       ,       __magic_name__							:		List[Any]="speaker_embeddings_path.json"       ,       **__magic_name__							:		Optional[int]						):







						"""simple docstring"""







						if speaker_embeddings_dict_path is not None:
									UpperCamelCase         =					get_file_from_repo(
									    __magic_name__       ,       __magic_name__       ,       subfolder=kwargs.pop("""subfolder"""       ,       __magic_name__						)       ,       cache_dir=kwargs.pop("""cache_dir"""       ,       __magic_name__						)       ,       force_download=kwargs.pop("""force_download"""       ,       __magic_name__						)       ,       proxies=kwargs.pop("""proxies"""       ,       __magic_name__						)       ,       resume_download=kwargs.pop("""resume_download"""       ,       __magic_name__						)       ,       local_files_only=kwargs.pop("""local_files_only"""       ,       __magic_name__						)       ,       use_auth_token=kwargs.pop("""use_auth_token"""       ,       __magic_name__						)       ,       revision=kwargs.pop("""revision"""       ,       __magic_name__						)       ,       )
									if speaker_embeddings_path is None:
												logger.warning(
												    F'`{os.path.join(__magic_name__       ,       __magic_name__						)}` does not exists\n                    , no preloaded speaker embeddings will be used - Make sure to provide a correct path to the json\n                    dictionnary if wanted, otherwise set `speaker_embeddings_dict_path=None`.'						)
												UpperCamelCase         =					None
									else:
												with open(__magic_name__						) as speaker_embeddings_json:
															UpperCamelCase         =					json.load(__magic_name__						)
						else:
									UpperCamelCase         =					None

						UpperCamelCase         =					AutoTokenizer.from_pretrained(__magic_name__       ,       **__magic_name__						)

						return cls(tokenizer=__magic_name__       ,       speaker_embeddings=__magic_name__						)




			def   lowerCamelCase_					(						self							:		Dict       ,       __magic_name__							:		Union[str, Any]       ,       __magic_name__							:		int="speaker_embeddings_path.json"       ,       __magic_name__							:		List[str]="speaker_embeddings"       ,       __magic_name__							:		bool = False       ,       **__magic_name__							:		Optional[Any]       ,       ):







						"""simple docstring"""







						if self.speaker_embeddings is not None:
									os.makedirs(os.path.join(__magic_name__       ,       __magic_name__       ,       """v2"""						)       ,       exist_ok=__magic_name__						)

									UpperCamelCase         =					{}

									UpperCamelCase         =					save_directory

									for prompt_key in self.speaker_embeddings:
												if prompt_key != "repo_or_path":
															UpperCamelCase         =					self._load_voice_preset(__magic_name__						)

															UpperCamelCase         =					{}
															for key in self.speaker_embeddings[prompt_key]:
																		np.save(
																		    os.path.join(
																		        embeddings_dict["""repo_or_path"""]       ,       __magic_name__       ,       F'{prompt_key}_{key}'						)       ,       voice_preset[key]       ,       allow_pickle=__magic_name__       ,       )
																		UpperCamelCase         =					os.path.join(__magic_name__       ,       F'{prompt_key}_{key}.npy'						)

															UpperCamelCase         =					tmp_dict

									with open(os.path.join(__magic_name__       ,       __magic_name__						)       ,       """w"""						) as fp:
												json.dump(__magic_name__       ,       __magic_name__						)

						super().save_pretrained(__magic_name__       ,       __magic_name__       ,       **__magic_name__						)




			def   lowerCamelCase_					(						self							:		int       ,       __magic_name__							:		str = None       ,       **__magic_name__							:		str						):







						"""simple docstring"""







						UpperCamelCase         =					self.speaker_embeddings[voice_preset]

						UpperCamelCase         =					{}
						for key in ["semantic_prompt", "coarse_prompt", "fine_prompt"]:
									if key not in voice_preset_paths:
												raise ValueError(
												    F'Voice preset unrecognized, missing {key} as a key in self.speaker_embeddings[{voice_preset}].'						)

									UpperCamelCase         =					get_file_from_repo(
									    self.speaker_embeddings.get("""repo_or_path"""       ,       """/"""						)       ,       voice_preset_paths[key]       ,       subfolder=kwargs.pop("""subfolder"""       ,       __magic_name__						)       ,       cache_dir=kwargs.pop("""cache_dir"""       ,       __magic_name__						)       ,       force_download=kwargs.pop("""force_download"""       ,       __magic_name__						)       ,       proxies=kwargs.pop("""proxies"""       ,       __magic_name__						)       ,       resume_download=kwargs.pop("""resume_download"""       ,       __magic_name__						)       ,       local_files_only=kwargs.pop("""local_files_only"""       ,       __magic_name__						)       ,       use_auth_token=kwargs.pop("""use_auth_token"""       ,       __magic_name__						)       ,       revision=kwargs.pop("""revision"""       ,       __magic_name__						)       ,       )
									if path is None:
												raise ValueError(
												    F'`{os.path.join(self.speaker_embeddings.get("repo_or_path"       ,       "/"						)       ,       voice_preset_paths[key]						)}` does not exists\n                    , no preloaded voice preset will be used - Make sure to provide correct paths to the {voice_preset}\n                    embeddings.'						)

									UpperCamelCase         =					np.load(__magic_name__						)

						return voice_preset_dict




			def   lowerCamelCase_					(						self							:		str       ,       __magic_name__							:		Optional[dict] = None						):







						"""simple docstring"""







						for key in ["semantic_prompt", "coarse_prompt", "fine_prompt"]:
									if key not in voice_preset:
												raise ValueError(F'Voice preset unrecognized, missing {key} as a key.'						)

									if not isinstance(voice_preset[key]       ,       np.ndarray						):
												raise ValueError(F'{key} voice preset must be a {str(self.preset_shape[key]						)}D ndarray.'						)

									if len(voice_preset[key].shape						) != self.preset_shape[key]:
												raise ValueError(F'{key} voice preset must be a {str(self.preset_shape[key]						)}D ndarray.'						)





			def __call__(						self							:		Optional[int]       ,       __magic_name__							:		Tuple=None       ,       __magic_name__							:		Optional[Any]=None       ,       __magic_name__							:		str="pt"       ,       __magic_name__							:		str=2_5_6       ,       __magic_name__							:		Optional[Any]=False       ,       __magic_name__							:		Dict=True       ,       __magic_name__							:		Union[str, Any]=False       ,       **__magic_name__							:		Dict       ,       ):







						"""simple docstring"""







						if voice_preset is not None and not isinstance(__magic_name__       ,       __magic_name__						):
									if (
									    isinstance(__magic_name__       ,       __magic_name__						)
									    and self.speaker_embeddings is not None
									    and voice_preset in self.speaker_embeddings
									):
												UpperCamelCase         =					self._load_voice_preset(__magic_name__						)

									else:
												if isinstance(__magic_name__       ,       __magic_name__						) and not voice_preset.endswith(""".npz"""						):
															UpperCamelCase         =					voice_preset + """.npz"""

												UpperCamelCase         =					np.load(__magic_name__						)

						if voice_preset is not None:
									self._validate_voice_preset_dict(__magic_name__       ,       **__magic_name__						)
									UpperCamelCase         =					BatchFeature(data=__magic_name__       ,       tensor_type=__magic_name__						)

						UpperCamelCase         =					self.tokenizer(
						    __magic_name__       ,       return_tensors=__magic_name__       ,       padding="""max_length"""       ,       max_length=__magic_name__       ,       return_attention_mask=__magic_name__       ,       return_token_type_ids=__magic_name__       ,       add_special_tokens=__magic_name__       ,       **__magic_name__       ,       )

						if voice_preset is not None:
									UpperCamelCase         =					voice_preset

						return encoded_text








import unittest

from transformers import BertGenerationConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
			import torch

			from transformers import BertGenerationDecoder, BertGenerationEncoder




class    UpperCAmelCase					:




			def __init__(						self							:		Union[str, Any]       ,       __magic_name__							:		List[Any]       ,       __magic_name__							:		str=1_3       ,       __magic_name__							:		str=7       ,       __magic_name__							:		Optional[Any]=True       ,       __magic_name__							:		Optional[Any]=True       ,       __magic_name__							:		Tuple=9_9       ,       __magic_name__							:		List[str]=3_2       ,       __magic_name__							:		List[str]=5       ,       __magic_name__							:		int=4       ,       __magic_name__							:		Union[str, Any]=3_7       ,       __magic_name__							:		List[Any]="gelu"       ,       __magic_name__							:		Any=0.1       ,       __magic_name__							:		Union[str, Any]=0.1       ,       __magic_name__							:		List[Any]=5_0       ,       __magic_name__							:		Tuple=0.02       ,       __magic_name__							:		Optional[int]=True       ,       __magic_name__							:		Optional[Any]=None       ,       ):







						"""simple docstring"""







						UpperCamelCase         =					parent
						UpperCamelCase         =					batch_size
						UpperCamelCase         =					seq_length
						UpperCamelCase         =					is_training
						UpperCamelCase         =					use_input_mask
						UpperCamelCase         =					vocab_size
						UpperCamelCase         =					hidden_size
						UpperCamelCase         =					num_hidden_layers
						UpperCamelCase         =					num_attention_heads
						UpperCamelCase         =					intermediate_size
						UpperCamelCase         =					hidden_act
						UpperCamelCase         =					hidden_dropout_prob
						UpperCamelCase         =					attention_probs_dropout_prob
						UpperCamelCase         =					max_position_embeddings
						UpperCamelCase         =					initializer_range
						UpperCamelCase         =					use_labels
						UpperCamelCase         =					scope




			def   lowerCamelCase_					(						self							:		Optional[int]						):







						"""simple docstring"""







						UpperCamelCase         =					ids_tensor([self.batch_size, self.seq_length]       ,       self.vocab_size						)

						UpperCamelCase         =					None
						if self.use_input_mask:
									UpperCamelCase         =					random_attention_mask([self.batch_size, self.seq_length]						)

						if self.use_labels:
									UpperCamelCase         =					ids_tensor([self.batch_size, self.seq_length]       ,       self.vocab_size						)

						UpperCamelCase         =					self.get_config()

						return config, input_ids, input_mask, token_labels




			def   lowerCamelCase_					(						self							:		str						):







						"""simple docstring"""







						return BertGenerationConfig(
						    vocab_size=self.vocab_size       ,       hidden_size=self.hidden_size       ,       num_hidden_layers=self.num_hidden_layers       ,       num_attention_heads=self.num_attention_heads       ,       intermediate_size=self.intermediate_size       ,       hidden_act=self.hidden_act       ,       hidden_dropout_prob=self.hidden_dropout_prob       ,       attention_probs_dropout_prob=self.attention_probs_dropout_prob       ,       max_position_embeddings=self.max_position_embeddings       ,       is_decoder=__magic_name__       ,       initializer_range=self.initializer_range       ,       )




			def   lowerCamelCase_					(						self							:		Union[str, Any]						):







						"""simple docstring"""







						(
						    (
						    UpperCamelCase
						)  ,  (
						    UpperCamelCase
						)  ,  (
						    UpperCamelCase
						)  ,  (
						    UpperCamelCase
						)  ,  
						)         =					self.prepare_config_and_inputs()

						UpperCamelCase         =					True
						UpperCamelCase         =					floats_tensor([self.batch_size, self.seq_length, self.hidden_size]						)
						UpperCamelCase         =					ids_tensor([self.batch_size, self.seq_length]       ,       vocab_size=2						)

						return (
						    config,
						    input_ids,
						    input_mask,
						    token_labels,
						    encoder_hidden_states,
						    encoder_attention_mask,
						)




			def   lowerCamelCase_					(						self							:		Tuple       ,       __magic_name__							:		List[Any]       ,       __magic_name__							:		Tuple       ,       __magic_name__							:		str       ,       __magic_name__							:		List[str]       ,       **__magic_name__							:		Optional[Any]       ,       ):







						"""simple docstring"""







						UpperCamelCase         =					BertGenerationEncoder(config=__magic_name__						)
						model.to(__magic_name__						)
						model.eval()
						UpperCamelCase         =					model(__magic_name__       ,       attention_mask=__magic_name__						)
						UpperCamelCase         =					model(__magic_name__						)
						self.parent.assertEqual(result.last_hidden_state.shape       ,       (self.batch_size, self.seq_length, self.hidden_size)						)




			def   lowerCamelCase_					(						self							:		Optional[int]       ,       __magic_name__							:		Dict       ,       __magic_name__							:		str       ,       __magic_name__							:		Union[str, Any]       ,       __magic_name__							:		List[str]       ,       __magic_name__							:		Optional[Any]       ,       __magic_name__							:		Tuple       ,       **__magic_name__							:		Union[str, Any]       ,       ):







						"""simple docstring"""







						UpperCamelCase         =					True
						UpperCamelCase         =					BertGenerationEncoder(config=__magic_name__						)
						model.to(__magic_name__						)
						model.eval()
						UpperCamelCase         =					model(
						    __magic_name__       ,       attention_mask=__magic_name__       ,       encoder_hidden_states=__magic_name__       ,       encoder_attention_mask=__magic_name__       ,       )
						UpperCamelCase         =					model(
						    __magic_name__       ,       attention_mask=__magic_name__       ,       encoder_hidden_states=__magic_name__       ,       )
						self.parent.assertEqual(result.last_hidden_state.shape       ,       (self.batch_size, self.seq_length, self.hidden_size)						)




			def   lowerCamelCase_					(						self							:		Optional[int]       ,       __magic_name__							:		Optional[Any]       ,       __magic_name__							:		str       ,       __magic_name__							:		List[str]       ,       __magic_name__							:		Optional[int]       ,       __magic_name__							:		str       ,       __magic_name__							:		Union[str, Any]       ,       **__magic_name__							:		Dict       ,       ):







						"""simple docstring"""







						UpperCamelCase         =					True
						UpperCamelCase         =					True
						UpperCamelCase         =					BertGenerationDecoder(config=__magic_name__						).to(__magic_name__						).eval()

						# first forward pass
						UpperCamelCase         =					model(
						    __magic_name__       ,       attention_mask=__magic_name__       ,       encoder_hidden_states=__magic_name__       ,       encoder_attention_mask=__magic_name__       ,       use_cache=__magic_name__       ,       )
						UpperCamelCase         =					outputs.past_key_values

						# create hypothetical multiple next token and extent to next_input_ids
						UpperCamelCase         =					ids_tensor((self.batch_size, 3)       ,       config.vocab_size						)
						UpperCamelCase         =					ids_tensor((self.batch_size, 3)       ,       vocab_size=2						)

						# append to next input_ids and
						UpperCamelCase         =					torch.cat([input_ids, next_tokens]       ,       dim=-1						)
						UpperCamelCase         =					torch.cat([input_mask, next_mask]       ,       dim=-1						)

						UpperCamelCase         =					model(
						    __magic_name__       ,       attention_mask=__magic_name__       ,       encoder_hidden_states=__magic_name__       ,       encoder_attention_mask=__magic_name__       ,       output_hidden_states=__magic_name__       ,       )["""hidden_states"""][0]
						UpperCamelCase         =					model(
						    __magic_name__       ,       attention_mask=__magic_name__       ,       encoder_hidden_states=__magic_name__       ,       encoder_attention_mask=__magic_name__       ,       past_key_values=__magic_name__       ,       output_hidden_states=__magic_name__       ,       )["""hidden_states"""][0]

						# select random slice
						UpperCamelCase         =					ids_tensor((1,)       ,       output_from_past.shape[-1]						).item()
						UpperCamelCase         =					output_from_no_past[:, -3:, random_slice_idx].detach()
						UpperCamelCase         =					output_from_past[:, :, random_slice_idx].detach()

						self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]						)

						# test that outputs are equal for slice
						self.parent.assertTrue(torch.allclose(__magic_name__       ,       __magic_name__       ,       atol=1e-3						)						)




			def   lowerCamelCase_					(						self							:		int       ,       __magic_name__							:		Any       ,       __magic_name__							:		List[str]       ,       __magic_name__							:		Union[str, Any]       ,       __magic_name__							:		List[Any]       ,       *__magic_name__							:		Optional[Any]       ,       ):







						"""simple docstring"""







						UpperCamelCase         =					BertGenerationDecoder(__magic_name__						)
						model.to(__magic_name__						)
						model.eval()
						UpperCamelCase         =					model(__magic_name__       ,       attention_mask=__magic_name__       ,       labels=__magic_name__						)
						self.parent.assertEqual(result.logits.shape       ,       (self.batch_size, self.seq_length, self.vocab_size)						)





			def   lowerCamelCase_					(						self							:		List[Any]						):







						"""simple docstring"""







						UpperCamelCase  ,  UpperCamelCase  ,  UpperCamelCase  ,  UpperCamelCase         =					self.prepare_config_and_inputs()
						UpperCamelCase         =					{"""input_ids""": input_ids, """attention_mask""": input_mask}
						return config, inputs_dict




@require_torch
class    UpperCAmelCase					(       __snake_case      , __snake_case      , __snake_case      , unittest.TestCase    ):
			lowercase						= (BertGenerationEncoder, BertGenerationDecoder) if is_torch_available() else ()
			lowercase						= (BertGenerationDecoder,) if is_torch_available() else ()
			lowercase						= (
			    {"""feature-extraction""": BertGenerationEncoder, """text-generation""": BertGenerationDecoder}
			    if is_torch_available()
			    else {}
			)




			def   lowerCamelCase_					(						self							:		Tuple						):







						"""simple docstring"""







						UpperCamelCase         =					BertGenerationEncoderTester(self						)
						UpperCamelCase         =					ConfigTester(self       ,       config_class=__magic_name__       ,       hidden_size=3_7						)




			def   lowerCamelCase_					(						self							:		Optional[int]						):







						"""simple docstring"""







						self.config_tester.run_common_tests()




			def   lowerCamelCase_					(						self							:		Any						):







						"""simple docstring"""







						UpperCamelCase         =					self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_model(*__magic_name__						)




			def   lowerCamelCase_					(						self							:		List[str]						):







						"""simple docstring"""







						UpperCamelCase  ,  UpperCamelCase  ,  UpperCamelCase  ,  UpperCamelCase         =					self.model_tester.prepare_config_and_inputs()
						UpperCamelCase         =					"""bert"""
						self.model_tester.create_and_check_model(__magic_name__       ,       __magic_name__       ,       __magic_name__       ,       __magic_name__						)




			def   lowerCamelCase_					(						self							:		Tuple						):







						"""simple docstring"""







						UpperCamelCase         =					self.model_tester.prepare_config_and_inputs_for_decoder()
						self.model_tester.create_and_check_model_as_decoder(*__magic_name__						)




			def   lowerCamelCase_					(						self							:		List[Any]						):







						"""simple docstring"""







						UpperCamelCase         =					self.model_tester.prepare_config_and_inputs_for_decoder()
						self.model_tester.create_and_check_decoder_model_past_large_inputs(*__magic_name__						)




			def   lowerCamelCase_					(						self							:		Dict						):







						"""simple docstring"""







						(
						    (
						    UpperCamelCase
						)  ,  (
						    UpperCamelCase
						)  ,  (
						    UpperCamelCase
						)  ,  (
						    UpperCamelCase
						)  ,  (
						    UpperCamelCase
						)  ,  (
						    UpperCamelCase
						)  ,  
						)         =					self.model_tester.prepare_config_and_inputs_for_decoder()

						UpperCamelCase         =					None

						self.model_tester.create_and_check_model_as_decoder(
						    __magic_name__       ,       __magic_name__       ,       __magic_name__       ,       __magic_name__       ,       __magic_name__       ,       __magic_name__       ,       )




			def   lowerCamelCase_					(						self							:		str						):







						"""simple docstring"""







						UpperCamelCase         =					self.model_tester.prepare_config_and_inputs_for_decoder()
						self.model_tester.create_and_check_for_causal_lm(*__magic_name__						)





			@slow
			def   lowerCamelCase_					(						self							:		List[Any]						):







						"""simple docstring"""







						UpperCamelCase         =					BertGenerationEncoder.from_pretrained("""google/bert_for_seq_generation_L-24_bbc_encoder"""						)
						self.assertIsNotNone(__magic_name__						)




@require_torch
class    UpperCAmelCase					(       unittest.TestCase    ):




			@slow
			def   lowerCamelCase_					(						self							:		Tuple						):







						"""simple docstring"""







						UpperCamelCase         =					BertGenerationEncoder.from_pretrained("""google/bert_for_seq_generation_L-24_bbc_encoder"""						)
						UpperCamelCase         =					torch.tensor([[1_0_1, 7_5_9_2, 1_0_1_0, 2_0_2_6, 3_8_9_9, 2_0_0_3, 1_0_1_4_0, 1_0_2]]						)
						with torch.no_grad():
									UpperCamelCase         =					model(__magic_name__						)[0]
						UpperCamelCase         =					torch.Size([1, 8, 1_0_2_4]						)
						self.assertEqual(output.shape       ,       __magic_name__						)
						UpperCamelCase         =					torch.tensor(
						    [[[0.1_775, 0.0_083, -0.0_321], [1.6_002, 0.1_287, 0.3_912], [2.1_473, 0.5_791, 0.6_066]]]						)
						self.assertTrue(torch.allclose(output[:, :3, :3]       ,       __magic_name__       ,       atol=1e-4						)						)




@require_torch
class    UpperCAmelCase					(       unittest.TestCase    ):




			@slow
			def   lowerCamelCase_					(						self							:		int						):







						"""simple docstring"""







						UpperCamelCase         =					BertGenerationDecoder.from_pretrained("""google/bert_for_seq_generation_L-24_bbc_encoder"""						)
						UpperCamelCase         =					torch.tensor([[1_0_1, 7_5_9_2, 1_0_1_0, 2_0_2_6, 3_8_9_9, 2_0_0_3, 1_0_1_4_0, 1_0_2]]						)
						with torch.no_grad():
									UpperCamelCase         =					model(__magic_name__						)[0]
						UpperCamelCase         =					torch.Size([1, 8, 5_0_3_5_8]						)
						self.assertEqual(output.shape       ,       __magic_name__						)
						UpperCamelCase         =					torch.tensor(
						    [[[-0.5_788, -2.5_994, -3.7_054], [0.0_438, 4.7_997, 1.8_795], [1.5_862, 6.6_409, 4.4_638]]]						)
						self.assertTrue(torch.allclose(output[:, :3, :3]       ,       __magic_name__       ,       atol=1e-4						)						)




