

import json
import os
import re
import shutil
import tempfile
import unittest
from typing import Tuple

from transformers import AddedToken, BatchEncoding, ByTaTokenizer
from transformers.utils import cached_property, is_tf_available, is_torch_available

from ...test_tokenization_common import TokenizerTesterMixin


if is_torch_available():
     lowercase_       		=  "pt"
elif is_tf_available():
     lowercase_       		=  "tf"
else:
     lowercase_       		=  "jax"


class  A						(							_UpperCAmelCase    ,							unittest.TestCase ):







    """simple docstring"""





    lowerCamelCase			=							ByTaTokenizer
    lowerCamelCase			=							False


    def 						snake_case__     (      self :       Tuple		)->    Any:





        '''simple docstring'''







        super().setUp()
        A__          =   ByTaTokenizer()
        tokenizer.save_pretrained(self.tmpdirname		)


    @cached_property
    def 						snake_case__     (      self :       List[Any]		)->    Optional[int]:





        '''simple docstring'''







        return ByTaTokenizer.from_pretrained('google/byt5-small'		)


    def 						snake_case__     (      self :       List[Any],**lowercase_ :       Tuple		)->    ByTaTokenizer:





        '''simple docstring'''







        return self.tokenizer_class.from_pretrained(self.tmpdirname,**lowercase_		)


    def 						snake_case__     (      self :       Tuple,lowercase_ :       List[str],lowercase_ :       List[Any]=False,lowercase_ :       Optional[int]=2_0,lowercase_ :       Dict=5		)->    Tuple[str, list]:





        '''simple docstring'''







        A__          =   []
        for i in range(len(lowercase_		)		):
            try:
                A__          =   tokenizer.decode([i],clean_up_tokenization_spaces=lowercase_		)
            except UnicodeDecodeError:
                pass
            toks.append((i, tok)		)

        A__          =   list(filter(lambda lowercase_		: re.match(r'^[ a-zA-Z]+$',t[1]		),lowercase_		)		)
        A__          =   list(filter(lambda lowercase_		: [t[0]] == tokenizer.encode(t[1],add_special_tokens=lowercase_		),lowercase_		)		)
        if max_length is not None and len(lowercase_		) > max_length:
            A__          =   toks[:max_length]
        if min_length is not None and len(lowercase_		) < min_length and len(lowercase_		) > 0:
            while len(lowercase_		) < min_length:
                A__          =   toks + toks
        # toks_str = [t[1] for t in toks]
        A__          =   [t[0] for t in toks]

        # Ensure consistency
        A__          =   tokenizer.decode(lowercase_,clean_up_tokenization_spaces=lowercase_		)
        if " " not in output_txt and len(lowercase_		) > 1:
            A__          =   (
                tokenizer.decode([toks_ids[0]],clean_up_tokenization_spaces=lowercase_		)
                + ' '
                + tokenizer.decode(toks_ids[1:],clean_up_tokenization_spaces=lowercase_		)
            )
        if with_prefix_space:
            A__          =   ' ' + output_txt
        A__          =   tokenizer.encode(lowercase_,add_special_tokens=lowercase_		)
        return output_txt, output_ids


    def 						snake_case__     (      self :       Optional[int]		)->    Any:





        '''simple docstring'''







        A__          =   self.ta_base_tokenizer
        A__          =   tokenizer(['hi</s>', 'I went to the gym</s>', '</s>']		)
        A__          =   tokenizer(['hi', 'I went to the gym', '']		)
        self.assertListEqual(batch_with_eos_added['input_ids'],batch_without_eos_added['input_ids']		)


    def 						snake_case__     (      self :       Union[str, Any]		)->    Any:





        '''simple docstring'''







        A__          =   self.ta_base_tokenizer
        A__          =   'Unicode €.'
        A__          =   tokenizer(lowercase_		)
        A__          =   [8_8, 1_1_3, 1_0_8, 1_0_2, 1_1_4, 1_0_3, 1_0_4, 3_5, 2_2_9, 1_3_3, 1_7_5, 4_9, 1]
        self.assertEqual(encoded['input_ids'],lowercase_		)

        # decoding
        A__          =   tokenizer.decode(lowercase_		)
        self.assertEqual(lowercase_,'Unicode €.</s>'		)

        A__          =   tokenizer('e è é ê ë'		)
        A__          =   [1_0_4, 3_5, 1_9_8, 1_7_1, 3_5, 1_9_8, 1_7_2, 3_5, 1_9_8, 1_7_3, 3_5, 1_9_8, 1_7_4, 1]
        self.assertEqual(encoded['input_ids'],lowercase_		)
        # decoding
        A__          =   tokenizer.decode(lowercase_		)
        self.assertEqual(lowercase_,'e è é ê ë</s>'		)

        # encode/decode, but with `encode` instead of `__call__`
        self.assertEqual(tokenizer.decode(tokenizer.encode('e è é ê ë'		)		),'e è é ê ë</s>'		)


    def 						snake_case__     (      self :       str		)->    Dict:





        '''simple docstring'''







        A__          =   self.ta_base_tokenizer
        A__          =   ['A long paragraph for summarization.', 'Another paragraph for summarization.']
        # fmt: off
        A__          =   [6_8, 3_5, 1_1_1, 1_1_4, 1_1_3, 1_0_6, 3_5, 1_1_5, 1_0_0, 1_1_7, 1_0_0, 1_0_6, 1_1_7, 1_0_0, 1_1_5, 1_0_7, 3_5, 1_0_5, 1_1_4, 1_1_7, 3_5, 1_1_8, 1_2_0, 1_1_2, 1_1_2, 1_0_0, 1_1_7, 1_0_8, 1_2_5, 1_0_0, 1_1_9, 1_0_8, 1_1_4, 1_1_3, 4_9, 1, 0]
        # fmt: on
        A__          =   tokenizer(lowercase_,padding=lowercase_,return_tensors=lowercase_		)
        self.assertIsInstance(lowercase_,lowercase_		)

        if FRAMEWORK != "jax":
            A__          =   list(batch.input_ids.numpy()[0]		)
        else:
            A__          =   list(batch.input_ids.tolist()[0]		)

        self.assertListEqual(lowercase_,lowercase_		)

        self.assertEqual((2, 3_7),batch.input_ids.shape		)
        self.assertEqual((2, 3_7),batch.attention_mask.shape		)


    def 						snake_case__     (      self :       Optional[Any]		)->    Tuple:





        '''simple docstring'''







        A__          =   self.ta_base_tokenizer
        A__          =   ['A long paragraph for summarization.', 'Another paragraph for summarization.']
        A__          =   tokenizer(lowercase_,padding=lowercase_,return_tensors=lowercase_		)
        # check if input_ids are returned and no decoder_input_ids
        self.assertIn('input_ids',lowercase_		)
        self.assertIn('attention_mask',lowercase_		)
        self.assertNotIn('decoder_input_ids',lowercase_		)
        self.assertNotIn('decoder_attention_mask',lowercase_		)


    def 						snake_case__     (      self :       Optional[Any]		)->    List[str]:





        '''simple docstring'''







        A__          =   self.ta_base_tokenizer
        A__          =   [
            'Summary of the text.',
            'Another summary.',
        ]
        A__          =   tokenizer(
            text_target=lowercase_,max_length=3_2,padding='max_length',truncation=lowercase_,return_tensors=lowercase_		)
        self.assertEqual(3_2,targets['input_ids'].shape[1]		)


    def 						snake_case__     (      self :       str		)->    Optional[int]:





        '''simple docstring'''







        A__          =   self.ta_base_tokenizer
        A__          =   ['A long paragraph for summarization. </s>']
        A__          =   ['Summary of the text. </s>']
        # fmt: off
        A__          =   [6_8, 3_5, 1_1_1, 1_1_4, 1_1_3, 1_0_6, 3_5, 1_1_5, 1_0_0, 1_1_7, 1_0_0, 1_0_6, 1_1_7, 1_0_0, 1_1_5, 1_0_7, 3_5, 1_0_5, 1_1_4, 1_1_7, 3_5, 1_1_8, 1_2_0, 1_1_2, 1_1_2, 1_0_0, 1_1_7, 1_0_8, 1_2_5, 1_0_0, 1_1_9, 1_0_8, 1_1_4, 1_1_3, 4_9, 3_5, 1]
        A__          =   [8_6, 1_2_0, 1_1_2, 1_1_2, 1_0_0, 1_1_7, 1_2_4, 3_5, 1_1_4, 1_0_5, 3_5, 1_1_9, 1_0_7, 1_0_4, 3_5, 1_1_9, 1_0_4, 1_2_3, 1_1_9, 4_9, 3_5, 1]
        # fmt: on

        A__          =   tokenizer(lowercase_,text_target=lowercase_		)

        self.assertEqual(lowercase_,batch['input_ids'][0]		)
        self.assertEqual(lowercase_,batch['labels'][0]		)


    def 						snake_case__     (      self :       Optional[int]		)->    Optional[Any]:





        '''simple docstring'''







        A__          =   self.get_tokenizers()
        for tokenizer in tokenizers:
            with self.subTest(F'{tokenizer.__class__.__name__}'		):
                self.assertNotEqual(tokenizer.model_max_length,4_2		)

        # Now let's start the test
        A__          =   self.get_tokenizers()
        for tokenizer in tokenizers:
            with self.subTest(F'{tokenizer.__class__.__name__}'		):
                # Isolate this from the other tests because we save additional tokens/etc
                A__          =   tempfile.mkdtemp()

                A__          =   ' He is very happy, UNwant\u00E9d,running'
                A__          =   tokenizer.encode(lowercase_,add_special_tokens=lowercase_		)
                tokenizer.save_pretrained(lowercase_		)

                A__          =   tokenizer.__class__.from_pretrained(lowercase_		)
                A__          =   after_tokenizer.encode(lowercase_,add_special_tokens=lowercase_		)
                self.assertListEqual(lowercase_,lowercase_		)

                shutil.rmtree(lowercase_		)

        A__          =   self.get_tokenizers(model_max_length=4_2		)
        for tokenizer in tokenizers:
            with self.subTest(F'{tokenizer.__class__.__name__}'		):
                # Isolate this from the other tests because we save additional tokens/etc
                A__          =   tempfile.mkdtemp()

                A__          =   ' He is very happy, UNwant\u00E9d,running'
                tokenizer.add_tokens(['bim', 'bambam']		)
                A__          =   tokenizer.additional_special_tokens
                additional_special_tokens.append('new_additional_special_token'		)
                tokenizer.add_special_tokens({'additional_special_tokens': additional_special_tokens}		)
                A__          =   tokenizer.encode(lowercase_,add_special_tokens=lowercase_		)
                tokenizer.save_pretrained(lowercase_		)

                A__          =   tokenizer.__class__.from_pretrained(lowercase_		)
                A__          =   after_tokenizer.encode(lowercase_,add_special_tokens=lowercase_		)
                self.assertListEqual(lowercase_,lowercase_		)
                self.assertIn('new_additional_special_token',after_tokenizer.additional_special_tokens		)
                self.assertEqual(after_tokenizer.model_max_length,4_2		)

                A__          =   tokenizer.__class__.from_pretrained(lowercase_,model_max_length=4_3		)
                self.assertEqual(tokenizer.model_max_length,4_3		)

                shutil.rmtree(lowercase_		)


    def 						snake_case__     (      self :       Tuple		)->    Union[str, Any]:





        '''simple docstring'''







        A__          =   []
        if self.test_slow_tokenizer:
            tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())		)

        if self.test_rust_tokenizer:
            tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())		)

        for tokenizer_class, tokenizer_utils in tokenizer_list:
            with tempfile.TemporaryDirectory() as tmp_dir:
                tokenizer_utils.save_pretrained(lowercase_		)

                with open(os.path.join(lowercase_,'special_tokens_map.json'		),encoding='utf-8'		) as json_file:
                    A__          =   json.load(lowercase_		)

                with open(os.path.join(lowercase_,'tokenizer_config.json'		),encoding='utf-8'		) as json_file:
                    A__          =   json.load(lowercase_		)

                A__          =   [F'<extra_id_{i}>' for i in range(1_2_5		)]

                A__          =   added_tokens_extra_ids + [
                    'an_additional_special_token'
                ]
                A__          =   added_tokens_extra_ids + [
                    'an_additional_special_token'
                ]

                with open(os.path.join(lowercase_,'special_tokens_map.json'		),'w',encoding='utf-8'		) as outfile:
                    json.dump(lowercase_,lowercase_		)
                with open(os.path.join(lowercase_,'tokenizer_config.json'		),'w',encoding='utf-8'		) as outfile:
                    json.dump(lowercase_,lowercase_		)

                # the following checks allow us to verify that our test works as expected, i.e. that the tokenizer takes
                # into account the new value of additional_special_tokens given in the "tokenizer_config.json" and
                # "special_tokens_map.json" files
                A__          =   tokenizer_class.from_pretrained(
                    lowercase_,)
                self.assertIn(
                    'an_additional_special_token',tokenizer_without_change_in_init.additional_special_tokens		)
                # self.assertIn("an_additional_special_token",tokenizer_without_change_in_init.get_vocab()) # ByT5Tokenization no vocab
                self.assertEqual(
                    ['an_additional_special_token'],tokenizer_without_change_in_init.convert_ids_to_tokens(
                        tokenizer_without_change_in_init.convert_tokens_to_ids(['an_additional_special_token']		)		),)

                # Now we test that we can change the value of additional_special_tokens in the from_pretrained
                A__          =   added_tokens_extra_ids + [AddedToken('a_new_additional_special_token',lstrip=lowercase_		)]
                A__          =   tokenizer_class.from_pretrained(
                    lowercase_,additional_special_tokens=lowercase_,)

                self.assertIn('a_new_additional_special_token',tokenizer.additional_special_tokens		)
                self.assertEqual(
                    ['a_new_additional_special_token'],tokenizer.convert_ids_to_tokens(
                        tokenizer.convert_tokens_to_ids(['a_new_additional_special_token']		)		),)


    def 						snake_case__     (      self :       List[str]		)->    Any:





        '''simple docstring'''







        A__          =   []
        if self.test_slow_tokenizer:
            tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())		)

        if self.test_rust_tokenizer:
            tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())		)

        for tokenizer_class, tokenizer_utils in tokenizer_list:
            with tempfile.TemporaryDirectory() as tmp_dir:
                tokenizer_utils.save_pretrained(lowercase_		)

                A__          =   tokenizer_class.from_pretrained(lowercase_		)

                self.assertTrue(tokenizer.decode([2_5_5]		) == ''		)


    def 						snake_case__     (      self :       List[str]		)->    Optional[Any]:





        '''simple docstring'''







        pass


    def 						snake_case__     (      self :       Optional[Any]		)->    List[str]:





        '''simple docstring'''







        pass


    def 						snake_case__     (      self :       List[str]		)->    Any:





        '''simple docstring'''







        pass


    def 						snake_case__     (      self :       Dict		)->    Tuple:





        '''simple docstring'''







        pass


    def 						snake_case__     (      self :       Optional[Any]		)->    Optional[Any]:





        '''simple docstring'''







        A__          =   self.get_tokenizers(fast=lowercase_,do_lower_case=lowercase_		)
        for tokenizer in tokenizers:
            with self.subTest(F'{tokenizer.__class__.__name__}'		):
                A__          =   ['t', 'h', 'i', 's', ' ', 'i', 's', ' ', 'a', ' ', 't', 'e', 'x', 't', '</s>']
                A__          =   tokenizer.convert_tokens_to_string(lowercase_		)

                self.assertIsInstance(lowercase_,lowercase_		)


    def 						snake_case__     (      self :       Tuple		)->    str:





        '''simple docstring'''







        A__          =   self.get_tokenizers()
        for tokenizer in tokenizers:
            with self.subTest(F'{tokenizer.__class__.__name__}'		):
                A__          =   [
                    'bos_token',
                    'eos_token',
                    'unk_token',
                    'sep_token',
                    'pad_token',
                    'cls_token',
                    'mask_token',
                ]

                A__          =   0
                A__          =   tokenizer.convert_ids_to_tokens(
                    lowercase_,skip_special_tokens=lowercase_		)

                for attr in attributes_list:
                    setattr(lowercase_,attr + '_id',lowercase_		)
                    self.assertEqual(getattr(lowercase_,lowercase_		),lowercase_		)
                    self.assertEqual(getattr(lowercase_,attr + '_id'		),lowercase_		)

                    setattr(lowercase_,attr + '_id',lowercase_		)
                    self.assertEqual(getattr(lowercase_,lowercase_		),lowercase_		)
                    self.assertEqual(getattr(lowercase_,attr + '_id'		),lowercase_		)

                setattr(lowercase_,'additional_special_tokens_ids',[]		)
                self.assertListEqual(getattr(lowercase_,'additional_special_tokens'		),[]		)
                self.assertListEqual(getattr(lowercase_,'additional_special_tokens_ids'		),[]		)

                setattr(lowercase_,'additional_special_tokens_ids',[token_id_to_test_setters]		)
                self.assertListEqual(getattr(lowercase_,'additional_special_tokens'		),[token_to_test_setters]		)
                self.assertListEqual(getattr(lowercase_,'additional_special_tokens_ids'		),[token_id_to_test_setters]		)






from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


a_   :List[Any]        =      logging.get_logger(__name__)




@add_end_docstrings(lowerCAmelCase_     )
class 					snake_case__						(       lowerCAmelCase_     ):






     """simple docstring"""


     def __init__(       self  :  Optional[Any], **_snake_case  :  str					)   ->Dict:
            super().__init__(**_snake_case					)

            if self.framework != "pt":
                   raise ValueError(F'''The {self.__class__} is only available in PyTorch.'''					)
            # No specific FOR_XXX available yet






     def __call__(       self  :  Union[str, Any], _snake_case  :  Union[np.ndarray, bytes, str], **_snake_case  :  Tuple					)   ->Dict:
            return super().__call__(_snake_case, **_snake_case					)






     def    lowercase_   (       self  :  Tuple, **_snake_case  :  Any					)   ->Union[str, Any]:
            snake_case__							:	str   	=	{}
            if "candidate_labels" in kwargs:
                   snake_case__							:	str   	=	kwargs['candidate_labels']
            if "hypothesis_template" in kwargs:
                   snake_case__							:	str   	=	kwargs['hypothesis_template']

            return preprocess_params, {}, {}






     def    lowercase_   (       self  :  Dict, _snake_case  :  str, _snake_case  :  Optional[int]=None, _snake_case  :  List[str]="This is a sound of {}."					)   ->int:
            if isinstance(_snake_case, _snake_case					):
                   if audio.startswith('http://'					) or audio.startswith('https://'					):
                          # We need to actually check for a real protocol, otherwise it's impossible to use a local file
                          # like http_huggingface_co.png
                          snake_case__							:	List[Any]   	=	requests.get(_snake_case					).content
                   else:
                          with open(_snake_case, 'rb'					) as f:
                                 snake_case__							:	Union[str, Any]   	=	f.read()

            if isinstance(_snake_case, _snake_case					):
                   snake_case__							:	List[Any]   	=	ffmpeg_read(_snake_case, self.feature_extractor.sampling_rate					)

            if not isinstance(_snake_case, np.ndarray					):
                   raise ValueError('We expect a numpy ndarray as input'					)
            if len(audio.shape					) != 1:
                   raise ValueError('We expect a single channel audio input for ZeroShotAudioClassificationPipeline'					)

            snake_case__							:	Tuple   	=	self.feature_extractor(
                [audio], sampling_rate=self.feature_extractor.sampling_rate, return_tensors='pt'					)
            snake_case__							:	int   	=	candidate_labels
            snake_case__							:	int   	=	[hypothesis_template.format(_snake_case					) for x in candidate_labels]
            snake_case__							:	Optional[int]   	=	self.tokenizer(_snake_case, return_tensors=self.framework, padding=_snake_case					)
            snake_case__							:	List[Any]   	=	[text_inputs]
            return inputs






     def    lowercase_   (       self  :  Optional[int], _snake_case  :  Optional[Any]					)   ->int:
            snake_case__							:	Optional[int]   	=	model_inputs.pop('candidate_labels'					)
            snake_case__							:	str   	=	model_inputs.pop('text_inputs'					)
            if isinstance(text_inputs[0], _snake_case					):
                   snake_case__							:	Optional[Any]   	=	text_inputs[0]
            else:
                   # Batching case.
                   snake_case__							:	int   	=	text_inputs[0][0]

            snake_case__							:	Any   	=	self.model(**_snake_case, **_snake_case					)

            snake_case__							:	List[Any]   	=	{
                'candidate_labels': candidate_labels,
                'logits': outputs.logits_per_audio,
            }
            return model_outputs






     def    lowercase_   (       self  :  Union[str, Any], _snake_case  :  str					)   ->List[str]:
            snake_case__							:	int   	=	model_outputs.pop('candidate_labels'					)
            snake_case__							:	List[Any]   	=	model_outputs['logits'][0]

            if self.framework == "pt":
                   snake_case__							:	Tuple   	=	logits.softmax(dim=0					)
                   snake_case__							:	Union[str, Any]   	=	probs.tolist()
            else:
                   raise ValueError('`tf` framework not supported.'					)

            snake_case__							:	Union[str, Any]   	=	[
                {'score': score, 'label': candidate_label}
                for score, candidate_label in sorted(zip(_snake_case, _snake_case					), key=lambda _snake_case					: -x[0]					)
            ]
            return result






