

import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map


def   __SCREAMING_SNAKE_CASE					(SCREAMING_SNAKE_CASE__       ):
						snake_case_   =	[]
						if isinstance(__A  ,		__A       ):
												for v in tree.values():
																		shapes.extend(_fetch_dims(__A       )       )
						elif isinstance(__A  ,		(list, tuple)       ):
												for t in tree:
																		shapes.extend(_fetch_dims(__A       )       )
						elif isinstance(__A  ,		torch.Tensor       ):
												shapes.append(tree.shape       )
						else:
												raise ValueError('''Not supported'''       )

						return shapes


@torch.jit.ignore
def   __SCREAMING_SNAKE_CASE					(SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__       ):
						snake_case_   =	[]
						for d in reversed(__A       ):
												idx.append(flat_idx % d       )
												snake_case_   =	flat_idx // d

						return tuple(reversed(__A       )       )


@torch.jit.ignore
def   __SCREAMING_SNAKE_CASE					(SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__ = None  ,		SCREAMING_SNAKE_CASE__ = None  ,		):



						def reduce_edge_list(SCREAMING_SNAKE_CASE__       ) -> None:
												snake_case_   =	True
												for i in range(len(__A       )       ):
																		snake_case_   =	-1 * (i + 1)
																		l[reversed_idx] &= tally
																		snake_case_   =	l[reversed_idx]

						if start_edges is None:
												snake_case_   =	[s == 0 for s in start]
												reduce_edge_list(__A       )
						if end_edges is None:
												snake_case_   =	[e == (d - 1) for e, d in zip(__A  ,		__A       )]
												reduce_edge_list(__A       )

						# Base cases. Either start/end are empty and we're done, or the final,
						# one-dimensional tensor can be simply sliced
						if len(__A       ) == 0:
												return [()]
						elif len(__A       ) == 1:
												return [(slice(start[0]  ,		end[0] + 1       ),)]

						snake_case_   =	[]
						snake_case_   =	[]

						# Dimensions common to start and end can be selected directly
						for s, e in zip(__A  ,		__A       ):
												if s == e:
																		path_list.append(slice(__A  ,		s + 1       )       )
												else:
																		break

						snake_case_   =	tuple(__A       )
						snake_case_   =	len(__A       )

						# start == end, and we're done
						if divergence_idx == len(__A       ):
												return [path]

						def upper() -> Tuple[Tuple[slice, ...], ...]:
												assert start_edges is not None
												assert end_edges is not None

												snake_case_   =	start[divergence_idx]
												return tuple(
												    path + (slice(__A  ,		sdi + 1       ),) + s
												    for s in _get_minimal_slice_set(
												        start[divergence_idx + 1 :]  ,		[d - 1 for d in dims[divergence_idx + 1 :]]  ,		dims[divergence_idx + 1 :]  ,		start_edges=start_edges[divergence_idx + 1 :]  ,		end_edges=[True for _ in end_edges[divergence_idx + 1 :]]  ,		)       )

						def lower() -> Tuple[Tuple[slice, ...], ...]:
												assert start_edges is not None
												assert end_edges is not None

												snake_case_   =	end[divergence_idx]
												return tuple(
												    path + (slice(__A  ,		edi + 1       ),) + s
												    for s in _get_minimal_slice_set(
												        [0 for _ in start[divergence_idx + 1 :]]  ,		end[divergence_idx + 1 :]  ,		dims[divergence_idx + 1 :]  ,		start_edges=[True for _ in start_edges[divergence_idx + 1 :]]  ,		end_edges=end_edges[divergence_idx + 1 :]  ,		)       )

						# If both start and end are at the edges of the subtree rooted at
						# divergence_idx, we can just select the whole subtree at once
						if start_edges[divergence_idx] and end_edges[divergence_idx]:
												slices.append(path + (slice(start[divergence_idx]  ,		end[divergence_idx] + 1       ),)       )
						# If just start is at the edge, we can grab almost all of the subtree,
						# treating only the ragged bottom edge as an edge case
						elif start_edges[divergence_idx]:
												slices.append(path + (slice(start[divergence_idx]  ,		end[divergence_idx]       ),)       )
												slices.extend(lower()       )
						# Analogous to the previous case, but the top is ragged this time
						elif end_edges[divergence_idx]:
												slices.extend(upper()       )
												slices.append(path + (slice(start[divergence_idx] + 1  ,		end[divergence_idx] + 1       ),)       )
						# If both sides of the range are ragged, we need to handle both sides
						# separately. If there's contiguous meat in between them, we can index it
						# in one big chunk
						else:
												slices.extend(upper()       )
												snake_case_   =	end[divergence_idx] - start[divergence_idx]
												if middle_ground > 1:
																		slices.append(path + (slice(start[divergence_idx] + 1  ,		end[divergence_idx]       ),)       )
												slices.extend(lower()       )

						return slices


@torch.jit.ignore
def   __SCREAMING_SNAKE_CASE					(SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__       ):
						snake_case_   =	t.shape[:no_batch_dims]
						snake_case_   =	list(_flat_idx_to_idx(__A  ,		__A       )       )
						# _get_minimal_slice_set is inclusive
						snake_case_   =	list(_flat_idx_to_idx(flat_end - 1  ,		__A       )       )

						# Get an ordered list of slices to perform
						snake_case_   =	_get_minimal_slice_set(
						    __A  ,		__A  ,		__A  ,		)

						snake_case_   =	[t[s] for s in slices]

						return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]       ) for s in sliced_tensors]       )


def   __SCREAMING_SNAKE_CASE					(SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__ = False  ,		SCREAMING_SNAKE_CASE__ = None  ,		SCREAMING_SNAKE_CASE__ = False  ,		):



						if not (len(__A       ) > 0):
												raise ValueError('''Must provide at least one input'''       )

						snake_case_   =	[shape[:no_batch_dims] for shape in _fetch_dims(__A       )]
						snake_case_   =	tuple([max(__A       ) for s in zip(*__A       )]       )

						def _prep_inputs(SCREAMING_SNAKE_CASE__       ) -> torch.Tensor:
												if not low_mem:
																		if not sum(t.shape[:no_batch_dims]       ) == no_batch_dims:
																								snake_case_   =	t.expand(orig_batch_dims + t.shape[no_batch_dims:]       )
																		snake_case_   =	t.reshape(-1  ,		*t.shape[no_batch_dims:]       )
												else:
																		snake_case_   =	t.expand(orig_batch_dims + t.shape[no_batch_dims:]       )
												return t

						snake_case_   =	tensor_tree_map(_prep_inputs  ,		__A       )
						snake_case_   =	None
						if _out is not None:
												snake_case_   =	tensor_tree_map(lambda SCREAMING_SNAKE_CASE__       : t.view([-1] + list(t.shape[no_batch_dims:]       )       )  ,		_out       )

						snake_case_   =	1
						for d in orig_batch_dims:
												flat_batch_dim *= d

						snake_case_   =	flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

						def _select_chunk(SCREAMING_SNAKE_CASE__       ) -> torch.Tensor:
												return t[i : i + chunk_size] if t.shape[0] != 1 else t

						snake_case_   =	0
						snake_case_   =	prepped_outputs
						for _ in range(__A       ):
												# Chunk the input
												if not low_mem:
																		snake_case_   =	_select_chunk
												else:
																		snake_case_   =	partial(
																		    _chunk_slice  ,		flat_start=__A  ,		flat_end=min(__A  ,		i + chunk_size       )  ,		no_batch_dims=len(__A       )  ,		)

												snake_case_   =	tensor_tree_map(__A  ,		__A       )

												# Run the layer on the chunk
												snake_case_   =	layer(**__A       )

												# Allocate space for the output
												if out is None:
																		snake_case_   =	tensor_tree_map(lambda SCREAMING_SNAKE_CASE__       : t.new_zeros((flat_batch_dim,) + t.shape[1:]       )  ,		__A       )

												# Put the chunk in its pre-allocated space
												if isinstance(__A  ,		__A       ):

																		def assign(SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__       ) -> None:
																								for k, v in da.items():
																														if isinstance(__A  ,		__A       ):
																																				assign(__A  ,		da[k]       )
																														else:
																																				if _add_into_out:
																																										v[i : i + chunk_size] += da[k]
																																				else:
																																										snake_case_   =	da[k]

																		assign(__A  ,		__A       )
												elif isinstance(__A  ,		__A       ):
																		for xa, xa in zip(__A  ,		__A       ):
																								if _add_into_out:
																														xa[i : i + chunk_size] += xa
																								else:
																														snake_case_   =	xa
												elif isinstance(__A  ,		torch.Tensor       ):
																		if _add_into_out:
																								out[i : i + chunk_size] += output_chunk
																		else:
																								snake_case_   =	output_chunk
												else:
																		raise ValueError('''Not supported'''       )

												i += chunk_size

						snake_case_   =	tensor_tree_map(lambda SCREAMING_SNAKE_CASE__       : t.view(orig_batch_dims + t.shape[1:]       )  ,		__A       )

						return out





class 	snake_case_   :







			'''simple docstring'''

			def __init__(     self :				Dict							,   _UpperCamelCase :				List[str] = 5_1_2							,   )      ->List[str]:
									snake_case_   =	max_chunk_size
									snake_case_   =	None
									snake_case_   =	None







			def 				snake_case__(     self :				Union[str, Any]							,   _UpperCamelCase :				Optional[Any]							,   _UpperCamelCase :				Union[str, Any]							,   _UpperCamelCase :				Tuple							)      ->int:
									logging.info('''Tuning chunk size...'''							)

									if min_chunk_size >= self.max_chunk_size:
															return min_chunk_size

									snake_case_   =	[2**l for l in range(int(math.log(self.max_chunk_size							,   2							)							) + 1							)]
									snake_case_   =	[c for c in candidates if c > min_chunk_size]
									snake_case_   =	[min_chunk_size] + candidates
									candidates[-1] += 4

									def test_chunk_size(_UpperCamelCase :				Tuple							) -> bool:
															try:
																					with torch.no_grad():
																											fn(*_UpperCamelCase							,   chunk_size=_UpperCamelCase							)
																					return True
															except RuntimeError:
																					return False

									snake_case_   =	0
									snake_case_   =	len(_UpperCamelCase							) - 1
									while i > min_viable_chunk_size_index:
															snake_case_   =	test_chunk_size(candidates[i]							)
															if not viable:
																					snake_case_   =	(min_viable_chunk_size_index + i) // 2
															else:
																					snake_case_   =	i
																					snake_case_   =	(i + len(_UpperCamelCase							) - 1) // 2

									return candidates[min_viable_chunk_size_index]







			def 				snake_case__(     self :				Tuple							,   _UpperCamelCase :				List[str]							,   _UpperCamelCase :				int							)      ->bool:
									snake_case_   =	True
									for aa, aa in zip(_UpperCamelCase							,   _UpperCamelCase							):
															assert type(_UpperCamelCase							) == type(_UpperCamelCase							)
															if isinstance(_UpperCamelCase							,   (list, tuple)							):
																					consistent &= self._compare_arg_caches(_UpperCamelCase							,   _UpperCamelCase							)
															elif isinstance(_UpperCamelCase							,   _UpperCamelCase							):
																					snake_case_   =	[v for _, v in sorted(aa.items()							,   key=lambda _UpperCamelCase							: x[0]							)]
																					snake_case_   =	[v for _, v in sorted(aa.items()							,   key=lambda _UpperCamelCase							: x[0]							)]
																					consistent &= self._compare_arg_caches(_UpperCamelCase							,   _UpperCamelCase							)
															else:
																					consistent &= aa == aa

									return consistent







			def 				snake_case__(     self :				Dict							,   _UpperCamelCase :				Tuple							,   _UpperCamelCase :				int							,   _UpperCamelCase :				List[str]							,   )      ->int:
									snake_case_   =	True
									snake_case_   =	tree_map(lambda _UpperCamelCase							: a.shape if isinstance(_UpperCamelCase							,   torch.Tensor							) else a							,   _UpperCamelCase							,   _UpperCamelCase							)
									if self.cached_arg_data is not None:
															# If args have changed shape/value, we need to re-tune
															assert len(self.cached_arg_data							) == len(_UpperCamelCase							)
															snake_case_   =	self._compare_arg_caches(self.cached_arg_data							,   _UpperCamelCase							)
									else:
															# Otherwise, we can reuse the precomputed value
															snake_case_   =	False

									if not consistent:
															snake_case_   =	self._determine_favorable_chunk_size(
															    _UpperCamelCase							,   _UpperCamelCase							,   _UpperCamelCase							,   )
															snake_case_   =	arg_data

									assert self.cached_chunk_size is not None

									return self.cached_chunk_size




'''simple docstring'''





import warnings
from contextlib import contextmanager

from ....processing_utils import ProcessorMixin






class __A			(  A						):



 '''simple docstring'''







 __lowerCamelCase    :       Optional[Any]          =       'MCTCTFeatureExtractor'
 __lowerCamelCase    :       Optional[Any]          =       'AutoTokenizer'



 def __init__(self ,   A ,   A       )    ->     Dict:


      """simple docstring"""



      super().__init__(A ,   A       )
      _a      =  self.feature_extractor
      _a      =  False



 def __call__(self ,   *A ,   **A       )    ->     Optional[int]:


      """simple docstring"""



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

      if "raw_speech" in kwargs:
           warnings.warn('''Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.'''       )
           _a      =  kwargs.pop('''raw_speech'''       )
      else:
           _a      =  kwargs.pop('''audio''' ,   A       )
      _a      =  kwargs.pop('''sampling_rate''' ,   A       )
      _a      =  kwargs.pop('''text''' ,   A       )
      if len(A       ) > 0:
           _a      =  args[0]
           _a      =  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:
           _a      =  self.feature_extractor(A ,   *A ,   sampling_rate=A ,   **A       )
      if text is not None:
           _a      =  self.tokenizer(A ,   **A       )

      if text is None:
           return inputs
      elif audio is None:
           return encodings
      else:
           _a      =  encodings['''input_ids''']
           return inputs



 def 					a__  (self ,   *A ,   **A       )    ->     int:


      """simple docstring"""



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



 def 					a__  (self ,   *A ,   **A       )    ->     Tuple:


      """simple docstring"""



      if self._in_target_context_manager:
           return self.current_processor.pad(*A ,   **A       )

      _a      =  kwargs.pop('''input_features''' ,   A       )
      _a      =  kwargs.pop('''labels''' ,   A       )
      if len(A       ) > 0:
           _a      =  args[0]
           _a      =  args[1:]

      if input_features is not None:
           _a      =  self.feature_extractor.pad(A ,   *A ,   **A       )
      if labels is not None:
           _a      =  self.tokenizer.pad(A ,   **A       )

      if labels is None:
           return input_features
      elif input_features is None:
           return labels
      else:
           _a      =  labels['''input_ids''']
           return input_features



 def 					a__  (self ,   *A ,   **A       )    ->     List[str]:


      """simple docstring"""



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



 @contextmanager
 def 					a__  (self       )    ->     Dict:


      """simple docstring"""



      warnings.warn(
          '''`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '''
          '''labels by using the argument `text` of the regular `__call__` method (either in the same call as '''
          '''your audio inputs, or in a separate call.'''       )
      _a      =  True
      _a      =  self.tokenizer
      yield
      _a      =  self.feature_extractor
      _a      =  False



