


'''simple docstring'''
import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
  from .features import FeatureType


UpperCamelCase__   ,  UpperCamelCase__   ,  UpperCamelCase__          = False, False, False




@dataclass
class      lowerCamelCase_  :
      lowerCAmelCase__							=	None
      lowerCAmelCase__							=	True
      lowerCAmelCase__							=	True
      lowerCAmelCase__							=	None
      # Automatically constructed
      lowerCAmelCase__							=	"dict"
      lowerCAmelCase__							=	pa.struct({'bytes': pa.binary(), 'path': pa.string()} )
      lowerCAmelCase__							=	field(default='Audio'	,			init=__a	,			repr=__a )






      def __call__(   self							:       Optional[int]					):


          '''simple docstring'''


          return self.pa_type






      def lowercase_				(   self							:       Dict		,		_A							:       Union[str, bytes, dict]					):


          '''simple docstring'''


          try:
              import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
          except ImportError as err:
              raise ImportError('''To support encoding audio data, please install \'soundfile\'.'''					) from err
          if isinstance(_A		,		_A					):
              return {"bytes": None, "path": value}
          elif isinstance(_A		,		_A					):
              return {"bytes": value, "path": None}
          elif "array" in value:
              # convert the audio array to wav bytes
              UpperCAmelCase__     :						Optional[int]						=							BytesIO()
              sf.write(_A		,		value['''array''']		,		value['''sampling_rate''']		,		format='''wav'''					)
              return {"bytes": buffer.getvalue(), "path": None}
          elif value.get('''path'''					) is not None and os.path.isfile(value['''path''']					):
              # we set "bytes": None to not duplicate the data if they're already available locally
              if value["path"].endswith('''pcm'''					):
                  # "PCM" only has raw audio bytes
                  if value.get('''sampling_rate'''					) is None:
                      # At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
                      raise KeyError('''To use PCM files, please specify a \'sampling_rate\' in Audio object'''					)
                  if value.get('''bytes'''					):
                      # If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
                      UpperCAmelCase__     :						Dict						=							np.frombuffer(value['''bytes''']		,		dtype=np.intaa					).astype(np.floataa					) / 32_767
                  else:
                      UpperCAmelCase__     :						Optional[int]						=							np.memmap(value['''path''']		,		dtype='''h'''		,		mode='''r'''					).astype(np.floataa					) / 32_767

                  UpperCAmelCase__     :						Tuple						=							BytesIO(bytes()					)
                  sf.write(_A		,		_A		,		value['''sampling_rate''']		,		format='''wav'''					)
                  return {"bytes": buffer.getvalue(), "path": None}
              else:
                  return {"bytes": None, "path": value.get('''path'''					)}
          elif value.get('''bytes'''					) is not None or value.get('''path'''					) is not None:
              # store the audio bytes, and path is used to infer the audio format using the file extension
              return {"bytes": value.get('''bytes'''					), "path": value.get('''path'''					)}
          else:
              raise ValueError(
                  f"""An audio sample should have one of 'path' or 'bytes' but they are missing or None in {value}."""					)






      def lowercase_				(   self							:       Tuple		,		_A							:       dict		,		_A							:       Optional[Dict[str, Union[str, bool, None]]] = None					):


          '''simple docstring'''


          if not self.decode:
              raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.'''					)

          UpperCAmelCase__		,		UpperCAmelCase__     :						List[str]						=							(value['''path'''], BytesIO(value['''bytes''']					)) if value['''bytes'''] is not None else (value['''path'''], None)
          if path is None and file is None:
              raise ValueError(f"""An audio sample should have one of 'path' or 'bytes' but both are None in {value}."""					)

          try:
              import librosa
              import soundfile as sf
          except ImportError as err:
              raise ImportError('''To support decoding audio files, please install \'librosa\' and \'soundfile\'.'''					) from err

          UpperCAmelCase__     :						Optional[Any]						=							xsplitext(_A					)[1][1:].lower() if path is not None else None
          if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
              raise RuntimeError(
                  '''Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, '''
                  '''You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '''					)
          elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
              raise RuntimeError(
                  '''Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, '''
                  '''You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '''					)

          if file is None:
              UpperCAmelCase__     :						Tuple						=							token_per_repo_id or {}
              UpperCAmelCase__     :						Optional[Any]						=							path.split('''::'''					)[-1]
              try:
                  UpperCAmelCase__     :						int						=							string_to_dict(_A		,		config.HUB_DATASETS_URL					)['''repo_id''']
                  UpperCAmelCase__     :						Union[str, Any]						=							token_per_repo_id[repo_id]
              except (ValueError, KeyError):
                  UpperCAmelCase__     :						List[str]						=							None

              with xopen(_A		,		'''rb'''		,		use_auth_token=_A					) as f:
                  UpperCAmelCase__		,		UpperCAmelCase__     :						str						=							sf.read(_A					)

          else:
              UpperCAmelCase__		,		UpperCAmelCase__     :						str						=							sf.read(_A					)

          UpperCAmelCase__     :						int						=							array.T
          if self.mono:
              UpperCAmelCase__     :						str						=							librosa.to_mono(_A					)
          if self.sampling_rate and self.sampling_rate != sampling_rate:
              UpperCAmelCase__     :						Optional[Any]						=							librosa.resample(_A		,		orig_sr=_A		,		target_sr=self.sampling_rate					)
              UpperCAmelCase__     :						int						=							self.sampling_rate

          return {"path": path, "array": array, "sampling_rate": sampling_rate}






      def lowercase_				(   self							:       Dict					):


          '''simple docstring'''


          from .features import Value

          if self.decode:
              raise ValueError('''Cannot flatten a decoded Audio feature.'''					)
          return {
              "bytes": Value('''binary'''					),
              "path": Value('''string'''					),
          }






      def lowercase_				(   self							:       str		,		_A							:       Union[pa.StringArray, pa.StructArray]					):


          '''simple docstring'''


          if pa.types.is_string(storage.type					):
              UpperCAmelCase__     :						Optional[Any]						=							pa.array([None] * len(_A					)		,		type=pa.binary()					)
              UpperCAmelCase__     :						List[Any]						=							pa.StructArray.from_arrays([bytes_array, storage]		,		['''bytes''', '''path''']		,		mask=storage.is_null()					)
          elif pa.types.is_binary(storage.type					):
              UpperCAmelCase__     :						Optional[int]						=							pa.array([None] * len(_A					)		,		type=pa.string()					)
              UpperCAmelCase__     :						int						=							pa.StructArray.from_arrays([storage, path_array]		,		['''bytes''', '''path''']		,		mask=storage.is_null()					)
          elif pa.types.is_struct(storage.type					) and storage.type.get_all_field_indices('''array'''					):
              UpperCAmelCase__     :						List[Any]						=							pa.array([Audio().encode_example(_A					) if x is not None else None for x in storage.to_pylist()]					)
          elif pa.types.is_struct(storage.type					):
              if storage.type.get_field_index('''bytes'''					) >= 0:
                  UpperCAmelCase__     :						Optional[Any]						=							storage.field('''bytes'''					)
              else:
                  UpperCAmelCase__     :						Any						=							pa.array([None] * len(_A					)		,		type=pa.binary()					)
              if storage.type.get_field_index('''path'''					) >= 0:
                  UpperCAmelCase__     :						List[Any]						=							storage.field('''path'''					)
              else:
                  UpperCAmelCase__     :						List[str]						=							pa.array([None] * len(_A					)		,		type=pa.string()					)
              UpperCAmelCase__     :						Tuple						=							pa.StructArray.from_arrays([bytes_array, path_array]		,		['''bytes''', '''path''']		,		mask=storage.is_null()					)
          return array_cast(_A		,		self.pa_type					)






      def lowercase_				(   self							:       Tuple		,		_A							:       pa.StructArray					):


          '''simple docstring'''


          @no_op_if_value_is_null
          def path_to_bytes(_A							:       Optional[Any]					):
              with xopen(_A		,		'''rb'''					) as f:
                  UpperCAmelCase__     :						List[str]						=							f.read()
              return bytes_

          UpperCAmelCase__     :						Any						=							pa.array(
              [
                  (path_to_bytes(x['''path''']					) if x['''bytes'''] is None else x['''bytes''']) if x is not None else None
                  for x in storage.to_pylist()
              ]		,		type=pa.binary()		,		)
          UpperCAmelCase__     :						Tuple						=							pa.array(
              [os.path.basename(_A					) if path is not None else None for path in storage.field('''path'''					).to_pylist()]		,		type=pa.string()		,		)
          UpperCAmelCase__     :						List[str]						=							pa.StructArray.from_arrays([bytes_array, path_array]		,		['''bytes''', '''path''']		,		mask=bytes_array.is_null()					)
          return array_cast(_A		,		self.pa_type					)




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 				A       (			unittest.TestCase				):
					'''simple docstring'''






					A__     =					MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING

					def     lowerCamelCase__	(self       :		List[str] ,   _UpperCAmelCase       :		List[str] ,   _UpperCAmelCase       :		Dict ,   _UpperCAmelCase       :		Tuple						)		->  Tuple:




							"""simple docstring"""


							lowercase__	     =						hf_hub_download(
							    repo_id="""nateraw/video-demo""" ,   filename="""archery.mp4""" ,   repo_type="""dataset"""						)
							lowercase__	     =						VideoClassificationPipeline(model=_UpperCAmelCase ,   image_processor=_UpperCAmelCase ,   top_k=2						)
							lowercase__	     =						[
							    example_video_filepath,
							    """https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4""",
							]
							return video_classifier, examples

					def     lowerCamelCase__	(self       :		List[str] ,   _UpperCAmelCase       :		Union[str, Any] ,   _UpperCAmelCase       :		Tuple						)		->  List[Any]:




							"""simple docstring"""


							for example in examples:
									lowercase__	     =						video_classifier(_UpperCAmelCase						)

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

					@require_torch
					def     lowerCamelCase__	(self       :		str						)		->  List[Any]:




							"""simple docstring"""


							lowercase__	     =						"""hf-internal-testing/tiny-random-VideoMAEForVideoClassification"""
							lowercase__	     =						VideoMAEFeatureExtractor(
							    size={"""shortest_edge""": 10} ,   crop_size={"""height""": 10, """width""": 10}						)
							lowercase__	     =						pipeline(
							    """video-classification""" ,   model=_UpperCAmelCase ,   feature_extractor=_UpperCAmelCase ,   frame_sampling_rate=4						)

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

							lowercase__	     =						video_classifier(
							    [
							        video_file_path,
							        video_file_path,
							    ] ,   top_k=2 ,   )
							self.assertEqual(
							    nested_simplify(_UpperCAmelCase ,   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     lowerCamelCase__	(self       :		Any						)		->  Union[str, Any]:




							"""simple docstring"""


							pass

