






'''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 			UpperCamelCase_							(      _UpperCAmelCase						:   Any    ,			_UpperCAmelCase						:   int = False					)					->					Dict:


       """simple docstring"""







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

def 			UpperCamelCase_							(      _UpperCAmelCase						:   Optional[int] = None    ,			_UpperCAmelCase						:   Tuple = None					)					->					int:


       """simple docstring"""







       if rng_type == RNGType.TORCH:
              _UpperCAmelCase  :						Dict							   =					torch.get_rng_state()
       elif rng_type == RNGType.CUDA:
              _UpperCAmelCase  :						List[Any]							   =					torch.cuda.get_rng_state()
       elif rng_type == RNGType.XLA:
              assert is_tpu_available(), "Can't synchronize XLA seeds on an environment without TPUs."
              _UpperCAmelCase  :						Dict							   =					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."
              _UpperCAmelCase  :						Tuple							   =					torch.npu.get_rng_state()
       elif rng_type == RNGType.XPU:
              assert is_xpu_available(), "Can't synchronize XPU seeds on an environment without XPUs."
              _UpperCAmelCase  :						int							   =					torch.xpu.get_rng_state()
       elif rng_type == RNGType.GENERATOR:
              assert generator is not None, "Need a generator to synchronize its seed."
              _UpperCAmelCase  :						List[str]							   =					generator.get_state()

       # Broadcast the rng state from device 0 to other devices
       _UpperCAmelCase  :						Dict							   =					AcceleratorState()
       if state.distributed_type == DistributedType.TPU:
              _UpperCAmelCase  :						Optional[Any]							   =					rng_state.to(xm.xla_device()					)
              xm.collective_broadcast([rng_state]					)
              xm.mark_step()
              _UpperCAmelCase  :						Optional[Any]							   =					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
       ):
              _UpperCAmelCase  :						Tuple							   =					rng_state.to(state.device					)
              torch.distributed.broadcast(UpperCamelCase_    ,			0					)
              _UpperCAmelCase  :						int							   =					rng_state.cpu()
       elif state.distributed_type == DistributedType.MULTI_CPU:
              torch.distributed.broadcast(UpperCamelCase_    ,			0					)

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


def 			UpperCamelCase_							(      _UpperCAmelCase						:   Optional[int]    ,			_UpperCAmelCase						:   Any = None					)					->					int:


       """simple docstring"""







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


'''simple docstring'''




import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto.configuration_auto import AutoConfig


__SCREAMING_SNAKE_CASE :  Optional[int]  =						logging.get_logger(__name__)

class 			lowerCamelCase_			(snake_case__					):
      '''simple docstring'''

      __UpperCamelCase:						List[Any]      			=		"speech-encoder-decoder"
      __UpperCamelCase:						Optional[int]      			=		True




      def __init__( self		:       Optional[int]		,      **A		:       Tuple ):
             super().__init__(**A )
             if "encoder" not in kwargs or "decoder" not in kwargs:
                    raise ValueError(
                        F"""A configuraton of type {self.model_type} cannot be instantiated because not both `encoder` and"""
                        F""" `decoder` sub-configurations are passed, but only {kwargs}""" )

             _UpperCAmelCase  :						Dict							   =					kwargs.pop("encoder" )
             _UpperCAmelCase  :						Any							   =					encoder_config.pop("model_type" )
             _UpperCAmelCase  :						str							   =					kwargs.pop("decoder" )
             _UpperCAmelCase  :						int							   =					decoder_config.pop("model_type" )

             _UpperCAmelCase  :						List[Any]							   =					AutoConfig.for_model(A		,      **A )
             _UpperCAmelCase  :						str							   =					AutoConfig.for_model(A		,      **A )
             _UpperCAmelCase  :						Optional[int]							   =					True




      @classmethod
      def   _A			( cls		:       int		,      A		:       PretrainedConfig		,      A		:       PretrainedConfig		,      **A		:       Dict ):
             logger.info("Setting `config.is_decoder=True` and `config.add_cross_attention=True` for decoder_config" )
             _UpperCAmelCase  :						Any							   =					True
             _UpperCAmelCase  :						Union[str, Any]							   =					True

             return cls(encoder=encoder_config.to_dict()		,      decoder=decoder_config.to_dict()		,      **A )





      def   _A			( self		:       Tuple ):
             _UpperCAmelCase  :						List[str]							   =					copy.deepcopy(self.__dict__ )
             _UpperCAmelCase  :						Any							   =					self.encoder.to_dict()
             _UpperCAmelCase  :						Union[str, Any]							   =					self.decoder.to_dict()
             _UpperCAmelCase  :						Union[str, Any]							   =					self.__class__.model_type
             return output

