

import json
import os
import unittest

from transformers.models.gptsan_japanese.tokenization_gptsan_japanese import (
    VOCAB_FILES_NAMES,
    GPTSanJapaneseTokenizer,
)
from transformers.testing_utils import require_tokenizers, slow

from ...test_tokenization_common import TokenizerTesterMixin





@require_tokenizers
class 					__lowerCAmelCase	(   SCREAMING_SNAKE_CASE     ,     unittest.TestCase				):
  _a		     =    GPTSanJapaneseTokenizer
  _a		     =    False
  _a		     =    {"""do_clean_text""": False, """add_prefix_space""": False}




  def     A__	(							self   )    ->					List[str]:
        '''simple docstring'''






        super().setUp()

        # fmt: off
        _lowercase      			=['こん', 'こんに', 'にちは', 'ばんは', '世界,㔺界', '、', '。', '<BR>', '<SP>', '<TAB>', '<URL>', '<EMAIL>', '<TEL>', '<DATE>', '<PRICE>', '<BLOCK>', '<KIGOU>', '<U2000U2BFF>', '<|emoji1|>', '<unk>', '<|bagoftoken|>', '<|endoftext|>']
        # fmt: on
        _lowercase      			={'emoji': {'\ud83d\ude00': '<|emoji1|>'}, 'emoji_inv': {'<|emoji1|>': '\ud83d\ude00'}}  # 😀
        _lowercase      			={'unk_token': '<unk>'}

        _lowercase      			=os.path.join(self.tmpdirname				,  VOCAB_FILES_NAMES['vocab_file']   )
        _lowercase      			=os.path.join(self.tmpdirname				,  VOCAB_FILES_NAMES['emoji_file']   )
        with open(self.vocab_file				,  'w'				,  encoding='utf-8'   ) as vocab_writer:
              vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]   )   )
        with open(self.emoji_file				,  'w'   ) as emoji_writer:
              emoji_writer.write(json.dumps(lowerCAmelCase   )   )




  def     A__	(							self				,  **lowerCAmelCase   )    ->					Union[str, Any]:
        '''simple docstring'''






        kwargs.update(self.special_tokens_map   )
        return GPTSanJapaneseTokenizer.from_pretrained(self.tmpdirname				,  **lowerCAmelCase   )




  def     A__	(							self				,  lowerCAmelCase   )    ->					Union[str, Any]:
        '''simple docstring'''






        _lowercase      			='こんにちは、世界。 \nこんばんは、㔺界。😀'
        _lowercase      			='こんにちは、世界。 \nこんばんは、世界。😀'
        return input_text, output_text




  def     A__	(							self				,  lowerCAmelCase   )    ->					int:
        '''simple docstring'''






        _lowercase			,			_lowercase      			=self.get_input_output_texts(lowerCAmelCase   )
        _lowercase      			=tokenizer.encode(lowerCAmelCase				,  add_special_tokens=lowerCAmelCase   )
        _lowercase      			=tokenizer.decode(lowerCAmelCase				,  clean_up_tokenization_spaces=lowerCAmelCase   )
        return text, ids




  def     A__	(							self   )    ->					Optional[Any]:
        '''simple docstring'''






        pass  # TODO add if relevant




  def     A__	(							self   )    ->					List[Any]:
        '''simple docstring'''






        pass  # TODO add if relevant




  def     A__	(							self   )    ->					List[Any]:
        '''simple docstring'''






        pass  # TODO add if relevant




  def     A__	(							self   )    ->					int:
        '''simple docstring'''






        _lowercase      			=self.get_tokenizer()

        # Testing tokenization
        _lowercase      			='こんにちは、世界。　こんばんは、㔺界。'
        _lowercase      			=['こん', 'にちは', '、', '世界', '。', '<SP>', 'こん', 'ばんは', '、', '㔺界', '。']
        _lowercase      			=tokenizer.tokenize(lowerCAmelCase   )
        self.assertListEqual(lowerCAmelCase				,  lowerCAmelCase   )

        # Testing conversion to ids without special tokens
        _lowercase      			=[0, 2, 5, 4, 6, 8, 0, 3, 5, 4, 6]
        _lowercase      			=tokenizer.convert_tokens_to_ids(lowerCAmelCase   )
        self.assertListEqual(lowerCAmelCase				,  lowerCAmelCase   )

        # Testing conversion to ids with special tokens
        _lowercase      			=tokens + [tokenizer.unk_token]
        _lowercase      			=[0, 2, 5, 4, 6, 8, 0, 3, 5, 4, 6, 19]
        _lowercase      			=tokenizer.convert_tokens_to_ids(lowerCAmelCase   )
        self.assertListEqual(lowerCAmelCase				,  lowerCAmelCase   )




  def     A__	(							self   )    ->					Union[str, Any]:
        '''simple docstring'''






        _lowercase      			=self.get_tokenizer()

        # Testing tokenization
        _lowercase      			='こんにちは、<|bagoftoken|>世界。こんばんは、<|bagoftoken|>㔺界。'
        _lowercase      			='こんにちは、、、、世界。こんばんは、、、、世界。'
        _lowercase      			=tokenizer.encode(lowerCAmelCase   )
        _lowercase      			=tokenizer.decode(lowerCAmelCase   )
        self.assertEqual(lowerCAmelCase				,  lowerCAmelCase   )




  @slow
  def     A__	(							self   )    ->					Any:
        '''simple docstring'''






        _lowercase      			=self.tokenizer_class.from_pretrained('Tanrei/GPTSAN-japanese'   )

        # Testing tokenization
        _lowercase      			='こんにちは、世界。'
        _lowercase      			='こんばんは、㔺界。😀'
        _lowercase      			='こんにちは、世界。こんばんは、世界。😀'
        _lowercase      			=tokenizer.encode(prefix_text + input_text   )
        _lowercase      			=tokenizer.encode(''				,  prefix_text=prefix_text + input_text   )
        _lowercase      			=tokenizer.encode(lowerCAmelCase				,  prefix_text=lowerCAmelCase   )
        _lowercase      			=tokenizer.decode(lowerCAmelCase   )
        _lowercase      			=tokenizer.decode(lowerCAmelCase   )
        _lowercase      			=tokenizer.decode(lowerCAmelCase   )
        self.assertEqual(lowerCAmelCase				,  lowerCAmelCase   )
        self.assertEqual(lowerCAmelCase				,  lowerCAmelCase   )
        self.assertEqual(lowerCAmelCase				,  lowerCAmelCase   )




  @slow
  def     A__	(							self   )    ->					List[Any]:
        '''simple docstring'''






        _lowercase      			=self.tokenizer_class.from_pretrained('Tanrei/GPTSAN-japanese'   )

        # Testing tokenization
        _lowercase      			='こんにちは、世界。'
        _lowercase      			='こんばんは、㔺界。😀'

        _lowercase      			=len(tokenizer.encode(lowerCAmelCase   )   ) - 2
        _lowercase      			=len(tokenizer.encode(lowerCAmelCase   )   ) - 2

        _lowercase      			=[1] + [0] * (len_prefix + len_text + 1)
        _lowercase      			=[1] * (len_prefix + len_text + 1) + [0]
        _lowercase      			=[1] + [1] * (len_prefix) + [0] * (len_text + 1)

        _lowercase      			=tokenizer(prefix_text + input_text   ).token_type_ids
        _lowercase      			=tokenizer(''				,  prefix_text=prefix_text + input_text   ).token_type_ids
        _lowercase      			=tokenizer(lowerCAmelCase				,  prefix_text=lowerCAmelCase   ).token_type_ids
        self.assertListEqual(lowerCAmelCase				,  lowerCAmelCase   )
        self.assertListEqual(lowerCAmelCase				,  lowerCAmelCase   )
        self.assertListEqual(lowerCAmelCase				,  lowerCAmelCase   )




  @slow
  def     A__	(							self   )    ->					List[str]:
        '''simple docstring'''






        _lowercase      			=self.tokenizer_class.from_pretrained('Tanrei/GPTSAN-japanese'   )

        _lowercase      			=tokenizer.encode('あンいワ'   )
        _lowercase      			=tokenizer.encode(''				,  prefix_text='あンいワ'   )
        _lowercase      			=tokenizer.encode('いワ'				,  prefix_text='あン'   )

        self.assertEqual(tokenizer.decode(lowerCAmelCase   )				,  tokenizer.decode(lowerCAmelCase   )   )
        self.assertEqual(tokenizer.decode(lowerCAmelCase   )				,  tokenizer.decode(lowerCAmelCase   )   )
        self.assertNotEqual(lowerCAmelCase				,  lowerCAmelCase   )
        self.assertNotEqual(lowerCAmelCase				,  lowerCAmelCase   )
        self.assertEqual(x_token_a[1]				,  x_token_a[-1]   )  # SEG token
        self.assertEqual(x_token_a[1]				,  x_token_a[3]   )  # SEG token




  @slow
  def     A__	(							self   )    ->					Optional[int]:
        '''simple docstring'''






        _lowercase      			=self.tokenizer_class.from_pretrained('Tanrei/GPTSAN-japanese'   )

        _lowercase      			=[['武田信玄', 'は、'], ['織田信長', 'の配下の、']]
        _lowercase      			=tokenizer(lowerCAmelCase				,  padding=lowerCAmelCase   )
        _lowercase      			=tokenizer.batch_encode_plus(lowerCAmelCase				,  padding=lowerCAmelCase   )

        # fmt: off
        _lowercase      			=[[35_993, 8_640, 25_948, 35_998, 30_647, 35_675, 35_999, 35_999], [35_993, 10_382, 9_868, 35_998, 30_646, 9_459, 30_646, 35_675]]
        _lowercase      			=[[1, 1, 1, 0, 0, 0, 0, 0], [1, 1, 1, 0, 0, 0, 0, 0]]
        _lowercase      			=[[1, 1, 1, 1, 1, 1, 0, 0], [1, 1, 1, 1, 1, 1, 1, 1]]
        # fmt: on
        self.assertListEqual(x_token.input_ids				,  lowerCAmelCase   )
        self.assertListEqual(x_token.token_type_ids				,  lowerCAmelCase   )
        self.assertListEqual(x_token.attention_mask				,  lowerCAmelCase   )
        self.assertListEqual(x_token_a.input_ids				,  lowerCAmelCase   )
        self.assertListEqual(x_token_a.token_type_ids				,  lowerCAmelCase   )
        self.assertListEqual(x_token_a.attention_mask				,  lowerCAmelCase   )




  def     A__	(							self   )    ->					Any:
        '''simple docstring'''






        pass




  def     A__	(							self   )    ->					Any:
        '''simple docstring'''






        pass





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


lowercase_   							=  logging.get_logger(__name__)





@add_end_docstrings(SCREAMING_SNAKE_CASE				)
class 					__lowerCAmelCase	(   SCREAMING_SNAKE_CASE				):




  def __init__(							self				,  **lowerCAmelCase   )    ->					Tuple:
        '''simple docstring'''






        super().__init__(**lowerCAmelCase   )

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




  def __call__(							self				,  lowerCAmelCase				,  **lowerCAmelCase   )    ->					int:
        '''simple docstring'''






        return super().__call__(lowerCAmelCase				,  **lowerCAmelCase   )




  def     A__	(							self				,  **lowerCAmelCase   )    ->					Optional[Any]:
        '''simple docstring'''






        _lowercase      			={}
        if "candidate_labels" in kwargs:
              _lowercase      			=kwargs['candidate_labels']
        if "hypothesis_template" in kwargs:
              _lowercase      			=kwargs['hypothesis_template']

        return preprocess_params, {}, {}




  def     A__	(							self				,  lowerCAmelCase				,  lowerCAmelCase=None				,  lowerCAmelCase="This is a sound of {}."   )    ->					Tuple:
        '''simple docstring'''






        if isinstance(lowerCAmelCase				,  lowerCAmelCase   ):
              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
                    _lowercase      			=requests.get(lowerCAmelCase   ).content
              else:
                    with open(lowerCAmelCase				,  'rb'   ) as f:
                          _lowercase      			=f.read()

        if isinstance(lowerCAmelCase				,  lowerCAmelCase   ):
              _lowercase      			=ffmpeg_read(lowerCAmelCase				,  self.feature_extractor.sampling_rate   )

        if not isinstance(lowerCAmelCase				,  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'   )

        _lowercase      			=self.feature_extractor(
            [audio]				,  sampling_rate=self.feature_extractor.sampling_rate				,  return_tensors='pt'   )
        _lowercase      			=candidate_labels
        _lowercase      			=[hypothesis_template.format(lowerCAmelCase   ) for x in candidate_labels]
        _lowercase      			=self.tokenizer(lowerCAmelCase				,  return_tensors=self.framework				,  padding=lowerCAmelCase   )
        _lowercase      			=[text_inputs]
        return inputs




  def     A__	(							self				,  lowerCAmelCase   )    ->					Dict:
        '''simple docstring'''






        _lowercase      			=model_inputs.pop('candidate_labels'   )
        _lowercase      			=model_inputs.pop('text_inputs'   )
        if isinstance(text_inputs[0]				,  lowerCAmelCase   ):
              _lowercase      			=text_inputs[0]
        else:
              # Batching case.
              _lowercase      			=text_inputs[0][0]

        _lowercase      			=self.model(**lowerCAmelCase				,  **lowerCAmelCase   )

        _lowercase      			={
            'candidate_labels': candidate_labels,
            'logits': outputs.logits_per_audio,
        }
        return model_outputs




  def     A__	(							self				,  lowerCAmelCase   )    ->					Tuple:
        '''simple docstring'''






        _lowercase      			=model_outputs.pop('candidate_labels'   )
        _lowercase      			=model_outputs['logits'][0]

        if self.framework == "pt":
              _lowercase      			=logits.softmax(dim=0   )
              _lowercase      			=probs.tolist()
        else:
              raise ValueError('`tf` framework not supported.'   )

        _lowercase      			=[
            {'score': score, 'label': candidate_label}
            for score, candidate_label in sorted(zip(lowerCAmelCase				,  lowerCAmelCase   )				,  key=lambda lowerCAmelCase   : -x[0]   )
        ]
        return result


