




"""simple docstring"""






import importlib
import json
import os
from collections import OrderedDict
from typing import Dict, Optional, Union

# Build the list of all feature extractors
from ...configuration_utils import PretrainedConfig
from ...dynamic_module_utils import get_class_from_dynamic_module, resolve_trust_remote_code
from ...feature_extraction_utils import FeatureExtractionMixin
from ...utils import CONFIG_NAME, FEATURE_EXTRACTOR_NAME, get_file_from_repo, logging
from .auto_factory import _LazyAutoMapping
from .configuration_auto import (
    CONFIG_MAPPING_NAMES,
    AutoConfig,
    model_type_to_module_name,
    replace_list_option_in_docstrings,
)


lowercase__     						=						logging.get_logger(__name__)

lowercase__     						=						OrderedDict(
    [
        ("""audio-spectrogram-transformer""", """ASTFeatureExtractor"""),
        ("""beit""", """BeitFeatureExtractor"""),
        ("""chinese_clip""", """ChineseCLIPFeatureExtractor"""),
        ("""clap""", """ClapFeatureExtractor"""),
        ("""clip""", """CLIPFeatureExtractor"""),
        ("""clipseg""", """ViTFeatureExtractor"""),
        ("""conditional_detr""", """ConditionalDetrFeatureExtractor"""),
        ("""convnext""", """ConvNextFeatureExtractor"""),
        ("""cvt""", """ConvNextFeatureExtractor"""),
        ("""data2vec-audio""", """Wav2Vec2FeatureExtractor"""),
        ("""data2vec-vision""", """BeitFeatureExtractor"""),
        ("""deformable_detr""", """DeformableDetrFeatureExtractor"""),
        ("""deit""", """DeiTFeatureExtractor"""),
        ("""detr""", """DetrFeatureExtractor"""),
        ("""dinat""", """ViTFeatureExtractor"""),
        ("""donut-swin""", """DonutFeatureExtractor"""),
        ("""dpt""", """DPTFeatureExtractor"""),
        ("""encodec""", """EncodecFeatureExtractor"""),
        ("""flava""", """FlavaFeatureExtractor"""),
        ("""glpn""", """GLPNFeatureExtractor"""),
        ("""groupvit""", """CLIPFeatureExtractor"""),
        ("""hubert""", """Wav2Vec2FeatureExtractor"""),
        ("""imagegpt""", """ImageGPTFeatureExtractor"""),
        ("""layoutlmv2""", """LayoutLMv2FeatureExtractor"""),
        ("""layoutlmv3""", """LayoutLMv3FeatureExtractor"""),
        ("""levit""", """LevitFeatureExtractor"""),
        ("""maskformer""", """MaskFormerFeatureExtractor"""),
        ("""mctct""", """MCTCTFeatureExtractor"""),
        ("""mobilenet_v1""", """MobileNetV1FeatureExtractor"""),
        ("""mobilenet_v2""", """MobileNetV2FeatureExtractor"""),
        ("""mobilevit""", """MobileViTFeatureExtractor"""),
        ("""nat""", """ViTFeatureExtractor"""),
        ("""owlvit""", """OwlViTFeatureExtractor"""),
        ("""perceiver""", """PerceiverFeatureExtractor"""),
        ("""poolformer""", """PoolFormerFeatureExtractor"""),
        ("""regnet""", """ConvNextFeatureExtractor"""),
        ("""resnet""", """ConvNextFeatureExtractor"""),
        ("""segformer""", """SegformerFeatureExtractor"""),
        ("""sew""", """Wav2Vec2FeatureExtractor"""),
        ("""sew-d""", """Wav2Vec2FeatureExtractor"""),
        ("""speech_to_text""", """Speech2TextFeatureExtractor"""),
        ("""speecht5""", """SpeechT5FeatureExtractor"""),
        ("""swiftformer""", """ViTFeatureExtractor"""),
        ("""swin""", """ViTFeatureExtractor"""),
        ("""swinv2""", """ViTFeatureExtractor"""),
        ("""table-transformer""", """DetrFeatureExtractor"""),
        ("""timesformer""", """VideoMAEFeatureExtractor"""),
        ("""tvlt""", """TvltFeatureExtractor"""),
        ("""unispeech""", """Wav2Vec2FeatureExtractor"""),
        ("""unispeech-sat""", """Wav2Vec2FeatureExtractor"""),
        ("""van""", """ConvNextFeatureExtractor"""),
        ("""videomae""", """VideoMAEFeatureExtractor"""),
        ("""vilt""", """ViltFeatureExtractor"""),
        ("""vit""", """ViTFeatureExtractor"""),
        ("""vit_mae""", """ViTFeatureExtractor"""),
        ("""vit_msn""", """ViTFeatureExtractor"""),
        ("""wav2vec2""", """Wav2Vec2FeatureExtractor"""),
        ("""wav2vec2-conformer""", """Wav2Vec2FeatureExtractor"""),
        ("""wavlm""", """Wav2Vec2FeatureExtractor"""),
        ("""whisper""", """WhisperFeatureExtractor"""),
        ("""xclip""", """CLIPFeatureExtractor"""),
        ("""yolos""", """YolosFeatureExtractor"""),
    ]
)

lowercase__     						=						_LazyAutoMapping(CONFIG_MAPPING_NAMES, FEATURE_EXTRACTOR_MAPPING_NAMES)







def      _snake_case			(				lowercase__		):
				for module_name, extractors in FEATURE_EXTRACTOR_MAPPING_NAMES.items():
								if class_name in extractors:
												_lowerCamelCase     :			Optional[int]    =    model_type_to_module_name(lowercase__		)

												_lowerCamelCase     :			List[str]    =    importlib.import_module(f'''.{module_name}'''							,    'transformers.models'		)
												try:
																return getattr(lowercase__							,    lowercase__		)
												except AttributeError:
																continue

				for _, extractor in FEATURE_EXTRACTOR_MAPPING._extra_content.items():
								if getattr(lowercase__							,    '__name__'							,    lowercase__		) == class_name:
												return extractor

    # We did not fine the class, but maybe it's because a dep is missing. In that case, the class will be in the main
    # init and we return the proper dummy to get an appropriate error message.
				_lowerCamelCase     :			Tuple    =    importlib.import_module('transformers'		)
				if hasattr(lowercase__							,    lowercase__		):
								return getattr(lowercase__							,    lowercase__		)

				return None







def      _snake_case			(				lowercase__							,    lowercase__ = None							,    lowercase__ = False							,    lowercase__ = False							,    lowercase__ = None							,    lowercase__ = None							,    lowercase__ = None							,    lowercase__ = False							,    **lowercase__							,    ):
				_lowerCamelCase     :			Dict    =    get_file_from_repo(
				    lowercase__							,    lowercase__							,    cache_dir=lowercase__							,    force_download=lowercase__							,    resume_download=lowercase__							,    proxies=lowercase__							,    use_auth_token=lowercase__							,    revision=lowercase__							,    local_files_only=lowercase__							,    )
				if resolved_config_file is None:
								logger.info(
								    'Could not locate the feature extractor configuration file, will try to use the model config instead.'		)
								return {}

				with open(lowercase__							,    encoding='utf-8'		) as reader:
								return json.load(lowercase__		)






class lowerCAmelCase__	:






			'''simple docstring'''



			def __init__(  self  ):
							raise EnvironmentError(
							    'AutoFeatureExtractor is designed to be instantiated '
							    'using the `AutoFeatureExtractor.from_pretrained(pretrained_model_name_or_path)` method.'  )
			@classmethod
			@replace_list_option_in_docstrings(lowercase  )
			def 					A_    (  cls  ,      lowercase  ,      **lowercase  ):
							_lowerCamelCase     :			List[Any]    =    kwargs.pop('config'  ,      lowercase  )
							_lowerCamelCase     :			int    =    kwargs.pop('trust_remote_code'  ,      lowercase  )
							_lowerCamelCase     :			Any    =    True

							_lowerCamelCase, _lowerCamelCase     :			Tuple    =    FeatureExtractionMixin.get_feature_extractor_dict(lowercase  ,      **lowercase  )
							_lowerCamelCase     :			str    =    config_dict.get('feature_extractor_type'  ,      lowercase  )
							_lowerCamelCase     :			Optional[Any]    =    None
							if "AutoFeatureExtractor" in config_dict.get('auto_map'  ,      {}  ):
											_lowerCamelCase     :			Union[str, Any]    =    config_dict['auto_map']['AutoFeatureExtractor']

							# If we don't find the feature extractor class in the feature extractor config, let's try the model config.
							if feature_extractor_class is None and feature_extractor_auto_map is None:
											if not isinstance(lowercase  ,      lowercase  ):
															_lowerCamelCase     :			List[str]    =    AutoConfig.from_pretrained(lowercase  ,      **lowercase  )
											# It could be in `config.feature_extractor_type``
											_lowerCamelCase     :			Dict    =    getattr(lowercase  ,      'feature_extractor_type'  ,      lowercase  )
											if hasattr(lowercase  ,      'auto_map'  ) and "AutoFeatureExtractor" in config.auto_map:
															_lowerCamelCase     :			int    =    config.auto_map['AutoFeatureExtractor']

							if feature_extractor_class is not None:
											_lowerCamelCase     :			int    =    feature_extractor_class_from_name(lowercase  )

							_lowerCamelCase     :			List[str]    =    feature_extractor_auto_map is not None
							_lowerCamelCase     :			Union[str, Any]    =    feature_extractor_class is not None or type(lowercase  ) in FEATURE_EXTRACTOR_MAPPING
							_lowerCamelCase     :			Dict    =    resolve_trust_remote_code(
							    lowercase  ,      lowercase  ,      lowercase  ,      lowercase  )

							if has_remote_code and trust_remote_code:
											_lowerCamelCase     :			Optional[Any]    =    get_class_from_dynamic_module(
											    lowercase  ,      lowercase  ,      **lowercase  )
											_lowerCamelCase     :			List[Any]    =    kwargs.pop('code_revision'  ,      lowercase  )
											if os.path.isdir(lowercase  ):
															feature_extractor_class.register_for_auto_class()
											return feature_extractor_class.from_dict(lowercase  ,      **lowercase  )
							elif feature_extractor_class is not None:
											return feature_extractor_class.from_dict(lowercase  ,      **lowercase  )
							# Last try: we use the FEATURE_EXTRACTOR_MAPPING.
							elif type(lowercase  ) in FEATURE_EXTRACTOR_MAPPING:
											_lowerCamelCase     :			Tuple    =    FEATURE_EXTRACTOR_MAPPING[type(lowercase  )]
											return feature_extractor_class.from_dict(lowercase  ,      **lowercase  )

							raise ValueError(
							    F'''Unrecognized feature extractor in {pretrained_model_name_or_path}. Should have a '''
							    F'''`feature_extractor_type` key in its {FEATURE_EXTRACTOR_NAME} of {CONFIG_NAME}, or one of the following '''
							    F'''`model_type` keys in its {CONFIG_NAME}: {', '.join(c for c in FEATURE_EXTRACTOR_MAPPING_NAMES.keys()  )}'''  )






			@staticmethod
			def 					A_    (  lowercase  ,      lowercase  ):
							FEATURE_EXTRACTOR_MAPPING.register(lowercase  ,      lowercase  )





"""simple docstring"""






import datetime
import platform
import subprocess
from typing import Optional, Tuple, Union

import numpy as np







def      _snake_case			(				lowercase__							,    lowercase__		):
				_lowerCamelCase     :			Tuple    =    f'''{sampling_rate}'''
				_lowerCamelCase     :			str    =    '1'
				_lowerCamelCase     :			str    =    'f32le'
				_lowerCamelCase     :			Union[str, Any]    =    [
				    'ffmpeg',
				    '-i',
				    'pipe:0',
				    '-ac',
				    ac,
				    '-ar',
				    ar,
				    '-f',
				    format_for_conversion,
				    '-hide_banner',
				    '-loglevel',
				    'quiet',
				    'pipe:1',
				]

				try:
								with subprocess.Popen(lowercase__							,    stdin=subprocess.PIPE							,    stdout=subprocess.PIPE		) as ffmpeg_process:
												_lowerCamelCase     :			str    =    ffmpeg_process.communicate(lowercase__		)
				except FileNotFoundError as error:
								raise ValueError('ffmpeg was not found but is required to load audio files from filename'		) from error
				_lowerCamelCase     :			List[Any]    =    output_stream[0]
				_lowerCamelCase     :			Tuple    =    np.frombuffer(lowercase__							,    np.floataa		)
				if audio.shape[0] == 0:
								raise ValueError('Malformed soundfile'		)
				return audio







def      _snake_case			(				lowercase__							,    lowercase__							,    lowercase__ = "f32le"							,    ):
				_lowerCamelCase     :			Optional[Any]    =    f'''{sampling_rate}'''
				_lowerCamelCase     :			List[str]    =    '1'
				if format_for_conversion == "s16le":
								_lowerCamelCase     :			List[str]    =    2
				elif format_for_conversion == "f32le":
								_lowerCamelCase     :			List[Any]    =    4
				else:
								raise ValueError(f'''Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'''		)

				_lowerCamelCase     :			Dict    =    platform.system()
				if system == "Linux":
								_lowerCamelCase     :			Optional[int]    =    'alsa'
								_lowerCamelCase     :			Optional[Any]    =    'default'
				elif system == "Darwin":
								_lowerCamelCase     :			Optional[int]    =    'avfoundation'
								_lowerCamelCase     :			Any    =    ':0'
				elif system == "Windows":
								_lowerCamelCase     :			Tuple    =    'dshow'
								_lowerCamelCase     :			Tuple    =    'default'

				_lowerCamelCase     :			Optional[int]    =    [
				    'ffmpeg',
				    '-f',
				    format_,
				    '-i',
				    input_,
				    '-ac',
				    ac,
				    '-ar',
				    ar,
				    '-f',
				    format_for_conversion,
				    '-fflags',
				    'nobuffer',
				    '-hide_banner',
				    '-loglevel',
				    'quiet',
				    'pipe:1',
				]
				_lowerCamelCase     :			Tuple    =    int(round(sampling_rate * chunk_length_s		)		) * size_of_sample
				_lowerCamelCase     :			List[Any]    =    _ffmpeg_stream(lowercase__							,    lowercase__		)
				for item in iterator:
								yield item







def      _snake_case			(				lowercase__							,    lowercase__							,    lowercase__ = None							,    lowercase__ = None							,    lowercase__ = "f32le"							,    ):
				if stream_chunk_s is not None:
								_lowerCamelCase     :			int    =    stream_chunk_s
				else:
								_lowerCamelCase     :			Optional[Any]    =    chunk_length_s

				_lowerCamelCase     :			Optional[Any]    =    ffmpeg_microphone(lowercase__							,    lowercase__							,    format_for_conversion=lowercase__		)
				if format_for_conversion == "s16le":
								_lowerCamelCase     :			List[str]    =    np.intaa
								_lowerCamelCase     :			str    =    2
				elif format_for_conversion == "f32le":
								_lowerCamelCase     :			Any    =    np.floataa
								_lowerCamelCase     :			List[Any]    =    4
				else:
								raise ValueError(f'''Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'''		)

				if stride_length_s is None:
								_lowerCamelCase     :			Union[str, Any]    =    chunk_length_s / 6
				_lowerCamelCase     :			Optional[int]    =    int(round(sampling_rate * chunk_length_s		)		) * size_of_sample
				if isinstance(lowercase__							,    (int, float)		):
								_lowerCamelCase     :			Any    =    [stride_length_s, stride_length_s]

				_lowerCamelCase     :			Tuple    =    int(round(sampling_rate * stride_length_s[0]		)		) * size_of_sample
				_lowerCamelCase     :			Optional[Any]    =    int(round(sampling_rate * stride_length_s[1]		)		) * size_of_sample
				_lowerCamelCase     :			List[Any]    =    datetime.datetime.now()
				_lowerCamelCase     :			Optional[int]    =    datetime.timedelta(seconds=lowercase__		)
				for item in chunk_bytes_iter(lowercase__							,    lowercase__							,    stride=(stride_left, stride_right)							,    stream=lowercase__		):
								# Put everything back in numpy scale
								_lowerCamelCase     :			List[Any]    =    np.frombuffer(item['raw']							,    dtype=lowercase__		)
								_lowerCamelCase     :			int    =    (
								    item['stride'][0] // size_of_sample,
								    item['stride'][1] // size_of_sample,
								)
								_lowerCamelCase     :			Optional[int]    =    sampling_rate
								audio_time += delta
								if datetime.datetime.now() > audio_time + 10 * delta:
												# We're late !! SKIP
												continue
								yield item







def      _snake_case			(				lowercase__							,    lowercase__							,    lowercase__							,    lowercase__ = False		):
				_lowerCamelCase     :			int    =    B''
				_lowerCamelCase, _lowerCamelCase     :			Dict    =    stride
				if stride_left + stride_right >= chunk_len:
								raise ValueError(
								    f'''Stride needs to be strictly smaller than chunk_len: ({stride_left}, {stride_right}) vs {chunk_len}'''		)
				_lowerCamelCase     :			str    =    0
				for raw in iterator:
								acc += raw
								if stream and len(lowercase__		) < chunk_len:
												_lowerCamelCase     :			Optional[int]    =    (_stride_left, 0)
												yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
								else:
												while len(lowercase__		) >= chunk_len:
																# We are flushing the accumulator
																_lowerCamelCase     :			str    =    (_stride_left, stride_right)
																_lowerCamelCase     :			str    =    {'raw': acc[:chunk_len], 'stride': stride}
																if stream:
																				_lowerCamelCase     :			List[Any]    =    False
																yield item
																_lowerCamelCase     :			Optional[Any]    =    stride_left
																_lowerCamelCase     :			str    =    acc[chunk_len - stride_left - stride_right :]
    # Last chunk
				if len(lowercase__		) > stride_left:
								_lowerCamelCase     :			Optional[Any]    =    {'raw': acc, 'stride': (_stride_left, 0)}
								if stream:
												_lowerCamelCase     :			Tuple    =    False
								yield item







def      _snake_case			(				lowercase__							,    lowercase__		):
				_lowerCamelCase     :			int    =    2**24  # 16Mo
				try:
								with subprocess.Popen(lowercase__							,    stdout=subprocess.PIPE							,    bufsize=lowercase__		) as ffmpeg_process:
												while True:
																_lowerCamelCase     :			Optional[Any]    =    ffmpeg_process.stdout.read(lowercase__		)
																if raw == b"":
																				break
																yield raw
				except FileNotFoundError as error:
								raise ValueError('ffmpeg was not found but is required to stream audio files from filename'		) from error