


'''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_		:   int							   =		False, False, False





@dataclass
class     lowercase__  :





							'''simple docstring'''


							A_							:      Optional[int]      	=  None
							A_							:      bool      	=  True
							A_							:      bool      	=  True
							A_							:      Optional[str]      	=  None
							# Automatically constructed
							A_							:      ClassVar[str]      	=  "dict"
							A_							:      ClassVar[Any]      	=  pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}					)
							A_							:      str      	=  field(default="""Audio"""     ,     init=_snake_case     ,     repr=_snake_case					)






							def __call__(    self  ):
											return self.pa_type






							def UpperCAmelCase_	(    self    ,					__snake_case  ):
											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(__snake_case    ,					__snake_case  ):
															return {"bytes": None, "path": value}
											elif isinstance(__snake_case    ,					__snake_case  ):
															return {"bytes": value, "path": None}
											elif "array" in value:
															# convert the audio array to wav bytes
															_SCREAMING_SNAKE_CASE :       List[Any]     		=      BytesIO()
															sf.write(__snake_case    ,					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!)
																							_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      np.frombuffer(value["""bytes"""]    ,					dtype=np.intaa  ).astype(np.floataa  ) / 3_2767
																			else:
																							_SCREAMING_SNAKE_CASE :       int     		=      np.memmap(value["""path"""]    ,					dtype="""h"""    ,					mode="""r"""  ).astype(np.floataa  ) / 3_2767

																			_SCREAMING_SNAKE_CASE :       List[str]     		=      BytesIO(bytes()  )
																			sf.write(__snake_case    ,					__snake_case    ,					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 UpperCAmelCase_	(    self    ,					__snake_case    ,					__snake_case = None  ):
											if not self.decode:
															raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead."""  )

											_SCREAMING_SNAKE_CASE		,       _SCREAMING_SNAKE_CASE :       List[Any]     		=      (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

											_SCREAMING_SNAKE_CASE :       Any     		=      xsplitext(__snake_case  )[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:
															_SCREAMING_SNAKE_CASE :       Any     		=      token_per_repo_id or {}
															_SCREAMING_SNAKE_CASE :       List[Any]     		=      path.split("""::"""  )[-1]
															try:
																			_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      string_to_dict(__snake_case    ,					config.HUB_DATASETS_URL  )["""repo_id"""]
																			_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      token_per_repo_id[repo_id]
															except (ValueError, KeyError):
																			_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      None

															with xopen(__snake_case    ,					"""rb"""    ,					use_auth_token=__snake_case  ) as f:
																			_SCREAMING_SNAKE_CASE		,       _SCREAMING_SNAKE_CASE :       Optional[Any]     		=      sf.read(__snake_case  )

											else:
															_SCREAMING_SNAKE_CASE		,       _SCREAMING_SNAKE_CASE :       List[Any]     		=      sf.read(__snake_case  )

											_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      array.T
											if self.mono:
															_SCREAMING_SNAKE_CASE :       Dict     		=      librosa.to_mono(__snake_case  )
											if self.sampling_rate and self.sampling_rate != sampling_rate:
															_SCREAMING_SNAKE_CASE :       Optional[int]     		=      librosa.resample(__snake_case    ,					orig_sr=__snake_case    ,					target_sr=self.sampling_rate  )
															_SCREAMING_SNAKE_CASE :       List[str]     		=      self.sampling_rate

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






							def UpperCAmelCase_	(    self  ):
											from .features import Value

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






							def UpperCAmelCase_	(    self    ,					__snake_case  ):
											if pa.types.is_string(storage.type  ):
															_SCREAMING_SNAKE_CASE :       str     		=      pa.array([None] * len(__snake_case  )    ,					type=pa.binary()  )
															_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      pa.StructArray.from_arrays([bytes_array, storage]    ,					["""bytes""", """path"""]    ,					mask=storage.is_null()  )
											elif pa.types.is_binary(storage.type  ):
															_SCREAMING_SNAKE_CASE :       Any     		=      pa.array([None] * len(__snake_case  )    ,					type=pa.string()  )
															_SCREAMING_SNAKE_CASE :       str     		=      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"""  ):
															_SCREAMING_SNAKE_CASE :       str     		=      pa.array([Audio().encode_example(__snake_case  ) 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:
																			_SCREAMING_SNAKE_CASE :       Any     		=      storage.field("""bytes"""  )
															else:
																			_SCREAMING_SNAKE_CASE :       str     		=      pa.array([None] * len(__snake_case  )    ,					type=pa.binary()  )
															if storage.type.get_field_index("""path"""  ) >= 0:
																			_SCREAMING_SNAKE_CASE :       Dict     		=      storage.field("""path"""  )
															else:
																			_SCREAMING_SNAKE_CASE :       List[Any]     		=      pa.array([None] * len(__snake_case  )    ,					type=pa.string()  )
															_SCREAMING_SNAKE_CASE :       Dict     		=      pa.StructArray.from_arrays([bytes_array, path_array]    ,					["""bytes""", """path"""]    ,					mask=storage.is_null()  )
											return array_cast(__snake_case    ,					self.pa_type  )






							def UpperCAmelCase_	(    self    ,					__snake_case  ):

											@no_op_if_value_is_null
											def path_to_bytes(__snake_case  ):
															with xopen(__snake_case    ,					"""rb"""  ) as f:
																			_SCREAMING_SNAKE_CASE :       Tuple     		=      f.read()
															return bytes_

											_SCREAMING_SNAKE_CASE :       str     		=      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()    ,					)
											_SCREAMING_SNAKE_CASE :       List[str]     		=      pa.array(
											    [os.path.basename(__snake_case  ) if path is not None else None for path in storage.field("""path"""  ).to_pylist()]    ,					type=pa.string()    ,					)
											_SCREAMING_SNAKE_CASE :       Optional[int]     		=      pa.StructArray.from_arrays([bytes_array, path_array]    ,					["""bytes""", """path"""]    ,					mask=bytes_array.is_null()  )
											return array_cast(__snake_case    ,					self.pa_type  )








'''simple docstring'''


import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
				import torch

if is_vision_available():
				from PIL import Image

				from transformers import MobileNetVaImageProcessor





class     lowercase__  (						unittest.TestCase					):





							'''simple docstring'''


							def __init__(    self    ,					__snake_case    ,					__snake_case=7    ,					__snake_case=3    ,					__snake_case=18    ,					__snake_case=30    ,					__snake_case=400    ,					__snake_case=True    ,					__snake_case=None    ,					__snake_case=True    ,					__snake_case=None    ,					):
											_SCREAMING_SNAKE_CASE :       Any     		=      size if size is not None else {"""shortest_edge""": 20}
											_SCREAMING_SNAKE_CASE :       int     		=      crop_size if crop_size is not None else {"""height""": 18, """width""": 18}
											_SCREAMING_SNAKE_CASE :       Dict     		=      parent
											_SCREAMING_SNAKE_CASE :       Any     		=      batch_size
											_SCREAMING_SNAKE_CASE :       Any     		=      num_channels
											_SCREAMING_SNAKE_CASE :       Dict     		=      image_size
											_SCREAMING_SNAKE_CASE :       Any     		=      min_resolution
											_SCREAMING_SNAKE_CASE :       Any     		=      max_resolution
											_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      do_resize
											_SCREAMING_SNAKE_CASE :       Tuple     		=      size
											_SCREAMING_SNAKE_CASE :       Optional[int]     		=      do_center_crop
											_SCREAMING_SNAKE_CASE :       str     		=      crop_size






							def UpperCAmelCase_	(    self  ):
											return {
											    "do_resize": self.do_resize,
											    "size": self.size,
											    "do_center_crop": self.do_center_crop,
											    "crop_size": self.crop_size,
											}





@require_torch
@require_vision
class     lowercase__  (						_snake_case     ,     unittest.TestCase					):





							'''simple docstring'''


							A_							:      Union[str, Any]      	=  MobileNetVaImageProcessor if is_vision_available() else None






							def UpperCAmelCase_	(    self  ):
											_SCREAMING_SNAKE_CASE :       int     		=      MobileNetVaImageProcessingTester(self  )






							@property
							def UpperCAmelCase_	(    self  ):
											return self.image_processor_tester.prepare_image_processor_dict()






							def UpperCAmelCase_	(    self  ):
											_SCREAMING_SNAKE_CASE :       str     		=      self.image_processing_class(**self.image_processor_dict  )
											self.assertTrue(hasattr(__snake_case    ,					"""do_resize"""  )  )
											self.assertTrue(hasattr(__snake_case    ,					"""size"""  )  )
											self.assertTrue(hasattr(__snake_case    ,					"""do_center_crop"""  )  )
											self.assertTrue(hasattr(__snake_case    ,					"""crop_size"""  )  )






							def UpperCAmelCase_	(    self  ):
											_SCREAMING_SNAKE_CASE :       str     		=      self.image_processing_class.from_dict(self.image_processor_dict  )
											self.assertEqual(image_processor.size    ,					{"""shortest_edge""": 20}  )
											self.assertEqual(image_processor.crop_size    ,					{"""height""": 18, """width""": 18}  )

											_SCREAMING_SNAKE_CASE :       Dict     		=      self.image_processing_class.from_dict(self.image_processor_dict    ,					size=42    ,					crop_size=84  )
											self.assertEqual(image_processor.size    ,					{"""shortest_edge""": 42}  )
											self.assertEqual(image_processor.crop_size    ,					{"""height""": 84, """width""": 84}  )






							def UpperCAmelCase_	(    self  ):
											pass






							def UpperCAmelCase_	(    self  ):
											# Initialize image_processing
											_SCREAMING_SNAKE_CASE :       str     		=      self.image_processing_class(**self.image_processor_dict  )
											# create random PIL images
											_SCREAMING_SNAKE_CASE :       List[str]     		=      prepare_image_inputs(self.image_processor_tester    ,					equal_resolution=__snake_case  )
											for image in image_inputs:
															self.assertIsInstance(__snake_case    ,					Image.Image  )

											# Test not batched input
											_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      image_processing(image_inputs[0]    ,					return_tensors="""pt"""  ).pixel_values
											self.assertEqual(
											    encoded_images.shape    ,					(
											        1,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["""height"""],
											        self.image_processor_tester.crop_size["""width"""],
											    )    ,					)

											# Test batched
											_SCREAMING_SNAKE_CASE :       int     		=      image_processing(__snake_case    ,					return_tensors="""pt"""  ).pixel_values
											self.assertEqual(
											    encoded_images.shape    ,					(
											        self.image_processor_tester.batch_size,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["""height"""],
											        self.image_processor_tester.crop_size["""width"""],
											    )    ,					)






							def UpperCAmelCase_	(    self  ):
											# Initialize image_processing
											_SCREAMING_SNAKE_CASE :       Optional[int]     		=      self.image_processing_class(**self.image_processor_dict  )
											# create random numpy tensors
											_SCREAMING_SNAKE_CASE :       Tuple     		=      prepare_image_inputs(self.image_processor_tester    ,					equal_resolution=__snake_case    ,					numpify=__snake_case  )
											for image in image_inputs:
															self.assertIsInstance(__snake_case    ,					np.ndarray  )

											# Test not batched input
											_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      image_processing(image_inputs[0]    ,					return_tensors="""pt"""  ).pixel_values
											self.assertEqual(
											    encoded_images.shape    ,					(
											        1,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["""height"""],
											        self.image_processor_tester.crop_size["""width"""],
											    )    ,					)

											# Test batched
											_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      image_processing(__snake_case    ,					return_tensors="""pt"""  ).pixel_values
											self.assertEqual(
											    encoded_images.shape    ,					(
											        self.image_processor_tester.batch_size,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["""height"""],
											        self.image_processor_tester.crop_size["""width"""],
											    )    ,					)






							def UpperCAmelCase_	(    self  ):
											# Initialize image_processing
											_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      self.image_processing_class(**self.image_processor_dict  )
											# create random PyTorch tensors
											_SCREAMING_SNAKE_CASE :       str     		=      prepare_image_inputs(self.image_processor_tester    ,					equal_resolution=__snake_case    ,					torchify=__snake_case  )
											for image in image_inputs:
															self.assertIsInstance(__snake_case    ,					torch.Tensor  )

											# Test not batched input
											_SCREAMING_SNAKE_CASE :       List[Any]     		=      image_processing(image_inputs[0]    ,					return_tensors="""pt"""  ).pixel_values
											self.assertEqual(
											    encoded_images.shape    ,					(
											        1,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["""height"""],
											        self.image_processor_tester.crop_size["""width"""],
											    )    ,					)

											# Test batched
											_SCREAMING_SNAKE_CASE :       str     		=      image_processing(__snake_case    ,					return_tensors="""pt"""  ).pixel_values
											self.assertEqual(
											    encoded_images.shape    ,					(
											        self.image_processor_tester.batch_size,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["""height"""],
											        self.image_processor_tester.crop_size["""width"""],
											    )    ,					)




