





import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


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

snake_case__      :							str									=  logging.get_logger(__name__)



def _a       (			lowerCamelCase:     bytes      ,       lowerCamelCase:     int				)			->		np.array:







      '''simple docstring'''






      __A      =       F"""{sampling_rate}"""
      __A      =       '''1'''
      __A      =       '''f32le'''
      __A      =       [
          '''ffmpeg''',
          '''-i''',
          '''pipe:0''',
          '''-ac''',
          ac,
          '''-ar''',
          ar,
          '''-f''',
          format_for_conversion,
          '''-hide_banner''',
          '''-loglevel''',
          '''quiet''',
          '''pipe:1''',
      ]

      try:
            __A      =       subprocess.Popen(lowerCamelCase      ,       stdin=subprocess.PIPE      ,       stdout=subprocess.PIPE				)
      except FileNotFoundError:
            raise ValueError('''ffmpeg was not found but is required to load audio files from filename'''				)
      __A      =       ffmpeg_process.communicate(lowerCamelCase				)
      __A      =       output_stream[0]

      __A      =       np.frombuffer(lowerCamelCase      ,       np.floataa				)
      if audio.shape[0] == 0:
            raise ValueError('''Malformed soundfile'''				)
      return audio







@add_end_docstrings(_lowerCamelCase     )
class A_      (    _lowerCamelCase     ):




   def __init__(self			:Union[str, Any]					,     *_UpperCamelCase			:int					,     **_UpperCamelCase			:Dict				)->      str:
         # Default, might be overriden by the model.config.
         __A      =       5
         super().__init__(*_UpperCamelCase					,     **_UpperCamelCase				)

         if self.framework != "pt":
               raise ValueError(f"""The {self.__class__} is only available in PyTorch."""				)

         self.check_model_type(_UpperCamelCase				)




   def __call__(self			:List[str]					,     _UpperCamelCase			:Union[np.ndarray, bytes, str]					,     **_UpperCamelCase			:Any					,     )->      int:
         return super().__call__(_UpperCamelCase					,     **_UpperCamelCase				)




   def _lowerCAmelCase   (self			:Dict					,     _UpperCamelCase			:List[str]=None					,     **_UpperCamelCase			:List[Any]				)->      List[Any]:
         # No parameters on this pipeline right now
         __A      =       {}
         if top_k is not None:
               if top_k > self.model.config.num_labels:
                     __A      =       self.model.config.num_labels
               __A      =       top_k
         return {}, {}, postprocess_params




   def _lowerCAmelCase   (self			:List[str]					,     _UpperCamelCase			:int				)->      Any:
         if isinstance(_UpperCamelCase					,     _UpperCamelCase				):
               if inputs.startswith('''http://'''				) or inputs.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
                     __A      =       requests.get(_UpperCamelCase				).content
               else:
                     with open(_UpperCamelCase					,     '''rb'''				) as f:
                           __A      =       f.read()

         if isinstance(_UpperCamelCase					,     _UpperCamelCase				):
               __A      =       ffmpeg_read(_UpperCamelCase					,     self.feature_extractor.sampling_rate				)

         if isinstance(_UpperCamelCase					,     _UpperCamelCase				):
               # Accepting `"array"` which is the key defined in `datasets` for
               # better integration
               if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
                     raise ValueError(
                         '''When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a '''
                         '''"raw" key containing the numpy array representing the audio and a "sampling_rate" key, '''
                         '''containing the sampling_rate associated with that array'''				)

               __A      =       inputs.pop('''raw'''					,     _UpperCamelCase				)
               if _inputs is None:
                     # Remove path which will not be used from `datasets`.
                     inputs.pop('''path'''					,     _UpperCamelCase				)
                     __A      =       inputs.pop('''array'''					,     _UpperCamelCase				)
               __A      =       inputs.pop('''sampling_rate'''				)
               __A      =       _inputs
               if in_sampling_rate != self.feature_extractor.sampling_rate:
                     import torch

                     if is_torchaudio_available():
                           from torchaudio import functional as F
                     else:
                           raise ImportError(
                               '''torchaudio is required to resample audio samples in AudioClassificationPipeline. '''
                               '''The torchaudio package can be installed through: `pip install torchaudio`.'''				)

                     __A      =       F.resample(
                         torch.from_numpy(_UpperCamelCase				)					,     _UpperCamelCase					,     self.feature_extractor.sampling_rate				).numpy()

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

         __A      =       self.feature_extractor(
             _UpperCamelCase					,     sampling_rate=self.feature_extractor.sampling_rate					,     return_tensors='''pt'''				)
         return processed




   def _lowerCAmelCase   (self			:List[str]					,     _UpperCamelCase			:Dict				)->      int:
         __A      =       self.model(**_UpperCamelCase				)
         return model_outputs






   def _lowerCAmelCase   (self			:List[Any]					,     _UpperCamelCase			:List[Any]					,     _UpperCamelCase			:int=5				)->      Dict:
         __A      =       model_outputs.logits[0].softmax(-1				)
         __A     ,		__A      =       probs.topk(_UpperCamelCase				)

         __A      =       scores.tolist()
         __A      =       ids.tolist()

         __A      =       [{'''score''': score, '''label''': self.model.config.idalabel[_id]} for score, _id in zip(_UpperCamelCase					,     _UpperCamelCase				)]

         return labels









import argparse
import json

import gdown
import numpy as np
import torch
from huggingface_hub import hf_hub_download

from transformers import TimesformerConfig, TimesformerForVideoClassification, VideoMAEImageProcessor



def _a       (			lowerCamelCase:     str				)			->		List[str]:







      '''simple docstring'''






      __A      =       TimesformerConfig()

      if "large" in model_name:
            __A      =       96

      if "hr" in model_name:
            __A      =       16
            __A      =       4_48

      __A      =       '''huggingface/label-files'''
      if "k400" in model_name:
            __A      =       4_00
            __A      =       '''kinetics400-id2label.json'''
      elif "k600" in model_name:
            __A      =       6_00
            __A      =       '''kinetics600-id2label.json'''
      elif "ssv2" in model_name:
            __A      =       1_74
            __A      =       '''something-something-v2-id2label.json'''
      else:
            raise ValueError('''Model name should either contain \'k400\', \'k600\' or \'ssv2\'.'''				)
      __A      =       json.load(open(hf_hub_download(lowerCamelCase      ,       lowerCamelCase      ,       repo_type='''dataset'''				)      ,       '''r'''				)				)
      __A      =       {int(lowerCamelCase				): v for k, v in idalabel.items()}
      __A      =       idalabel
      __A      =       {v: k for k, v in idalabel.items()}

      return config



def _a       (			lowerCamelCase:     Optional[Any]				)			->		List[str]:







      '''simple docstring'''






      if "encoder." in name:
            __A      =       name.replace('''encoder.'''      ,       ''''''				)
      if "cls_token" in name:
            __A      =       name.replace('''cls_token'''      ,       '''timesformer.embeddings.cls_token'''				)
      if "pos_embed" in name:
            __A      =       name.replace('''pos_embed'''      ,       '''timesformer.embeddings.position_embeddings'''				)
      if "time_embed" in name:
            __A      =       name.replace('''time_embed'''      ,       '''timesformer.embeddings.time_embeddings'''				)
      if "patch_embed.proj" in name:
            __A      =       name.replace('''patch_embed.proj'''      ,       '''timesformer.embeddings.patch_embeddings.projection'''				)
      if "patch_embed.norm" in name:
            __A      =       name.replace('''patch_embed.norm'''      ,       '''timesformer.embeddings.norm'''				)
      if "blocks" in name:
            __A      =       name.replace('''blocks'''      ,       '''timesformer.encoder.layer'''				)
      if "attn.proj" in name:
            __A      =       name.replace('''attn.proj'''      ,       '''attention.output.dense'''				)
      if "attn" in name and "bias" not in name and "temporal" not in name:
            __A      =       name.replace('''attn'''      ,       '''attention.self'''				)
      if "attn" in name and "temporal" not in name:
            __A      =       name.replace('''attn'''      ,       '''attention.attention'''				)
      if "temporal_norm1" in name:
            __A      =       name.replace('''temporal_norm1'''      ,       '''temporal_layernorm'''				)
      if "temporal_attn.proj" in name:
            __A      =       name.replace('''temporal_attn'''      ,       '''temporal_attention.output.dense'''				)
      if "temporal_fc" in name:
            __A      =       name.replace('''temporal_fc'''      ,       '''temporal_dense'''				)
      if "norm1" in name and "temporal" not in name:
            __A      =       name.replace('''norm1'''      ,       '''layernorm_before'''				)
      if "norm2" in name:
            __A      =       name.replace('''norm2'''      ,       '''layernorm_after'''				)
      if "mlp.fc1" in name:
            __A      =       name.replace('''mlp.fc1'''      ,       '''intermediate.dense'''				)
      if "mlp.fc2" in name:
            __A      =       name.replace('''mlp.fc2'''      ,       '''output.dense'''				)
      if "norm.weight" in name and "fc" not in name and "temporal" not in name:
            __A      =       name.replace('''norm.weight'''      ,       '''timesformer.layernorm.weight'''				)
      if "norm.bias" in name and "fc" not in name and "temporal" not in name:
            __A      =       name.replace('''norm.bias'''      ,       '''timesformer.layernorm.bias'''				)
      if "head" in name:
            __A      =       name.replace('''head'''      ,       '''classifier'''				)

      return name



def _a       (			lowerCamelCase:     str      ,       lowerCamelCase:     Tuple				)			->		Tuple:







      '''simple docstring'''






      for key in orig_state_dict.copy().keys():
            __A      =       orig_state_dict.pop(lowerCamelCase				)

            if key.startswith('''model.'''				):
                  __A      =       key.replace('''model.'''      ,       ''''''				)

            if "qkv" in key:
                  __A      =       key.split('''.'''				)
                  __A      =       int(key_split[1]				)
                  __A      =       '''timesformer.encoder.layer.'''
                  if "temporal" in key:
                        __A      =       '''.temporal_attention.attention.qkv.'''
                  else:
                        __A      =       '''.attention.attention.qkv.'''
                  if "weight" in key:
                        __A      =       val
                  else:
                        __A      =       val
            else:
                  __A      =       val

      return orig_state_dict



def _a       (			)			->		str:







      '''simple docstring'''






      __A      =       hf_hub_download(
          repo_id='''hf-internal-testing/spaghetti-video'''      ,       filename='''eating_spaghetti.npy'''      ,       repo_type='''dataset'''				)
      __A      =       np.load(lowerCamelCase				)
      return list(lowerCamelCase				)



def _a       (			lowerCamelCase:     List[str]      ,       lowerCamelCase:     Tuple      ,       lowerCamelCase:     Optional[Any]      ,       lowerCamelCase:     Tuple				)			->		List[Any]:







      '''simple docstring'''






      __A      =       get_timesformer_config(lowerCamelCase				)

      __A      =       TimesformerForVideoClassification(lowerCamelCase				)

      # download original checkpoint, hosted on Google Drive
      __A      =       '''pytorch_model.bin'''
      gdown.cached_download(lowerCamelCase      ,       lowerCamelCase      ,       quiet=lowerCamelCase				)
      __A      =       torch.load(lowerCamelCase      ,       map_location='''cpu'''				)
      if "model" in files:
            __A      =       files['''model''']
      elif "module" in files:
            __A      =       files['''module''']
      else:
            __A      =       files['''model_state''']
      __A      =       convert_state_dict(lowerCamelCase      ,       lowerCamelCase				)

      model.load_state_dict(lowerCamelCase				)
      model.eval()

      # verify model on basic input
      __A      =       VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]      ,       image_std=[0.5, 0.5, 0.5]				)
      __A      =       prepare_video()
      __A      =       image_processor(video[:8]      ,       return_tensors='''pt'''				)

      __A      =       model(**lowerCamelCase				)
      __A      =       outputs.logits

      __A      =       [
          # Kinetics-400 checkpoints (hr = high resolution input of 448px instead of 224px)
          '''timesformer-base-finetuned-k400''',
          '''timesformer-large-finetuned-k400''',
          '''timesformer-hr-finetuned-k400''',
          # Kinetics-600 checkpoints (hr = high resolution input of 448px instead of 224px)
          '''timesformer-base-finetuned-k600''',
          '''timesformer-large-finetuned-k600''',
          '''timesformer-hr-finetuned-k600''',
          # Something-Something-v2 checkpoints (hr = high resolution input of 448px instead of 224px)
          '''timesformer-base-finetuned-ssv2''',
          '''timesformer-large-finetuned-ssv2''',
          '''timesformer-hr-finetuned-ssv2''',
      ]

      # NOTE: logits were tested with image_mean and image_std equal to [0.5, 0.5, 0.5] and [0.5, 0.5, 0.5]
      if model_name == "timesformer-base-finetuned-k400":
            __A      =       torch.Size([1, 4_00]				)
            __A      =       torch.tensor([-0.3_016, -0.7_713, -0.4_205]				)
      elif model_name == "timesformer-base-finetuned-k600":
            __A      =       torch.Size([1, 6_00]				)
            __A      =       torch.tensor([-0.7_267, -0.7_466, 3.2_404]				)
      elif model_name == "timesformer-base-finetuned-ssv2":
            __A      =       torch.Size([1, 1_74]				)
            __A      =       torch.tensor([-0.9_059, 0.6_433, -3.1_457]				)
      elif model_name == "timesformer-large-finetuned-k400":
            __A      =       torch.Size([1, 4_00]				)
            __A      =       torch.tensor([0, 0, 0]				)
      elif model_name == "timesformer-large-finetuned-k600":
            __A      =       torch.Size([1, 6_00]				)
            __A      =       torch.tensor([0, 0, 0]				)
      elif model_name == "timesformer-large-finetuned-ssv2":
            __A      =       torch.Size([1, 1_74]				)
            __A      =       torch.tensor([0, 0, 0]				)
      elif model_name == "timesformer-hr-finetuned-k400":
            __A      =       torch.Size([1, 4_00]				)
            __A      =       torch.tensor([-0.9_617, -3.7_311, -3.7_708]				)
      elif model_name == "timesformer-hr-finetuned-k600":
            __A      =       torch.Size([1, 6_00]				)
            __A      =       torch.tensor([2.5_273, 0.7_127, 1.8_848]				)
      elif model_name == "timesformer-hr-finetuned-ssv2":
            __A      =       torch.Size([1, 1_74]				)
            __A      =       torch.tensor([-3.6_756, -0.7_513, 0.7_180]				)
      else:
            raise ValueError(F"""Model name not supported. Should be one of {model_names}"""				)

      # verify logits
      assert logits.shape == expected_shape
      assert torch.allclose(logits[0, :3]      ,       lowerCamelCase      ,       atol=1e-4				)
      print('''Logits ok!'''				)

      if pytorch_dump_folder_path is not None:
            print(F"""Saving model and image processor to {pytorch_dump_folder_path}"""				)
            image_processor.save_pretrained(lowerCamelCase				)
            model.save_pretrained(lowerCamelCase				)

      if push_to_hub:
            print('''Pushing to the hub...'''				)
            model.push_to_hub(F"""fcakyon/{model_name}"""				)


if __name__ == "__main__":
       snake_case__      :							int									=  argparse.ArgumentParser()
       # Required parameters
       parser.add_argument(
           '--checkpoint_url',
           default='https://drive.google.com/u/1/uc?id=17yvuYp9L4mn-HpIcK5Zo6K3UoOy1kA5l&export=download',
           type=str,
           help=(
               'URL of the original PyTorch checkpoint (on Google Drive) you\'d like to convert. Should be a direct'
               ' download link.'
           ),
       )
       parser.add_argument(
           '--pytorch_dump_folder_path',
           default='',
           type=str,
           help='Path to the output PyTorch model directory.',
       )
       parser.add_argument('--model_name', default='timesformer-base-finetuned-k400', type=str, help='Name of the model.')
       parser.add_argument(
           '--push_to_hub', action='store_true', help='Whether or not to push the converted model to the 🤗 hub.'
       )

       snake_case__      :							str									=  parser.parse_args()
       convert_timesformer_checkpoint(
           args.checkpoint_url, args.pytorch_dump_folder_path, args.model_name, args.push_to_hub
       )





