



"""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


a_	,		a_	,		a_							=    False, False, False







@dataclass
class 					A_:







			"""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    _lowerCAmelCase		(	self				,   A							):
						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
									_lowerCamelCase					:					Union[str, Any]	       =    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!)
															_lowerCamelCase					:					Optional[int]	       =    np.frombuffer(value['bytes']				,   dtype=np.intaa							).astype(np.floataa							) / 3_2767
												else:
															_lowerCamelCase					:					Union[str, Any]	       =    np.memmap(value['path']				,   dtype='h'				,   mode='r'							).astype(np.floataa							) / 3_2767

												_lowerCamelCase					:					Optional[int]	       =    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    _lowerCAmelCase		(	self				,   A				,   A = None							):
						if not self.decode:
									raise RuntimeError('Decoding is disabled for this feature. Please use Audio(decode=True) instead.'							)

						_lowerCamelCase					:					Tuple	       =    (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

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

									with xopen(A				,   'rb'				,   use_auth_token=A							) as f:
												_lowerCamelCase					:					Optional[int]	       =    sf.read(A							)

						else:
									_lowerCamelCase					:					Union[str, Any]	       =    sf.read(A							)

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

						return {"path": path, "array": array, "sampling_rate": sampling_rate}
			def    _lowerCAmelCase		(	self							):
						from .features import Value

						if self.decode:
									raise ValueError('Cannot flatten a decoded Audio feature.'							)
						return {
						    "bytes": Value('binary'							),
						    "path": Value('string'							),
						}
			def    _lowerCAmelCase		(	self				,   A							):
						if pa.types.is_string(storage.type							):
									_lowerCamelCase					:					Tuple	       =    pa.array([None] * len(A							)				,   type=pa.binary()							)
									_lowerCamelCase					:					Optional[int]	       =    pa.StructArray.from_arrays([bytes_array, storage]				,   ['bytes', 'path']				,   mask=storage.is_null()							)
						elif pa.types.is_binary(storage.type							):
									_lowerCamelCase					:					List[Any]	       =    pa.array([None] * len(A							)				,   type=pa.string()							)
									_lowerCamelCase					:					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'							):
									_lowerCamelCase					:					Dict	       =    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:
												_lowerCamelCase					:					Optional[int]	       =    storage.field('bytes'							)
									else:
												_lowerCamelCase					:					str	       =    pa.array([None] * len(A							)				,   type=pa.binary()							)
									if storage.type.get_field_index('path'							) >= 0:
												_lowerCamelCase					:					Optional[int]	       =    storage.field('path'							)
									else:
												_lowerCamelCase					:					Optional[int]	       =    pa.array([None] * len(A							)				,   type=pa.string()							)
									_lowerCamelCase					:					List[str]	       =    pa.StructArray.from_arrays([bytes_array, path_array]				,   ['bytes', 'path']				,   mask=storage.is_null()							)
						return array_cast(A				,   self.pa_type							)

			def    _lowerCAmelCase		(	self				,   A							):

						@no_op_if_value_is_null
						def path_to_bytes(A							):
									with xopen(A				,   'rb'							) as f:
												_lowerCamelCase					:					Tuple	       =    f.read()
									return bytes_

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













"""simple docstring"""







from typing import Dict, List, Optional, Union

import numpy as np

from transformers.utils import is_vision_available
from transformers.utils.generic import TensorType

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    is_valid_image,
    to_numpy_array,
    valid_images,
)
from ...utils import logging


if is_vision_available():
				import PIL

_lowerCAmelCase            =	logging.get_logger(__name__)


def      UpperCamelCase       ( _A )     ->       List[List[ImageInput]]:
				if isinstance(_A  ,				(list, tuple) ) and isinstance(videos[0]  ,				(list, tuple) ) and is_valid_image(videos[0][0] ):
								return videos

				elif isinstance(_A  ,				(list, tuple) ) and is_valid_image(videos[0] ):
								return [videos]

				elif is_valid_image(_A ):
								return [[videos]]

				raise ValueError(F"""Could not make batched video from {videos}""" )


class      UpperCamelCase		(__snake_case     ):
						_SCREAMING_SNAKE_CASE  : str												=		["""pixel_values"""]






						def __init__(		self  :str			, __magic_name__  :bool = True			, __magic_name__  :Dict[str, int] = None			, __magic_name__  :PILImageResampling = PILImageResampling.BILINEAR			, __magic_name__  :bool = True			, __magic_name__  :Dict[str, int] = None			, __magic_name__  :bool = True			, __magic_name__  :Union[int, float] = 1 / 255			, __magic_name__  :bool = True			, __magic_name__  :bool = True			, __magic_name__  :Optional[Union[float, List[float]]] = None			, __magic_name__  :Optional[Union[float, List[float]]] = None			, **__magic_name__  :List[str]			, )      ->None:
										super().__init__(**__magic_name__				)
										lowercase    :      Union[str, Any]       = size if size is not None else {"""shortest_edge""": 256}
										lowercase    :      Tuple       = get_size_dict(__magic_name__			, default_to_square=__magic_name__				)
										lowercase    :      List[Any]       = crop_size if crop_size is not None else {"""height""": 224, """width""": 224}
										lowercase    :      List[str]       = get_size_dict(__magic_name__			, param_name="""crop_size"""				)

										lowercase    :      Tuple       = do_resize
										lowercase    :      Optional[Any]       = size
										lowercase    :      Union[str, Any]       = do_center_crop
										lowercase    :      List[Any]       = crop_size
										lowercase    :      Union[str, Any]       = resample
										lowercase    :      Optional[Any]       = do_rescale
										lowercase    :      int       = rescale_factor
										lowercase    :      Dict       = offset
										lowercase    :      Optional[int]       = do_normalize
										lowercase    :      Optional[Any]       = image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
										lowercase    :      Tuple       = image_std if image_std is not None else IMAGENET_STANDARD_STD






						def      __snake_case      (		self  :Any			, __magic_name__  :np.ndarray			, __magic_name__  :Dict[str, int]			, __magic_name__  :PILImageResampling = PILImageResampling.BILINEAR			, __magic_name__  :Optional[Union[str, ChannelDimension]] = None			, **__magic_name__  :Optional[Any]			, )      ->np.ndarray:
										lowercase    :      Optional[int]       = get_size_dict(__magic_name__			, default_to_square=__magic_name__				)
										if "shortest_edge" in size:
														lowercase    :      int       = get_resize_output_image_size(__magic_name__			, size["""shortest_edge"""]			, default_to_square=__magic_name__				)
										elif "height" in size and "width" in size:
														lowercase    :      Tuple       = (size["""height"""], size["""width"""])
										else:
														raise ValueError(f"""Size must have 'height' and 'width' or 'shortest_edge' as keys. Got {size.keys()}"""				)
										return resize(__magic_name__			, size=__magic_name__			, resample=__magic_name__			, data_format=__magic_name__			, **__magic_name__				)






						def      __snake_case      (		self  :Optional[Any]			, __magic_name__  :np.ndarray			, __magic_name__  :Dict[str, int]			, __magic_name__  :Optional[Union[str, ChannelDimension]] = None			, **__magic_name__  :str			, )      ->np.ndarray:
										lowercase    :      Any       = get_size_dict(__magic_name__				)
										if "height" not in size or "width" not in size:
														raise ValueError(f"""Size must have 'height' and 'width' as keys. Got {size.keys()}"""				)
										return center_crop(__magic_name__			, size=(size["""height"""], size["""width"""])			, data_format=__magic_name__			, **__magic_name__				)






						def      __snake_case      (		self  :Tuple			, __magic_name__  :np.ndarray			, __magic_name__  :Union[int, float]			, __magic_name__  :bool = True			, __magic_name__  :Optional[Union[str, ChannelDimension]] = None			, **__magic_name__  :Optional[Any]			, )      ->Any:
										lowercase    :      Optional[int]       = image.astype(np.floataa				)
										if offset:
														lowercase    :      Optional[int]       = image - (scale / 2)
										return rescale(__magic_name__			, scale=__magic_name__			, data_format=__magic_name__			, **__magic_name__				)






						def      __snake_case      (		self  :int			, __magic_name__  :np.ndarray			, __magic_name__  :Union[float, List[float]]			, __magic_name__  :Union[float, List[float]]			, __magic_name__  :Optional[Union[str, ChannelDimension]] = None			, **__magic_name__  :Union[str, Any]			, )      ->np.ndarray:
										return normalize(__magic_name__			, mean=__magic_name__			, std=__magic_name__			, data_format=__magic_name__			, **__magic_name__				)






						def      __snake_case      (		self  :List[Any]			, __magic_name__  :ImageInput			, __magic_name__  :bool = None			, __magic_name__  :Dict[str, int] = None			, __magic_name__  :PILImageResampling = None			, __magic_name__  :bool = None			, __magic_name__  :Dict[str, int] = None			, __magic_name__  :bool = None			, __magic_name__  :float = None			, __magic_name__  :bool = None			, __magic_name__  :bool = None			, __magic_name__  :Optional[Union[float, List[float]]] = None			, __magic_name__  :Optional[Union[float, List[float]]] = None			, __magic_name__  :Optional[ChannelDimension] = ChannelDimension.FIRST			, )      ->np.ndarray:
										if do_resize and size is None or resample is None:
														raise ValueError("""Size and resample must be specified if do_resize is True."""				)

										if do_center_crop and crop_size is None:
														raise ValueError("""Crop size must be specified if do_center_crop is True."""				)

										if do_rescale and rescale_factor is None:
														raise ValueError("""Rescale factor must be specified if do_rescale is True."""				)

										if do_normalize and (image_mean is None or image_std is None):
														raise ValueError("""Image mean and std must be specified if do_normalize is True."""				)

										if offset and not do_rescale:
														raise ValueError("""For offset, do_rescale must also be set to True."""				)

										# All transformations expect numpy arrays.
										lowercase    :      Optional[Any]       = to_numpy_array(__magic_name__				)

										if do_resize:
														lowercase    :      Any       = self.resize(image=__magic_name__			, size=__magic_name__			, resample=__magic_name__				)

										if do_center_crop:
														lowercase    :      Any       = self.center_crop(__magic_name__			, size=__magic_name__				)

										if do_rescale:
														lowercase    :      Optional[int]       = self.rescale(image=__magic_name__			, scale=__magic_name__			, offset=__magic_name__				)

										if do_normalize:
														lowercase    :      Union[str, Any]       = self.normalize(image=__magic_name__			, mean=__magic_name__			, std=__magic_name__				)

										lowercase    :      Optional[Any]       = to_channel_dimension_format(__magic_name__			, __magic_name__				)
										return image






						def      __snake_case      (		self  :Union[str, Any]			, __magic_name__  :ImageInput			, __magic_name__  :bool = None			, __magic_name__  :Dict[str, int] = None			, __magic_name__  :PILImageResampling = None			, __magic_name__  :bool = None			, __magic_name__  :Dict[str, int] = None			, __magic_name__  :bool = None			, __magic_name__  :float = None			, __magic_name__  :bool = None			, __magic_name__  :bool = None			, __magic_name__  :Optional[Union[float, List[float]]] = None			, __magic_name__  :Optional[Union[float, List[float]]] = None			, __magic_name__  :Optional[Union[str, TensorType]] = None			, __magic_name__  :ChannelDimension = ChannelDimension.FIRST			, **__magic_name__  :List[str]			, )      ->PIL.Image.Image:
										lowercase    :      List[Any]       = do_resize if do_resize is not None else self.do_resize
										lowercase    :      Dict       = resample if resample is not None else self.resample
										lowercase    :      Optional[int]       = do_center_crop if do_center_crop is not None else self.do_center_crop
										lowercase    :      List[Any]       = do_rescale if do_rescale is not None else self.do_rescale
										lowercase    :      Tuple       = rescale_factor if rescale_factor is not None else self.rescale_factor
										lowercase    :      Optional[Any]       = offset if offset is not None else self.offset
										lowercase    :      str       = do_normalize if do_normalize is not None else self.do_normalize
										lowercase    :      List[str]       = image_mean if image_mean is not None else self.image_mean
										lowercase    :      Union[str, Any]       = image_std if image_std is not None else self.image_std

										lowercase    :      Optional[int]       = size if size is not None else self.size
										lowercase    :      Optional[Any]       = get_size_dict(__magic_name__			, default_to_square=__magic_name__				)
										lowercase    :      Tuple       = crop_size if crop_size is not None else self.crop_size
										lowercase    :      List[str]       = get_size_dict(__magic_name__			, param_name="""crop_size"""				)

										if not valid_images(__magic_name__				):
														raise ValueError(
														    """Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, """
														    """torch.Tensor, tf.Tensor or jax.ndarray."""				)

										lowercase    :      Tuple       = make_batched(__magic_name__				)

										lowercase    :      List[str]       = [
										    [
										        self._preprocess_image(
										            image=__magic_name__			, do_resize=__magic_name__			, size=__magic_name__			, resample=__magic_name__			, do_center_crop=__magic_name__			, crop_size=__magic_name__			, do_rescale=__magic_name__			, rescale_factor=__magic_name__			, offset=__magic_name__			, do_normalize=__magic_name__			, image_mean=__magic_name__			, image_std=__magic_name__			, data_format=__magic_name__			, )
										        for img in video
										    ]
										    for video in videos
										]

										lowercase    :      Dict       = {"""pixel_values""": videos}
										return BatchFeature(data=__magic_name__			, tensor_type=__magic_name__				)
