


import os
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from huggingface_hub.file_download import http_get
from requests.exceptions import HTTPError

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


sys.path.append(str(Path(__file__).parent.parent / '''utils'''))

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
     from test_module.custom_tokenization_fast import CustomTokenizerFast







class 		_snake_case  (				unittest.TestCase							):




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:Any										=	mock.Mock()
       a						:int										=	500
       a						:Optional[int]										=	{}
       a						:Tuple										=	HTTPError
       a						:Dict										=	{}

       # Download this model to make sure it's in the cache.
       a						:Optional[Any]										=	BertTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert'''   )

       # Under the mock environment we get a 500 error when trying to reach the tokenizer.
       with mock.patch('''requests.Session.request''' ,  return_value=a_   ) as mock_head:
             a						:List[str]										=	BertTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert'''   )
             # This check we did call the fake head request
             mock_head.assert_called()




 @require_tokenizers
 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:Union[str, Any]										=	mock.Mock()
       a						:Optional[Any]										=	500
       a						:str										=	{}
       a						:Tuple										=	HTTPError
       a						:str										=	{}

       # Download this model to make sure it's in the cache.
       a						:int										=	GPTaTokenizerFast.from_pretrained('''gpt2'''   )

       # Under the mock environment we get a 500 error when trying to reach the tokenizer.
       with mock.patch('''requests.Session.request''' ,  return_value=a_   ) as mock_head:
             a						:Union[str, Any]										=	GPTaTokenizerFast.from_pretrained('''gpt2'''   )
             # This check we did call the fake head request
             mock_head.assert_called()




 def 	SCREAMING_SNAKE_CASE__    (		self   ):



       try:
             a						:str										=	tempfile.mktemp()
             with open(a_ ,  '''wb'''   ) as f:
                   http_get('''https://huggingface.co/albert-base-v1/resolve/main/spiece.model''' ,  a_   )

             a						:Union[str, Any]										=	AlbertTokenizer.from_pretrained(a_   )
       finally:
             os.remove(a_   )

       # Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
       # the current folder and have the right name.
       if os.path.isfile('''tokenizer.json'''   ):
             # We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
             return
       try:
             with open('''tokenizer.json''' ,  '''wb'''   ) as f:
                   http_get('''https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json''' ,  a_   )
             a						:Dict										=	AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''   )
             # The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
             self.assertEqual(tokenizer.vocab_size ,  1000   )
             # Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

       finally:
             os.remove('''tokenizer.json'''   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:str										=	AlbertTokenizer.from_pretrained('''https://huggingface.co/albert-base-v1/resolve/main/spiece.model'''   )







@is_staging_test
class 		_snake_case  (				unittest.TestCase							):
 SCREAMING_SNAKE_CASE__				=	['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'bla', 'blou']




 @classmethod
 def 	SCREAMING_SNAKE_CASE__    (		cls   ):
       a						:Optional[Any]										=	TOKEN
       HfFolder.save_token(a_   )




 @classmethod
 def 	SCREAMING_SNAKE_CASE__    (		cls   ):



       try:
             delete_repo(token=cls._token ,  repo_id='''test-tokenizer'''   )
       except HTTPError:
             pass

       try:
             delete_repo(token=cls._token ,  repo_id='''valid_org/test-tokenizer-org'''   )
       except HTTPError:
             pass

       try:
             delete_repo(token=cls._token ,  repo_id='''test-dynamic-tokenizer'''   )
       except HTTPError:
             pass




 def 	SCREAMING_SNAKE_CASE__    (		self   ):



       with tempfile.TemporaryDirectory() as tmp_dir:
             a						:List[str]										=	os.path.join(a_ ,  '''vocab.txt'''   )
             with open(a_ ,  '''w''' ,  encoding='''utf-8'''   ) as vocab_writer:
                   vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]   )   )
             a						:Any										=	BertTokenizer(a_   )

       tokenizer.push_to_hub('''test-tokenizer''' ,  use_auth_token=self._token   )
       a						:Union[str, Any]										=	BertTokenizer.from_pretrained(F'''{USER}/test-tokenizer'''   )
       self.assertDictEqual(new_tokenizer.vocab ,  tokenizer.vocab   )

       # Reset repo
       delete_repo(token=self._token ,  repo_id='''test-tokenizer'''   )

       # Push to hub via save_pretrained
       with tempfile.TemporaryDirectory() as tmp_dir:
             tokenizer.save_pretrained(a_ ,  repo_id='''test-tokenizer''' ,  push_to_hub=a_ ,  use_auth_token=self._token   )

       a						:List[str]										=	BertTokenizer.from_pretrained(F'''{USER}/test-tokenizer'''   )
       self.assertDictEqual(new_tokenizer.vocab ,  tokenizer.vocab   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):



       with tempfile.TemporaryDirectory() as tmp_dir:
             a						:Dict										=	os.path.join(a_ ,  '''vocab.txt'''   )
             with open(a_ ,  '''w''' ,  encoding='''utf-8'''   ) as vocab_writer:
                   vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]   )   )
             a						:Dict										=	BertTokenizer(a_   )

       tokenizer.push_to_hub('''valid_org/test-tokenizer-org''' ,  use_auth_token=self._token   )
       a						:Dict										=	BertTokenizer.from_pretrained('''valid_org/test-tokenizer-org'''   )
       self.assertDictEqual(new_tokenizer.vocab ,  tokenizer.vocab   )

       # Reset repo
       delete_repo(token=self._token ,  repo_id='''valid_org/test-tokenizer-org'''   )

       # Push to hub via save_pretrained
       with tempfile.TemporaryDirectory() as tmp_dir:
             tokenizer.save_pretrained(
                 a_ ,  repo_id='''valid_org/test-tokenizer-org''' ,  push_to_hub=a_ ,  use_auth_token=self._token   )

       a						:Optional[int]										=	BertTokenizer.from_pretrained('''valid_org/test-tokenizer-org'''   )
       self.assertDictEqual(new_tokenizer.vocab ,  tokenizer.vocab   )




 @require_tokenizers
 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       CustomTokenizer.register_for_auto_class()
       with tempfile.TemporaryDirectory() as tmp_dir:
             a						:Tuple										=	os.path.join(a_ ,  '''vocab.txt'''   )
             with open(a_ ,  '''w''' ,  encoding='''utf-8'''   ) as vocab_writer:
                   vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]   )   )
             a						:Any										=	CustomTokenizer(a_   )

       # No fast custom tokenizer
       tokenizer.push_to_hub('''test-dynamic-tokenizer''' ,  use_auth_token=self._token   )

       a						:Union[str, Any]										=	AutoTokenizer.from_pretrained(F'''{USER}/test-dynamic-tokenizer''' ,  trust_remote_code=a_   )
       # Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
       self.assertEqual(tokenizer.__class__.__name__ ,  '''CustomTokenizer'''   )

       # Fast and slow custom tokenizer
       CustomTokenizerFast.register_for_auto_class()
       with tempfile.TemporaryDirectory() as tmp_dir:
             a						:Dict										=	os.path.join(a_ ,  '''vocab.txt'''   )
             with open(a_ ,  '''w''' ,  encoding='''utf-8'''   ) as vocab_writer:
                   vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]   )   )

             a						:List[Any]										=	BertTokenizerFast.from_pretrained(a_   )
             bert_tokenizer.save_pretrained(a_   )
             a						:Tuple										=	CustomTokenizerFast.from_pretrained(a_   )

       tokenizer.push_to_hub('''test-dynamic-tokenizer''' ,  use_auth_token=self._token   )

       a						:List[Any]										=	AutoTokenizer.from_pretrained(F'''{USER}/test-dynamic-tokenizer''' ,  trust_remote_code=a_   )
       # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
       self.assertEqual(tokenizer.__class__.__name__ ,  '''CustomTokenizerFast'''   )
       a						:Any										=	AutoTokenizer.from_pretrained(
           F'''{USER}/test-dynamic-tokenizer''' ,  use_fast=a_ ,  trust_remote_code=a_   )
       # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
       self.assertEqual(tokenizer.__class__.__name__ ,  '''CustomTokenizer'''   )







class 		_snake_case  (				unittest.TestCase							):




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:int										=	Trie()
       trie.add('''Hello 友達'''   )
       self.assertEqual(trie.data ,  {'''H''': {'''e''': {'''l''': {'''l''': {'''o''': {''' ''': {'''友''': {'''達''': {'''''': 1}}}}}}}}}   )
       trie.add('''Hello'''   )
       trie.data
       self.assertEqual(trie.data ,  {'''H''': {'''e''': {'''l''': {'''l''': {'''o''': {'''''': 1, ''' ''': {'''友''': {'''達''': {'''''': 1}}}}}}}}}   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:Union[str, Any]										=	Trie()
       self.assertEqual(trie.split('''[CLS] This is a extra_id_100'''   ) ,  ['''[CLS] This is a extra_id_100''']   )
       trie.add('''[CLS]'''   )
       trie.add('''extra_id_1'''   )
       trie.add('''extra_id_100'''   )
       self.assertEqual(trie.split('''[CLS] This is a extra_id_100'''   ) ,  ['''[CLS]''', ''' This is a ''', '''extra_id_100''']   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:str										=	Trie()
       trie.add('''A'''   )
       self.assertEqual(trie.split('''ABC'''   ) ,  ['''A''', '''BC''']   )
       self.assertEqual(trie.split('''BCA'''   ) ,  ['''BC''', '''A''']   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:Any										=	Trie()
       trie.add('''TOKEN]'''   )
       trie.add('''[SPECIAL_TOKEN]'''   )
       self.assertEqual(trie.split('''This is something [SPECIAL_TOKEN]'''   ) ,  ['''This is something ''', '''[SPECIAL_TOKEN]''']   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:List[str]										=	Trie()
       trie.add('''A'''   )
       trie.add('''P'''   )
       trie.add('''[SPECIAL_TOKEN]'''   )
       self.assertEqual(trie.split('''This is something [SPECIAL_TOKEN]'''   ) ,  ['''This is something ''', '''[SPECIAL_TOKEN]''']   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:List[Any]										=	Trie()
       trie.add('''AB'''   )
       trie.add('''B'''   )
       trie.add('''C'''   )
       self.assertEqual(trie.split('''ABC'''   ) ,  ['''AB''', '''C''']   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:str										=	Trie()
       trie.add('''ABC'''   )
       trie.add('''B'''   )
       trie.add('''CD'''   )
       self.assertEqual(trie.split('''ABCD'''   ) ,  ['''ABC''', '''D''']   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:Optional[int]										=	Trie()
       a						:Optional[Any]										=	trie.cut_text('''ABC''' ,  [0, 0, 2, 1, 2, 3]   )
       self.assertEqual(a_ ,  ['''AB''', '''C''']   )









# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import torch

from ..models.speechta import SpeechTaForTextToSpeech, SpeechTaHifiGan, SpeechTaProcessor
from ..utils import is_datasets_available
from .base import PipelineTool


if is_datasets_available():
     from datasets import load_dataset







class 		_snake_case  (				_snake_case							):
 SCREAMING_SNAKE_CASE__				=	'microsoft/speecht5_tts'
 SCREAMING_SNAKE_CASE__				=	(
     'This is a tool that reads an English text out loud. It takes an input named `text` which should contain the '
     'text to read (in English) and returns a waveform object containing the sound.'
 )
 SCREAMING_SNAKE_CASE__				=	'text_reader'
 SCREAMING_SNAKE_CASE__				=	SpeechTaProcessor
 SCREAMING_SNAKE_CASE__				=	SpeechTaForTextToSpeech
 SCREAMING_SNAKE_CASE__				=	SpeechTaHifiGan

 SCREAMING_SNAKE_CASE__				=	['text']
 SCREAMING_SNAKE_CASE__				=	['audio']




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       if self.post_processor is None:
             a						:List[Any]										=	'''microsoft/speecht5_hifigan'''
       super().setup()




 def 	SCREAMING_SNAKE_CASE__    (		self ,  _lowerCamelCase ,  _lowerCamelCase=None   ):
       a						:Tuple										=	self.pre_processor(text=_lowerCamelCase ,  return_tensors='''pt''' ,  truncation=_lowerCamelCase   )

       if speaker_embeddings is None:
             if not is_datasets_available():
                   raise ImportError('''Datasets needs to be installed if not passing speaker embeddings.'''   )

             a						:List[Any]										=	load_dataset('''Matthijs/cmu-arctic-xvectors''' ,  split='''validation'''   )
             a						:int										=	torch.tensor(embeddings_dataset[7305]['''xvector''']   ).unsqueeze(0   )

       return {"input_ids": inputs["input_ids"], "speaker_embeddings": speaker_embeddings}




 def 	SCREAMING_SNAKE_CASE__    (		self ,  _lowerCamelCase   ):
       with torch.no_grad():
             return self.model.generate_speech(**_lowerCamelCase   )




 def 	SCREAMING_SNAKE_CASE__    (		self ,  _lowerCamelCase   ):
       with torch.no_grad():
             return self.post_processor(_lowerCamelCase   ).cpu().detach()





