
import enum
import warnings

from ..tokenization_utils import TruncationStrategy
from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
     import tensorflow as tf

     from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

if is_torch_available():
     from ..models.auto.modeling_auto import MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

UpperCamelCase								=					logging.get_logger(__name__)





class 			__lowerCamelCase	(   enum.Enum   ):




    """simple docstring"""

    snake_case__ =				0
    snake_case__ =				1






@add_end_docstrings(UpperCamelCase__   )
class 			__lowerCamelCase	(   UpperCamelCase__   ):




    """simple docstring"""

    snake_case__ =				'''generated'''


    def __init__(	self					:					Optional[int]       ,     *SCREAMING_SNAKE_CASE__					:					Optional[Any]       ,     **SCREAMING_SNAKE_CASE__					:					Tuple       )				->	Tuple:
         super().__init__(*_snake_case       ,     **_snake_case       )

         self.check_model_type(
             TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
             if self.framework == "tf"
             else MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING       )


    def        a				(	self					:					Optional[int]       ,     SCREAMING_SNAKE_CASE__					:					Optional[int]=None       ,     SCREAMING_SNAKE_CASE__					:					Optional[int]=None       ,     SCREAMING_SNAKE_CASE__					:					Tuple=None       ,     SCREAMING_SNAKE_CASE__					:					Tuple=None       ,     SCREAMING_SNAKE_CASE__					:					Any=None       ,     SCREAMING_SNAKE_CASE__					:					Optional[int]=None       ,     **SCREAMING_SNAKE_CASE__					:					str       ,     )				->	Tuple:
         lowerCAmelCase__          =			{}
         if truncation is not None:
              lowerCAmelCase__          =			truncation

         lowerCAmelCase__          =			generate_kwargs

         lowerCAmelCase__          =			{}
         if return_tensors is not None and return_type is None:
              lowerCAmelCase__          =			ReturnType.TENSORS if return_tensors else ReturnType.TEXT
         if return_type is not None:
              lowerCAmelCase__          =			return_type

         if clean_up_tokenization_spaces is not None:
              lowerCAmelCase__          =			clean_up_tokenization_spaces

         if stop_sequence is not None:
              lowerCAmelCase__          =			self.tokenizer.encode(_snake_case       ,     add_special_tokens=_snake_case       )
              if len(_snake_case       ) > 1:
                   warnings.warn(
                       "Stopping on a multiple token sequence is not yet supported on transformers. The first token of"
                       " the stop sequence will be used as the stop sequence string in the interim."       )
              lowerCAmelCase__          =			stop_sequence_ids[0]

         return preprocess_params, forward_params, postprocess_params


    def        a				(	self					:					List[str]       ,     SCREAMING_SNAKE_CASE__					:					int       ,     SCREAMING_SNAKE_CASE__					:					int       ,     SCREAMING_SNAKE_CASE__					:					int       )				->	List[Any]:
         return True


    def        a				(	self					:					Union[str, Any]       ,     *SCREAMING_SNAKE_CASE__					:					Union[str, Any]       ,     SCREAMING_SNAKE_CASE__					:					Optional[Any]       )				->	Dict:
         lowerCAmelCase__          =			self.model.config.prefix if self.model.config.prefix is not None else ""
         if isinstance(args[0]       ,     _snake_case       ):
              if self.tokenizer.pad_token_id is None:
                   raise ValueError("Please make sure that the tokenizer has a pad_token_id when using a batch input"       )
              lowerCAmelCase__          =			([prefix + arg for arg in args[0]],)
              lowerCAmelCase__          =			True

         elif isinstance(args[0]       ,     _snake_case       ):
              lowerCAmelCase__          =			(prefix + args[0],)
              lowerCAmelCase__          =			False
         else:
              raise ValueError(
                  f' `args[0]`: {args[0]} have the wrong format. The should be either of type `str` or type `list`'       )
         lowerCAmelCase__          =			self.tokenizer(*_snake_case       ,     padding=_snake_case       ,     truncation=_snake_case       ,     return_tensors=self.framework       )
         # This is produced by tokenizers but is an invalid generate kwargs
         if "token_type_ids" in inputs:
              del inputs["token_type_ids"]
         return inputs


    def __call__(	self					:					Optional[Any]       ,     *SCREAMING_SNAKE_CASE__					:					str       ,     **SCREAMING_SNAKE_CASE__					:					Any       )				->	int:
         lowerCAmelCase__          =			super().__call__(*_snake_case       ,     **_snake_case       )
         if (
             isinstance(args[0]       ,     _snake_case       )
             and all(isinstance(_snake_case       ,     _snake_case       ) for el in args[0]       )
             and all(len(_snake_case       ) == 1 for res in result       )
         ):
              return [res[0] for res in result]
         return result


    def        a				(	self					:					Tuple       ,     SCREAMING_SNAKE_CASE__					:					Any       ,     SCREAMING_SNAKE_CASE__					:					Dict=TruncationStrategy.DO_NOT_TRUNCATE       ,     **SCREAMING_SNAKE_CASE__					:					Tuple       )				->	Optional[int]:
         lowerCAmelCase__          =			self._parse_and_tokenize(_snake_case       ,     truncation=_snake_case       ,     **_snake_case       )
         return inputs


    def        a				(	self					:					Optional[int]       ,     SCREAMING_SNAKE_CASE__					:					List[str]       ,     **SCREAMING_SNAKE_CASE__					:					List[str]       )				->	Tuple:


         if self.framework == "pt":
              lowerCAmelCase__     ,		lowerCAmelCase__          =			model_inputs["input_ids"].shape
         elif self.framework == "tf":
              lowerCAmelCase__     ,		lowerCAmelCase__          =			tf.shape(model_inputs["input_ids"]       ).numpy()

         lowerCAmelCase__          =			generate_kwargs.get("min_length"       ,     self.model.config.min_length       )
         lowerCAmelCase__          =			generate_kwargs.get("max_length"       ,     self.model.config.max_length       )
         self.check_inputs(_snake_case       ,     generate_kwargs["min_length"]       ,     generate_kwargs["max_length"]       )
         lowerCAmelCase__          =			self.model.generate(**_snake_case       ,     **_snake_case       )
         lowerCAmelCase__          =			output_ids.shape[0]
         if self.framework == "pt":
              lowerCAmelCase__          =			output_ids.reshape(_snake_case       ,     out_b // in_b       ,     *output_ids.shape[1:]       )
         elif self.framework == "tf":
              lowerCAmelCase__          =			tf.reshape(_snake_case       ,     (in_b, out_b // in_b, *output_ids.shape[1:])       )
         return {"output_ids": output_ids}


    def        a				(	self					:					Optional[int]       ,     SCREAMING_SNAKE_CASE__					:					List[Any]       ,     SCREAMING_SNAKE_CASE__					:					Any=ReturnType.TEXT       ,     SCREAMING_SNAKE_CASE__					:					Dict=False       )				->	Union[str, Any]:
         lowerCAmelCase__          =			[]
         for output_ids in model_outputs["output_ids"][0]:
              if return_type == ReturnType.TENSORS:
                   lowerCAmelCase__          =			{f'{self.return_name}_token_ids': output_ids}
              elif return_type == ReturnType.TEXT:
                   lowerCAmelCase__          =			{
                       f'{self.return_name}_text': self.tokenizer.decode(
                           _snake_case       ,     skip_special_tokens=_snake_case       ,     clean_up_tokenization_spaces=_snake_case       ,     )
                   }
              records.append(_snake_case       )
         return records






@add_end_docstrings(UpperCamelCase__   )
class 			__lowerCamelCase	(   UpperCamelCase__   ):




    """simple docstring"""

    snake_case__ =				'''summary'''


    def __call__(	self					:					Dict       ,     *SCREAMING_SNAKE_CASE__					:					Dict       ,     **SCREAMING_SNAKE_CASE__					:					Dict       )				->	List[Any]:
         return super().__call__(*_snake_case       ,     **_snake_case       )


    def        a				(	self					:					Union[str, Any]       ,     SCREAMING_SNAKE_CASE__					:					int       ,     SCREAMING_SNAKE_CASE__					:					int       ,     SCREAMING_SNAKE_CASE__					:					int       )				->	Union[str, Any]:


         if max_length < min_length:
              logger.warning(f'Your min_length={min_length} must be inferior than your max_length={max_length}.'       )

         if input_length < max_length:
              logger.warning(
                  f'Your max_length is set to {max_length}, but your input_length is only {input_length}. Since this is '
                  "a summarization task, where outputs shorter than the input are typically wanted, you might "
                  f'consider decreasing max_length manually, e.g. summarizer(\'...\', max_length={input_length//2})'       )






@add_end_docstrings(UpperCamelCase__   )
class 			__lowerCamelCase	(   UpperCamelCase__   ):




    """simple docstring"""

    snake_case__ =				'''translation'''


    def        a				(	self					:					Any       ,     SCREAMING_SNAKE_CASE__					:					int       ,     SCREAMING_SNAKE_CASE__					:					int       ,     SCREAMING_SNAKE_CASE__					:					int       )				->	List[str]:


         if input_length > 0.9 * max_length:
              logger.warning(
                  f'Your input_length: {input_length} is bigger than 0.9 * max_length: {max_length}. You might consider '
                  "increasing your max_length manually, e.g. translator(\'...\', max_length=400)"       )
         return True


    def        a				(	self					:					Any       ,     *SCREAMING_SNAKE_CASE__					:					Any       ,     SCREAMING_SNAKE_CASE__					:					Optional[Any]=TruncationStrategy.DO_NOT_TRUNCATE       ,     SCREAMING_SNAKE_CASE__					:					int=None       ,     SCREAMING_SNAKE_CASE__					:					str=None       )				->	List[str]:


         if getattr(self.tokenizer       ,     "_build_translation_inputs"       ,     _snake_case       ):
              return self.tokenizer._build_translation_inputs(
                  *_snake_case       ,     return_tensors=self.framework       ,     truncation=_snake_case       ,     src_lang=_snake_case       ,     tgt_lang=_snake_case       )
         else:
              return super()._parse_and_tokenize(*_snake_case       ,     truncation=_snake_case       )


    def        a				(	self					:					List[Any]       ,     SCREAMING_SNAKE_CASE__					:					Optional[Any]=None       ,     SCREAMING_SNAKE_CASE__					:					Dict=None       ,     **SCREAMING_SNAKE_CASE__					:					List[Any]       )				->	Union[str, Any]:
         lowerCAmelCase__     ,		lowerCAmelCase__     ,		lowerCAmelCase__          =			super()._sanitize_parameters(**_snake_case       )
         if src_lang is not None:
              lowerCAmelCase__          =			src_lang
         if tgt_lang is not None:
              lowerCAmelCase__          =			tgt_lang
         if src_lang is None and tgt_lang is None:
              # Backward compatibility, direct arguments use is preferred.
              lowerCAmelCase__          =			kwargs.get("task"       ,     self.task       )
              lowerCAmelCase__          =			task.split("_"       )
              if task and len(_snake_case       ) == 4:
                   # translation, XX, to YY
                   lowerCAmelCase__          =			items[1]
                   lowerCAmelCase__          =			items[3]
         return preprocess_params, forward_params, postprocess_params


    def __call__(	self					:					List[Any]       ,     *SCREAMING_SNAKE_CASE__					:					List[str]       ,     **SCREAMING_SNAKE_CASE__					:					Optional[Any]       )				->	List[Any]:
         return super().__call__(*_snake_case       ,     **_snake_case       )






import unittest

from huggingface_hub import hf_hub_download

from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VideoMAEFeatureExtractor
from transformers.pipelines import VideoClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_decord,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
)

from .test_pipelines_common import ANY





@is_pipeline_test
@require_torch_or_tf
@require_vision
@require_decord
class 			__lowerCamelCase	(   unittest.TestCase   ):




    """simple docstring"""

    snake_case__ =				MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING


    def        a				(	self					:					str       ,     SCREAMING_SNAKE_CASE__					:					Tuple       ,     SCREAMING_SNAKE_CASE__					:					List[str]       ,     SCREAMING_SNAKE_CASE__					:					Optional[int]       )				->	Any:
         lowerCAmelCase__          =			hf_hub_download(
             repo_id="nateraw/video-demo"       ,     filename="archery.mp4"       ,     repo_type="dataset"       )
         lowerCAmelCase__          =			VideoClassificationPipeline(model=SCREAMING_SNAKE_CASE__       ,     image_processor=SCREAMING_SNAKE_CASE__       ,     top_k=2       )
         lowerCAmelCase__          =			[
             example_video_filepath,
             "https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4",
         ]
         return video_classifier, examples


    def        a				(	self					:					Any       ,     SCREAMING_SNAKE_CASE__					:					Dict       ,     SCREAMING_SNAKE_CASE__					:					Union[str, Any]       )				->	Tuple:
         for example in examples:
              lowerCAmelCase__          =			video_classifier(SCREAMING_SNAKE_CASE__       )

              self.assertEqual(
                  SCREAMING_SNAKE_CASE__       ,     [
                      {"score": ANY(SCREAMING_SNAKE_CASE__       ), "label": ANY(SCREAMING_SNAKE_CASE__       )},
                      {"score": ANY(SCREAMING_SNAKE_CASE__       ), "label": ANY(SCREAMING_SNAKE_CASE__       )},
                  ]       ,     )


    @require_torch
    def        a				(	self					:					Dict       )				->	Optional[Any]:
         lowerCAmelCase__          =			"hf-internal-testing/tiny-random-VideoMAEForVideoClassification"
         lowerCAmelCase__          =			VideoMAEFeatureExtractor(
             size={"shortest_edge": 10}       ,     crop_size={"height": 10, "width": 10}       )
         lowerCAmelCase__          =			pipeline(
             "video-classification"       ,     model=SCREAMING_SNAKE_CASE__       ,     feature_extractor=SCREAMING_SNAKE_CASE__       ,     frame_sampling_rate=4       )

         lowerCAmelCase__          =			hf_hub_download(repo_id="nateraw/video-demo"       ,     filename="archery.mp4"       ,     repo_type="dataset"       )
         lowerCAmelCase__          =			video_classifier(SCREAMING_SNAKE_CASE__       ,     top_k=2       )
         self.assertEqual(
             nested_simplify(SCREAMING_SNAKE_CASE__       ,     decimals=4       )       ,     [{"score": 0.5_199, "label": "LABEL_0"}, {"score": 0.4_801, "label": "LABEL_1"}]       ,     )

         lowerCAmelCase__          =			video_classifier(
             [
                 video_file_path,
                 video_file_path,
             ]       ,     top_k=2       ,     )
         self.assertEqual(
             nested_simplify(SCREAMING_SNAKE_CASE__       ,     decimals=4       )       ,     [
                 [{"score": 0.5_199, "label": "LABEL_0"}, {"score": 0.4_801, "label": "LABEL_1"}],
                 [{"score": 0.5_199, "label": "LABEL_0"}, {"score": 0.4_801, "label": "LABEL_1"}],
             ]       ,     )


    @require_tf
    def        a				(	self					:					Optional[Any]       )				->	Optional[int]:
         pass
