


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

import numpy as np






def 						a__			(				A__,      A__				):
			SCREAMING_SNAKE_CASE_		:     Optional[int]					=					F'''{sampling_rate}'''
			SCREAMING_SNAKE_CASE_		:     str					=					'1'
			SCREAMING_SNAKE_CASE_		:     Optional[Any]					=					'f32le'
			SCREAMING_SNAKE_CASE_		:     str					=					[
			    'ffmpeg',
			    '-i',
			    'pipe:0',
			    '-ac',
			    ac,
			    '-ar',
			    ar,
			    '-f',
			    format_for_conversion,
			    '-hide_banner',
			    '-loglevel',
			    'quiet',
			    'pipe:1',
			]

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






def 						a__			(				A__,      A__,      A__ = "f32le",      ):
			SCREAMING_SNAKE_CASE_		:     int					=					F'''{sampling_rate}'''
			SCREAMING_SNAKE_CASE_		:     Any					=					'1'
			if format_for_conversion == "s16le":
						SCREAMING_SNAKE_CASE_		:     Tuple					=					2
			elif format_for_conversion == "f32le":
						SCREAMING_SNAKE_CASE_		:     Optional[int]					=					4
			else:
						raise ValueError(F'''Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'''				)

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

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






def 						a__			(				A__,      A__,      A__ = None,      A__ = None,      A__ = "f32le",      ):
			if stream_chunk_s is not None:
						SCREAMING_SNAKE_CASE_		:     int					=					stream_chunk_s
			else:
						SCREAMING_SNAKE_CASE_		:     Any					=					chunk_length_s

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

			if stride_length_s is None:
						SCREAMING_SNAKE_CASE_		:     Any					=					chunk_length_s / 6
			SCREAMING_SNAKE_CASE_		:     Any					=					int(round(sampling_rate * chunk_length_s				)				) * size_of_sample
			if isinstance(A__,      (int, float)				):
						SCREAMING_SNAKE_CASE_		:     Dict					=					[stride_length_s, stride_length_s]

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






def 						a__			(				A__,      A__,      A__,      A__ = False				):
			SCREAMING_SNAKE_CASE_		:     str					=					B''
			SCREAMING_SNAKE_CASE_		,      SCREAMING_SNAKE_CASE_		:     str					=					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}'''				)
			SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					0
			for raw in iterator:
						acc += raw
						if stream and len(A__				) < chunk_len:
									SCREAMING_SNAKE_CASE_		:     List[Any]					=					(_stride_left, 0)
									yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
						else:
									while len(A__				) >= chunk_len:
												# We are flushing the accumulator
												SCREAMING_SNAKE_CASE_		:     Optional[int]					=					(_stride_left, stride_right)
												SCREAMING_SNAKE_CASE_		:     int					=					{'raw': acc[:chunk_len], 'stride': stride}
												if stream:
															SCREAMING_SNAKE_CASE_		:     Dict					=					False
												yield item
												SCREAMING_SNAKE_CASE_		:     int					=					stride_left
												SCREAMING_SNAKE_CASE_		:     Any					=					acc[chunk_len - stride_left - stride_right :]
    # Last chunk
			if len(A__				) > stride_left:
						SCREAMING_SNAKE_CASE_		:     int					=					{'raw': acc, 'stride': (_stride_left, 0)}
						if stream:
									SCREAMING_SNAKE_CASE_		:     Optional[int]					=					False
						yield item






def 						a__			(				A__,      A__				):
			SCREAMING_SNAKE_CASE_		:     Tuple					=					2**2_4  # 16Mo
			try:
						with subprocess.Popen(A__,      stdout=subprocess.PIPE,      bufsize=A__				) as ffmpeg_process:
									while True:
												SCREAMING_SNAKE_CASE_		:     str					=					ffmpeg_process.stdout.read(A__				)
												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









from ...processing_utils import ProcessorMixin







class 	__lowercase    (__SCREAMING_SNAKE_CASE  ):





							"""simple docstring"""







							_UpperCAmelCase      =						"""WhisperFeatureExtractor"""
							_UpperCAmelCase      =						"""WhisperTokenizer"""






							def __init__(      self			,						lowerCAmelCase__			,						lowerCAmelCase__       ):





										"""simple docstring"""




										super().__init__(lowerCAmelCase__			,						lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					self.feature_extractor
										SCREAMING_SNAKE_CASE_		:     List[Any]					=					False






							def 					UpperCamelCase__     (      self			,						lowerCAmelCase__=None			,						lowerCAmelCase__=None			,						lowerCAmelCase__=True       ):





										"""simple docstring"""




										return self.tokenizer.get_decoder_prompt_ids(task=lowerCAmelCase__			,						language=lowerCAmelCase__			,						no_timestamps=lowerCAmelCase__       )






							def __call__(      self			,						*lowerCAmelCase__			,						**lowerCAmelCase__       ):





										"""simple docstring"""




										if self._in_target_context_manager:
													return self.current_processor(*lowerCAmelCase__			,						**lowerCAmelCase__       )

										SCREAMING_SNAKE_CASE_		:     Tuple					=					kwargs.pop('audio'			,						lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     str					=					kwargs.pop('sampling_rate'			,						lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Optional[Any]					=					kwargs.pop('text'			,						lowerCAmelCase__       )
										if len(lowerCAmelCase__       ) > 0:
													SCREAMING_SNAKE_CASE_		:     List[Any]					=					args[0]
													SCREAMING_SNAKE_CASE_		:     List[Any]					=					args[1:]

										if audio is None and text is None:
													raise ValueError('You need to specify either an `audio` or `text` input to process.'       )

										if audio is not None:
													SCREAMING_SNAKE_CASE_		:     List[Any]					=					self.feature_extractor(lowerCAmelCase__			,						*lowerCAmelCase__			,						sampling_rate=lowerCAmelCase__			,						**lowerCAmelCase__       )
										if text is not None:
													SCREAMING_SNAKE_CASE_		:     Optional[int]					=					self.tokenizer(lowerCAmelCase__			,						**lowerCAmelCase__       )

										if text is None:
													return inputs

										elif audio is None:
													return encodings
										else:
													SCREAMING_SNAKE_CASE_		:     Optional[int]					=					encodings['input_ids']
													return inputs






							def 					UpperCamelCase__     (      self			,						*lowerCAmelCase__			,						**lowerCAmelCase__       ):





										"""simple docstring"""




										return self.tokenizer.batch_decode(*lowerCAmelCase__			,						**lowerCAmelCase__       )






							def 					UpperCamelCase__     (      self			,						*lowerCAmelCase__			,						**lowerCAmelCase__       ):





										"""simple docstring"""




										return self.tokenizer.decode(*lowerCAmelCase__			,						**lowerCAmelCase__       )






							def 					UpperCamelCase__     (      self			,						lowerCAmelCase__			,						lowerCAmelCase__="np"       ):





										"""simple docstring"""




										return self.tokenizer.get_prompt_ids(lowerCAmelCase__			,						return_tensors=lowerCAmelCase__       )





