
"""simple docstring"""




import numpy as np
import torch
from torch.utils.data import Dataset

from utils import logger


class       __magic_name__							(							SCREAMING_SNAKE_CASE__			):
       def __init__(  self			,					A_			,					A_							)  ->						List[str]:





         """simple docstring"""




         _lowercase:      List[str]   			=  params

         _lowercase:      str   			=  np.array(A_							)
         _lowercase:      Optional[int]   			=  np.array([len(A_							) for t in data]							)

         self.check()
         self.remove_long_sequences()
         self.remove_empty_sequences()
         self.remove_unknown_sequences()
         self.check()
         self.print_statistics()
       def __getitem__(  self			,					A_							)  ->						Optional[Any]:





         """simple docstring"""




         return (self.token_ids[index], self.lengths[index])
       def __len__(  self							)  ->						List[str]:





         """simple docstring"""




         return len(self.lengths							)
       def 				lowercase_				(  self							)  ->						Dict:





         """simple docstring"""




         assert len(self.token_ids							) == len(self.lengths							)
         assert all(self.lengths[i] == len(self.token_ids[i]							) for i in range(len(self.lengths							)							)							)
       def 				lowercase_				(  self							)  ->						str:





         """simple docstring"""




         _lowercase:      Tuple   			=  self.params.max_model_input_size
         _lowercase:      Tuple   			=  self.lengths > max_len
         logger.info(f'''Splitting {sum(A_							)} too long sequences.'''							)

         def divide_chunks(A_			,					A_							):
           return [l[i : i + n] for i in range(0			,					len(A_							)			,					A_							)]

         _lowercase:      Dict   			=  []
         _lowercase:      Union[str, Any]   			=  []
         if self.params.mlm:
           _lowercase			,					_lowercase:      int   			=  self.params.special_tok_ids['''cls_token'''], self.params.special_tok_ids['''sep_token''']
         else:
           _lowercase			,					_lowercase:      Dict   			=  self.params.special_tok_ids['''bos_token'''], self.params.special_tok_ids['''eos_token''']

         for seq_, len_ in zip(self.token_ids			,					self.lengths							):
           assert (seq_[0] == cls_id) and (seq_[-1] == sep_id), seq_
           if len_ <= max_len:
             new_tok_ids.append(seq_							)
             new_lengths.append(len_							)
           else:
             _lowercase:      Optional[Any]   			=  []
             for sub_s in divide_chunks(seq_			,					max_len - 2							):
               if sub_s[0] != cls_id:
                 _lowercase:      Any   			=  np.insert(A_			,					0			,					A_							)
               if sub_s[-1] != sep_id:
                 _lowercase:      Optional[int]   			=  np.insert(A_			,					len(A_							)			,					A_							)
               assert len(A_							) <= max_len
               assert (sub_s[0] == cls_id) and (sub_s[-1] == sep_id), sub_s
               sub_seqs.append(A_							)

             new_tok_ids.extend(A_							)
             new_lengths.extend([len(A_							) for l in sub_seqs]							)

         _lowercase:      Optional[Any]   			=  np.array(A_							)
         _lowercase:      List[str]   			=  np.array(A_							)
       def 				lowercase_				(  self							)  ->						Tuple:





         """simple docstring"""




         _lowercase:      List[Any]   			=  len(self							)
         _lowercase:      Optional[Any]   			=  self.lengths > 11
         _lowercase:      int   			=  self.token_ids[indices]
         _lowercase:      List[str]   			=  self.lengths[indices]
         _lowercase:      Optional[int]   			=  len(self							)
         logger.info(f'''Remove {init_size - new_size} too short (<=11 tokens) sequences.'''							)
       def 				lowercase_				(  self							)  ->						int:





         """simple docstring"""




         if "unk_token" not in self.params.special_tok_ids:
           return
         else:
           _lowercase:      Dict   			=  self.params.special_tok_ids['''unk_token''']
         _lowercase:      int   			=  len(self							)
         _lowercase:      Tuple   			=  np.array([np.count_nonzero(a == unk_token_id							) for a in self.token_ids]							)
         _lowercase:      Dict   			=  (unk_occs / self.lengths) < 0.5
         _lowercase:      str   			=  self.token_ids[indices]
         _lowercase:      Dict   			=  self.lengths[indices]
         _lowercase:      int   			=  len(self							)
         logger.info(f'''Remove {init_size - new_size} sequences with a high level of unknown tokens (50%).'''							)
       def 				lowercase_				(  self							)  ->						Optional[Any]:





         """simple docstring"""




         if not self.params.is_master:
           return
         logger.info(f'''{len(self							)} sequences'''							)
         # data_len = sum(self.lengths)
         # nb_unique_tokens = len(Counter(list(chain(*self.token_ids))))
         # logger.info(f'{data_len} tokens ({nb_unique_tokens} unique)')

         # unk_idx = self.params.special_tok_ids['unk_token']
         # nb_unknown = sum([(t==unk_idx).sum() for t in self.token_ids])
         # logger.info(f'{nb_unknown} unknown tokens (covering {100*nb_unknown/data_len:.2f}% of the data)')



       def 				lowercase_				(  self			,					A_							)  ->						Optional[int]:





         """simple docstring"""




         _lowercase:      List[str]   			=  [t[0] for t in batch]
         _lowercase:      Dict   			=  [t[1] for t in batch]
         assert len(A_							) == len(A_							)

         # Max for paddings
         _lowercase:      Tuple   			=  max(A_							)

         # Pad token ids
         if self.params.mlm:
           _lowercase:      str   			=  self.params.special_tok_ids['''pad_token''']
         else:
           _lowercase:      Union[str, Any]   			=  self.params.special_tok_ids['''unk_token''']
         _lowercase:      int   			=  [list(t.astype(A_							)							) + [pad_idx] * (max_seq_len_ - len(A_							)) for t in token_ids]
         assert len(tk_							) == len(A_							)
         assert all(len(A_							) == max_seq_len_ for t in tk_							)

         _lowercase:      str   			=  torch.tensor(tk_							)  # (bs, max_seq_len_)
         _lowercase:      int   			=  torch.tensor(A_							)  # (bs)
         return tk_t, lg_t


"""simple docstring"""




import unittest

from transformers import is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
  from PIL import Image
else:


  class       __magic_name__							:
         @staticmethod
         def 				lowercase_				(  *A_			,					**A_							)  ->						Optional[int]:





           """simple docstring"""




           pass


@is_pipeline_test
@require_vision
class       __magic_name__							(							unittest.TestCase			):
       @require_torch
       def 				lowercase_				(  self							)  ->						int:





         """simple docstring"""




         _lowercase:      Optional[Any]   			=  pipeline(
             model='''hf-internal-testing/tiny-random-clip-zero-shot-image-classification'''			,					)
         _lowercase:      List[Any]   			=  Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''							)
         _lowercase:      str   			=  image_classifier(A_			,					candidate_labels=['''a''', '''b''', '''c''']							)

         # The floating scores are so close, we enter floating error approximation and the order is not guaranteed across
         # python and torch versions.
         self.assertIn(
             nested_simplify(A_							)			,					[
                 [{'''score''': 0.3_33, '''label''': '''a'''}, {'''score''': 0.3_33, '''label''': '''b'''}, {'''score''': 0.3_33, '''label''': '''c'''}],
                 [{'''score''': 0.3_33, '''label''': '''a'''}, {'''score''': 0.3_33, '''label''': '''c'''}, {'''score''': 0.3_33, '''label''': '''b'''}],
             ]			,					)

         _lowercase:      Any   			=  image_classifier([image] * 5			,					candidate_labels=['''A''', '''B''', '''C''']			,					batch_size=2							)
         self.assertEqual(
             nested_simplify(A_							)			,					[
                 [
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                 ],
                 [
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                 ],
                 [
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                 ],
                 [
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                 ],
                 [
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                 ],
             ]			,					)
       @require_tf
       def 				lowercase_				(  self							)  ->						int:





         """simple docstring"""




         _lowercase:      Union[str, Any]   			=  pipeline(
             model='''hf-internal-testing/tiny-random-clip-zero-shot-image-classification'''			,					framework='''tf'''							)
         _lowercase:      List[str]   			=  Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''							)
         _lowercase:      Union[str, Any]   			=  image_classifier(A_			,					candidate_labels=['''a''', '''b''', '''c''']							)

         self.assertEqual(
             nested_simplify(A_							)			,					[{'''score''': 0.3_33, '''label''': '''a'''}, {'''score''': 0.3_33, '''label''': '''b'''}, {'''score''': 0.3_33, '''label''': '''c'''}]			,					)

         _lowercase:      List[str]   			=  image_classifier([image] * 5			,					candidate_labels=['''A''', '''B''', '''C''']			,					batch_size=2							)
         self.assertEqual(
             nested_simplify(A_							)			,					[
                 [
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                 ],
                 [
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                 ],
                 [
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                 ],
                 [
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                 ],
                 [
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                     {'''score''': 0.3_33, '''label''': ANY(A_							)},
                 ],
             ]			,					)
       @slow
       @require_torch
       def 				lowercase_				(  self							)  ->						Any:





         """simple docstring"""




         _lowercase:      Tuple   			=  pipeline(
             task='''zero-shot-image-classification'''			,					model='''openai/clip-vit-base-patch32'''			,					)
         # This is an image of 2 cats with remotes and no planes
         _lowercase:      str   			=  Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''							)
         _lowercase:      Optional[Any]   			=  image_classifier(A_			,					candidate_labels=['''cat''', '''plane''', '''remote''']							)

         self.assertEqual(
             nested_simplify(A_							)			,					[
                 {'''score''': 0.5_11, '''label''': '''remote'''},
                 {'''score''': 0.4_85, '''label''': '''cat'''},
                 {'''score''': 0.0_04, '''label''': '''plane'''},
             ]			,					)

         _lowercase:      Optional[Any]   			=  image_classifier([image] * 5			,					candidate_labels=['''cat''', '''plane''', '''remote''']			,					batch_size=2							)
         self.assertEqual(
             nested_simplify(A_							)			,					[
                 [
                     {'''score''': 0.5_11, '''label''': '''remote'''},
                     {'''score''': 0.4_85, '''label''': '''cat'''},
                     {'''score''': 0.0_04, '''label''': '''plane'''},
                 ],
             ]
             * 5			,					)



       @slow
       @require_tf
       def 				lowercase_				(  self							)  ->						Optional[Any]:





         """simple docstring"""




         _lowercase:      Tuple   			=  pipeline(
             task='''zero-shot-image-classification'''			,					model='''openai/clip-vit-base-patch32'''			,					framework='''tf'''							)
         # This is an image of 2 cats with remotes and no planes
         _lowercase:      str   			=  Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''							)
         _lowercase:      Optional[int]   			=  image_classifier(A_			,					candidate_labels=['''cat''', '''plane''', '''remote''']							)
         self.assertEqual(
             nested_simplify(A_							)			,					[
                 {'''score''': 0.5_11, '''label''': '''remote'''},
                 {'''score''': 0.4_85, '''label''': '''cat'''},
                 {'''score''': 0.0_04, '''label''': '''plane'''},
             ]			,					)

         _lowercase:      Optional[Any]   			=  image_classifier([image] * 5			,					candidate_labels=['''cat''', '''plane''', '''remote''']			,					batch_size=2							)
         self.assertEqual(
             nested_simplify(A_							)			,					[
                 [
                     {'''score''': 0.5_11, '''label''': '''remote'''},
                     {'''score''': 0.4_85, '''label''': '''cat'''},
                     {'''score''': 0.0_04, '''label''': '''plane'''},
                 ],
             ]
             * 5			,					)
