"""simple docstring"""


import random
from typing import List, Optional, Union

import numpy as np
import torch

from ..state import AcceleratorState
from .constants import CUDA_DISTRIBUTED_TYPES
from .dataclasses import DistributedType, RNGType
from .imports import is_npu_available, is_tpu_available, is_xpu_available


if is_tpu_available(check_device=False):
	import torch_xla.core.xla_model as xm





def 			lowerCamelCase_ (					__lowerCAmelCase   ,    __lowerCAmelCase = False				)							->					int:




						'''simple docstring'''



						if device_specific:
												seed += AcceleratorState().process_index
						random.seed(__lowerCAmelCase				)
						np.random.seed(__lowerCAmelCase				)
						torch.manual_seed(__lowerCAmelCase				)
						if is_xpu_available():
												torch.xpu.manual_seed_all(__lowerCAmelCase				)
						elif is_npu_available():
												torch.npu.manual_seed_all(__lowerCAmelCase				)
						else:
												torch.cuda.manual_seed_all(__lowerCAmelCase				)
						# ^^ safe to call this function even if cuda is not available
						if is_tpu_available():
												xm.set_rng_state(__lowerCAmelCase				)





def 			lowerCamelCase_ (					__lowerCAmelCase = None   ,    __lowerCAmelCase = None				)							->					Optional[Any]:




						'''simple docstring'''



						if rng_type == RNGType.TORCH:
												lowerCamelCase__					=torch.get_rng_state()
						elif rng_type == RNGType.CUDA:
												lowerCamelCase__					=torch.cuda.get_rng_state()
						elif rng_type == RNGType.XLA:
												assert is_tpu_available(), "Can't synchronize XLA seeds on an environment without TPUs."
												lowerCamelCase__					=torch.tensor(xm.get_rng_state()				)
						elif rng_type == RNGType.NPU:
												assert is_npu_available(), "Can't synchronize NPU seeds on an environment without NPUs."
												lowerCamelCase__					=torch.npu.get_rng_state()
						elif rng_type == RNGType.XPU:
												assert is_xpu_available(), "Can't synchronize XPU seeds on an environment without XPUs."
												lowerCamelCase__					=torch.xpu.get_rng_state()
						elif rng_type == RNGType.GENERATOR:
												assert generator is not None, "Need a generator to synchronize its seed."
												lowerCamelCase__					=generator.get_state()

						# Broadcast the rng state from device 0 to other devices
						lowerCamelCase__					=AcceleratorState()
						if state.distributed_type == DistributedType.TPU:
												lowerCamelCase__					=rng_state.to(xm.xla_device()				)
												xm.collective_broadcast([rng_state]				)
												xm.mark_step()
												lowerCamelCase__					=rng_state.cpu()
						elif (
						    state.distributed_type in CUDA_DISTRIBUTED_TYPES
						    or state.distributed_type == DistributedType.MULTI_NPU
						    or state.distributed_type == DistributedType.MULTI_XPU
						):
												lowerCamelCase__					=rng_state.to(state.device				)
												torch.distributed.broadcast(__lowerCAmelCase   ,    0				)
												lowerCamelCase__					=rng_state.cpu()
						elif state.distributed_type == DistributedType.MULTI_CPU:
												torch.distributed.broadcast(__lowerCAmelCase   ,    0				)

						# Set the broadcast rng state
						if rng_type == RNGType.TORCH:
												torch.set_rng_state(__lowerCAmelCase				)
						elif rng_type == RNGType.CUDA:
												torch.cuda.set_rng_state(__lowerCAmelCase				)
						elif rng_type == RNGType.NPU:
												torch.npu.set_rng_state(__lowerCAmelCase				)
						elif rng_type == RNGType.XPU:
												torch.xpu.set_rng_state(__lowerCAmelCase				)
						elif rng_type == RNGType.XLA:
												xm.set_rng_state(rng_state.item()				)
						elif rng_type == RNGType.GENERATOR:
												generator.set_state(__lowerCAmelCase				)





def 			lowerCamelCase_ (					__lowerCAmelCase   ,    __lowerCAmelCase = None				)							->					Dict:




						'''simple docstring'''



						for rng_type in rng_types:
												synchronize_rng_state(RNGType(__lowerCAmelCase				)   ,    generator=__lowerCAmelCase				)






"""simple docstring"""


import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin
class snake_case					(     _UpperCamelCase):
		__UpperCamelCase							=	'AutoFeatureExtractor'
		__UpperCamelCase							=	'Speech2Text2Tokenizer'







		def __init__(    self					:				int			,      a__					:				Tuple			,      a__					:				Optional[Any]						)				->  str:





						'''simple docstring'''







						super().__init__(a__			,      a__						)
						_A         =	self.feature_extractor
						_A         =	False







		def __call__(    self					:				Optional[int]			,      *a__					:				List[str]			,      **a__					:				Any						)				->  List[Any]:





						'''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					:				List[Any]			,      *a__					:				Tuple			,      **a__					:				List[Any]						)				->  List[Any]:





						'''simple docstring'''







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







		def        a_       (    self					:				Any			,      *a__					:				Dict			,      **a__					:				Tuple						)				->  Union[str, Any]:





						'''simple docstring'''







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







		@contextmanager
		def        a_       (    self					:				Optional[int]						)				->  Optional[int]:





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