






import json
import os
import shutil
import tempfile
import unittest

from transformers import is_speech_available, is_torch_available
from transformers.models.speechta import SpeechTaTokenizer
from transformers.testing_utils import get_tests_dir, require_torch
from transformers.utils import FEATURE_EXTRACTOR_NAME


if is_speech_available() and is_torch_available():
							from transformers import SpeechTaFeatureExtractor, SpeechTaProcessor

							from .test_feature_extraction_speechta import floats_list


A__     :      List[Any]             =			get_tests_dir('fixtures/test_sentencepiece_bpe_char.model')
@require_torch
class 				_UpperCAmelCase (	unittest.TestCase     ):

				"""simple docstring"""




				def     lowercase__       (					self						:   List[Any]  ):

								'''simple docstring'''
								lowercase__       			=      tempfile.mkdtemp()

								lowercase__       			=      SpeechTaTokenizer(SCREAMING_SNAKE_CASE_  )
								tokenizer.save_pretrained(self.tmpdirname  )

								lowercase__       			=      {
								    '''feature_size''': 1,
								    '''padding_value''': 0.0,
								    '''sampling_rate''': 16_000,
								    '''do_normalize''': False,
								    '''num_mel_bins''': 80,
								    '''hop_length''': 16,
								    '''win_length''': 64,
								    '''win_function''': '''hann_window''',
								    '''fmin''': 80,
								    '''fmax''': 7_600,
								    '''mel_floor''': 1E-10,
								    '''reduction_factor''': 2,
								    '''return_attention_mask''': True,
								}

								lowercase__       			=      os.path.join(self.tmpdirname,  SCREAMING_SNAKE_CASE_  )
								with open(self.feature_extraction_file,  '''w''',  encoding='''utf-8'''  ) as fp:
												fp.write(json.dumps(SCREAMING_SNAKE_CASE_  ) + '''\n'''  )




				def     lowercase__       (					self						:   Any,  **lowerCamelCase						:   Tuple  ):

								'''simple docstring'''
								return SpeechTaTokenizer.from_pretrained(self.tmpdirname,  **SCREAMING_SNAKE_CASE_  )




				def     lowercase__       (					self						:   Union[str, Any],  **lowerCamelCase						:   str  ):

								'''simple docstring'''
								return SpeechTaFeatureExtractor.from_pretrained(self.tmpdirname,  **SCREAMING_SNAKE_CASE_  )




				def     lowercase__       (					self						:   str  ):

								'''simple docstring'''
								shutil.rmtree(self.tmpdirname  )




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

								'''simple docstring'''
								lowercase__       			=      self.get_tokenizer()
								lowercase__       			=      self.get_feature_extractor()

								lowercase__       			=      SpeechTaProcessor(tokenizer=SCREAMING_SNAKE_CASE_,  feature_extractor=SCREAMING_SNAKE_CASE_  )

								processor.save_pretrained(self.tmpdirname  )
								lowercase__       			=      SpeechTaProcessor.from_pretrained(self.tmpdirname  )

								self.assertEqual(processor.tokenizer.get_vocab(),  tokenizer.get_vocab()  )
								self.assertIsInstance(processor.tokenizer,  SCREAMING_SNAKE_CASE_  )

								self.assertEqual(processor.feature_extractor.to_json_string(),  feature_extractor.to_json_string()  )
								self.assertIsInstance(processor.feature_extractor,  SCREAMING_SNAKE_CASE_  )




				def     lowercase__       (					self						:   Dict  ):

								'''simple docstring'''
								lowercase__       			=      SpeechTaProcessor(tokenizer=self.get_tokenizer(),  feature_extractor=self.get_feature_extractor()  )
								processor.save_pretrained(self.tmpdirname  )

								lowercase__       			=      self.get_tokenizer(bos_token='''(BOS)''',  eos_token='''(EOS)'''  )
								lowercase__       			=      self.get_feature_extractor(do_normalize=SCREAMING_SNAKE_CASE_,  padding_value=1.0  )

								lowercase__       			=      SpeechTaProcessor.from_pretrained(
								    self.tmpdirname,  bos_token='''(BOS)''',  eos_token='''(EOS)''',  do_normalize=SCREAMING_SNAKE_CASE_,  padding_value=1.0  )

								self.assertEqual(processor.tokenizer.get_vocab(),  tokenizer_add_kwargs.get_vocab()  )
								self.assertIsInstance(processor.tokenizer,  SCREAMING_SNAKE_CASE_  )

								self.assertEqual(processor.feature_extractor.to_json_string(),  feature_extractor_add_kwargs.to_json_string()  )
								self.assertIsInstance(processor.feature_extractor,  SCREAMING_SNAKE_CASE_  )




				def     lowercase__       (					self						:   Any  ):

								'''simple docstring'''
								lowercase__       			=      self.get_feature_extractor()
								lowercase__       			=      self.get_tokenizer()

								lowercase__       			=      SpeechTaProcessor(tokenizer=SCREAMING_SNAKE_CASE_,  feature_extractor=SCREAMING_SNAKE_CASE_  )

								lowercase__       			=      floats_list((3, 1_000)  )

								lowercase__       			=      feature_extractor(audio=SCREAMING_SNAKE_CASE_,  return_tensors='''np'''  )
								lowercase__       			=      processor(audio=SCREAMING_SNAKE_CASE_,  return_tensors='''np'''  )

								for key in input_feat_extract.keys():
												self.assertAlmostEqual(input_feat_extract[key].sum(),  input_processor[key].sum(),  delta=1E-2  )




				def     lowercase__       (					self						:   Tuple  ):

								'''simple docstring'''
								lowercase__       			=      self.get_feature_extractor()
								lowercase__       			=      self.get_tokenizer()

								lowercase__       			=      SpeechTaProcessor(tokenizer=SCREAMING_SNAKE_CASE_,  feature_extractor=SCREAMING_SNAKE_CASE_  )

								lowercase__       			=      floats_list((3, 1_000)  )

								lowercase__       			=      feature_extractor(audio_target=SCREAMING_SNAKE_CASE_,  return_tensors='''np'''  )
								lowercase__       			=      processor(audio_target=SCREAMING_SNAKE_CASE_,  return_tensors='''np'''  )

								for key in input_feat_extract.keys():
												self.assertAlmostEqual(input_feat_extract[key].sum(),  input_processor[key].sum(),  delta=1E-2  )




				def     lowercase__       (					self						:   str  ):

								'''simple docstring'''
								lowercase__       			=      self.get_feature_extractor()
								lowercase__       			=      self.get_tokenizer()

								lowercase__       			=      SpeechTaProcessor(tokenizer=SCREAMING_SNAKE_CASE_,  feature_extractor=SCREAMING_SNAKE_CASE_  )

								lowercase__       			=      '''This is a test string'''

								lowercase__       			=      processor(text=SCREAMING_SNAKE_CASE_  )
								lowercase__       			=      tokenizer(SCREAMING_SNAKE_CASE_  )

								for key in encoded_tok.keys():
												self.assertListEqual(encoded_tok[key],  encoded_processor[key]  )




				def     lowercase__       (					self						:   int  ):

								'''simple docstring'''
								lowercase__       			=      self.get_feature_extractor()
								lowercase__       			=      self.get_tokenizer()

								lowercase__       			=      SpeechTaProcessor(tokenizer=SCREAMING_SNAKE_CASE_,  feature_extractor=SCREAMING_SNAKE_CASE_  )

								lowercase__       			=      '''This is a test string'''

								lowercase__       			=      processor(text_target=SCREAMING_SNAKE_CASE_  )
								lowercase__       			=      tokenizer(SCREAMING_SNAKE_CASE_  )

								for key in encoded_tok.keys():
												self.assertListEqual(encoded_tok[key],  encoded_processor[key]  )




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

								'''simple docstring'''
								lowercase__       			=      self.get_feature_extractor()
								lowercase__       			=      self.get_tokenizer()

								lowercase__       			=      SpeechTaProcessor(tokenizer=SCREAMING_SNAKE_CASE_,  feature_extractor=SCREAMING_SNAKE_CASE_  )

								lowercase__       			=      [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

								lowercase__       			=      processor.batch_decode(SCREAMING_SNAKE_CASE_  )
								lowercase__       			=      tokenizer.batch_decode(SCREAMING_SNAKE_CASE_  )

								self.assertListEqual(SCREAMING_SNAKE_CASE_,  SCREAMING_SNAKE_CASE_  )




				def     lowercase__       (					self						:   Optional[Any]  ):

								'''simple docstring'''
								lowercase__       			=      self.get_feature_extractor()
								lowercase__       			=      self.get_tokenizer()

								lowercase__       			=      SpeechTaProcessor(tokenizer=SCREAMING_SNAKE_CASE_,  feature_extractor=SCREAMING_SNAKE_CASE_  )

								self.assertListEqual(
								    processor.model_input_names,  feature_extractor.model_input_names,  msg='''`processor` and `feature_extractor` model input names do not match''',  )







import textwrap

import pyarrow as pa
import pytest

from datasets import Features, Image
from datasets.packaged_modules.text.text import Text

from ..utils import require_pil






@pytest.fixture
def   a      (						lowerCamelCase_    ):






				'''simple docstring'''
				lowercase__       			=      tmp_path / '''text.txt'''
				lowercase__       			=      textwrap.dedent(
				    '''\
        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.
        Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.
        Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.
        Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.

        Second paragraph:
        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.
        Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.
        Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.
        Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.
        '''    )
				with open(lowerCamelCase_		,       '''w'''		,       encoding='''utf-8'''    ) as f:
								f.write(lowerCamelCase_    )
				return str(lowerCamelCase_    )






@pytest.fixture
def   a      (						lowerCamelCase_		,       lowerCamelCase_    ):






				'''simple docstring'''
				lowercase__       			=      tmp_path / '''text_with_image.txt'''
				with open(lowerCamelCase_		,       '''w'''		,       encoding='''utf-8'''    ) as f:
								f.write(lowerCamelCase_    )
				return str(lowerCamelCase_    )






@pytest.mark.parametrize('''keep_linebreaks'''		,       [True, False]    )
def   a      (						lowerCamelCase_		,       lowerCamelCase_    ):






				'''simple docstring'''
				with open(lowerCamelCase_		,       encoding='''utf-8'''    ) as f:
								lowercase__       			=      f.read().splitlines(keepends=lowerCamelCase_    )
				lowercase__       			=      Text(keep_linebreaks=lowerCamelCase_		,       encoding='''utf-8'''    )
				lowercase__       			=      text._generate_tables([[text_file]]    )
				lowercase__       			=      pa.concat_tables([table for _, table in generator]    ).to_pydict()['''text''']
				assert generated_content == expected_content






@require_pil
def   a      (						lowerCamelCase_    ):






				'''simple docstring'''
				with open(lowerCamelCase_		,       encoding='''utf-8'''    ) as f:
								lowercase__       			=      f.read().splitlines()[0]
				lowercase__       			=      Text(encoding='''utf-8'''		,       features=Features({'''image''': Image()}    )    )
				lowercase__       			=      text._generate_tables([[text_file_with_image]]    )
				lowercase__       			=      pa.concat_tables([table for _, table in generator]    )
				assert pa_table.schema.field('''image'''    ).type == Image()()
				lowercase__       			=      pa_table.to_pydict()['''image''']
				assert generated_content == [{"path": image_file, "bytes": None}]






@pytest.mark.parametrize('''sample_by'''		,       ['''line''', '''paragraph''', '''document''']    )
def   a      (						lowerCamelCase_		,       lowerCamelCase_    ):






				'''simple docstring'''
				with open(lowerCamelCase_		,       encoding='''utf-8'''    ) as f:
								lowercase__       			=      f.read()
				if sample_by == "line":
								lowercase__       			=      expected_content.splitlines()
				elif sample_by == "paragraph":
								lowercase__       			=      expected_content.split('''\n\n'''    )
				elif sample_by == "document":
								lowercase__       			=      [expected_content]
				lowercase__       			=      Text(sample_by=lowerCamelCase_		,       encoding='''utf-8'''		,       chunksize=100    )
				lowercase__       			=      text._generate_tables([[text_file]]    )
				lowercase__       			=      pa.concat_tables([table for _, table in generator]    ).to_pydict()['''text''']
				assert generated_content == expected_content




