





'''simple docstring'''







import shutil
import tempfile
import unittest

import pytest

from transformers import WhisperTokenizer, is_speech_available
from transformers.testing_utils import require_sentencepiece, require_torch, require_torchaudio

from .test_feature_extraction_whisper import floats_list


if is_speech_available():
					from transformers import WhisperFeatureExtractor, WhisperProcessor


snake_case_  :							Any            =	5_03_58
snake_case_  :							Optional[Any]            =	5_03_62


@require_torch
@require_torchaudio
@require_sentencepiece
class        __a   (unittest.TestCase  ):

							def      UpperCAmelCase__						(				self				:  Dict )       ->							int:







													"""simple docstring"""
													UpperCAmelCase_      :   int      =	'openai/whisper-small.en'
													UpperCAmelCase_      :   Optional[Any]      =	tempfile.mkdtemp()

							def      UpperCAmelCase__						(				self				:  Tuple							,  **__magic_name__				:  Optional[Any] )       ->							int:







													"""simple docstring"""
													return WhisperTokenizer.from_pretrained(self.checkpoint							,  **_lowerCamelCase )

							def      UpperCAmelCase__						(				self				:  int							,  **__magic_name__				:  List[str] )       ->							List[str]:







													"""simple docstring"""
													return WhisperFeatureExtractor.from_pretrained(self.checkpoint							,  **_lowerCamelCase )

							def      UpperCAmelCase__						(				self				:  str )       ->							str:







													"""simple docstring"""
													shutil.rmtree(self.tmpdirname )

							def      UpperCAmelCase__						(				self				:  Any )       ->							str:







													"""simple docstring"""
													UpperCAmelCase_      :   Optional[Any]      =	self.get_tokenizer()
													UpperCAmelCase_      :   Tuple      =	self.get_feature_extractor()

													UpperCAmelCase_      :   List[str]      =	WhisperProcessor(tokenizer=_lowerCamelCase							,  feature_extractor=_lowerCamelCase )

													processor.save_pretrained(self.tmpdirname )
													UpperCAmelCase_      :   Any      =	WhisperProcessor.from_pretrained(self.tmpdirname )

													self.assertEqual(processor.tokenizer.get_vocab()							,  tokenizer.get_vocab() )
													self.assertIsInstance(processor.tokenizer							,  _lowerCamelCase )

													self.assertEqual(processor.feature_extractor.to_json_string()							,  feature_extractor.to_json_string() )
													self.assertIsInstance(processor.feature_extractor							,  _lowerCamelCase )

							def      UpperCAmelCase__						(				self				:  Union[str, Any] )       ->							str:







													"""simple docstring"""
													UpperCAmelCase_      :   int      =	WhisperProcessor(tokenizer=self.get_tokenizer()							,  feature_extractor=self.get_feature_extractor() )
													processor.save_pretrained(self.tmpdirname )

													UpperCAmelCase_      :   Optional[int]      =	self.get_tokenizer(bos_token='''(BOS)'''							,  eos_token='''(EOS)''' )
													UpperCAmelCase_      :   Tuple      =	self.get_feature_extractor(do_normalize=_lowerCamelCase							,  padding_value=1.0 )

													UpperCAmelCase_      :   Optional[int]      =	WhisperProcessor.from_pretrained(
													    self.tmpdirname							,  bos_token='''(BOS)'''							,  eos_token='''(EOS)'''							,  do_normalize=_lowerCamelCase							,  padding_value=1.0 )

													self.assertEqual(processor.tokenizer.get_vocab()							,  tokenizer_add_kwargs.get_vocab() )
													self.assertIsInstance(processor.tokenizer							,  _lowerCamelCase )

													self.assertEqual(processor.feature_extractor.to_json_string()							,  feature_extractor_add_kwargs.to_json_string() )
													self.assertIsInstance(processor.feature_extractor							,  _lowerCamelCase )

							def      UpperCAmelCase__						(				self				:  Dict )       ->							int:







													"""simple docstring"""
													UpperCAmelCase_      :   Dict      =	self.get_feature_extractor()
													UpperCAmelCase_      :   Optional[Any]      =	self.get_tokenizer()

													UpperCAmelCase_      :   Optional[Any]      =	WhisperProcessor(tokenizer=_lowerCamelCase							,  feature_extractor=_lowerCamelCase )

													UpperCAmelCase_      :   Optional[Any]      =	floats_list((3, 10_00) )

													UpperCAmelCase_      :   Any      =	feature_extractor(_lowerCamelCase							,  return_tensors='''np''' )
													UpperCAmelCase_      :   List[str]      =	processor(_lowerCamelCase							,  return_tensors='''np''' )

													for key in input_feat_extract.keys():
																			self.assertAlmostEqual(input_feat_extract[key].sum()							,  input_processor[key].sum()							,  delta=1E-2 )

							def      UpperCAmelCase__						(				self				:  Optional[Any] )       ->							int:







													"""simple docstring"""
													UpperCAmelCase_      :   Any      =	self.get_feature_extractor()
													UpperCAmelCase_      :   Any      =	self.get_tokenizer()

													UpperCAmelCase_      :   Union[str, Any]      =	WhisperProcessor(tokenizer=_lowerCamelCase							,  feature_extractor=_lowerCamelCase )

													UpperCAmelCase_      :   Union[str, Any]      =	'This is a test string'

													UpperCAmelCase_      :   Dict      =	processor(text=_lowerCamelCase )

													UpperCAmelCase_      :   Union[str, Any]      =	tokenizer(_lowerCamelCase )

													for key in encoded_tok.keys():
																			self.assertListEqual(encoded_tok[key]							,  encoded_processor[key] )

							def      UpperCAmelCase__						(				self				:  Dict )       ->							Optional[Any]:







													"""simple docstring"""
													UpperCAmelCase_      :   List[Any]      =	self.get_feature_extractor()
													UpperCAmelCase_      :   Optional[Any]      =	self.get_tokenizer()

													UpperCAmelCase_      :   Union[str, Any]      =	WhisperProcessor(tokenizer=_lowerCamelCase							,  feature_extractor=_lowerCamelCase )

													UpperCAmelCase_      :   Union[str, Any]      =	[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

													UpperCAmelCase_      :   Any      =	processor.batch_decode(_lowerCamelCase )
													UpperCAmelCase_      :   List[str]      =	tokenizer.batch_decode(_lowerCamelCase )

													self.assertListEqual(_lowerCamelCase							,  _lowerCamelCase )

							def      UpperCAmelCase__						(				self				:  List[str] )       ->							str:







													"""simple docstring"""
													UpperCAmelCase_      :   List[Any]      =	self.get_feature_extractor()
													UpperCAmelCase_      :   Dict      =	self.get_tokenizer()

													UpperCAmelCase_      :   List[str]      =	WhisperProcessor(tokenizer=_lowerCamelCase							,  feature_extractor=_lowerCamelCase )

													self.assertListEqual(
													    processor.model_input_names							,  feature_extractor.model_input_names							,  msg='''`processor` and `feature_extractor` model input names do not match'''							,  )

							def      UpperCAmelCase__						(				self				:  str )       ->							Optional[Any]:







													"""simple docstring"""
													UpperCAmelCase_      :   str      =	self.get_feature_extractor()
													UpperCAmelCase_      :   Dict      =	self.get_tokenizer()

													UpperCAmelCase_      :   Optional[Any]      =	WhisperProcessor(tokenizer=_lowerCamelCase							,  feature_extractor=_lowerCamelCase )
													UpperCAmelCase_      :   int      =	processor.get_decoder_prompt_ids(task='''transcribe'''							,  no_timestamps=_lowerCamelCase )

													self.assertIsInstance(_lowerCamelCase							,  _lowerCamelCase )
													for ids in forced_decoder_ids:
																			self.assertIsInstance(_lowerCamelCase							,  (list, tuple) )

													UpperCAmelCase_      :   Tuple      =	[TRANSCRIBE, NOTIMESTAMPS]
													self.assertListEqual([ids[-1] for ids in forced_decoder_ids]							,  _lowerCamelCase )

							def      UpperCAmelCase__						(				self				:  Union[str, Any] )       ->							Optional[int]:







													"""simple docstring"""
													UpperCAmelCase_      :   Dict      =	WhisperProcessor(tokenizer=self.get_tokenizer()							,  feature_extractor=self.get_feature_extractor() )
													UpperCAmelCase_      :   Tuple      =	processor.get_prompt_ids('''Mr. Quilter''' )
													UpperCAmelCase_      :   List[str]      =	processor.tokenizer.decode(_lowerCamelCase )

													self.assertListEqual(prompt_ids.tolist()							,  [5_03_60, 17_70, 13, 22_64, 3_46, 3_53] )
													self.assertEqual(_lowerCamelCase							,  '''<|startofprev|> Mr. Quilter''' )

							def      UpperCAmelCase__						(				self				:  Union[str, Any] )       ->							Tuple:







													"""simple docstring"""
													UpperCAmelCase_      :   Dict      =	WhisperProcessor(tokenizer=self.get_tokenizer()							,  feature_extractor=self.get_feature_extractor() )
													UpperCAmelCase_      :   Dict      =	processor.get_prompt_ids('''''' )
													UpperCAmelCase_      :   Tuple      =	processor.tokenizer.decode(_lowerCamelCase )

													self.assertListEqual(prompt_ids.tolist()							,  [5_03_60, 2_20] )
													self.assertEqual(_lowerCamelCase							,  '''<|startofprev|> ''' )



							def      UpperCAmelCase__						(				self				:  Dict )       ->							str:







													"""simple docstring"""
													UpperCAmelCase_      :   Dict      =	WhisperProcessor(tokenizer=self.get_tokenizer()							,  feature_extractor=self.get_feature_extractor() )

													def _test_prompt_error_raised_helper(__magic_name__				:  List[str]							,  __magic_name__				:  Union[str, Any] ):
																			with pytest.raises(_lowerCamelCase ) as excinfo:
																									processor.get_prompt_ids(_lowerCamelCase )
																			UpperCAmelCase_      :   List[Any]      =	F"""Encountered text in the prompt corresponding to disallowed special token: {special_token}."""
																			self.assertEqual(_lowerCamelCase							,  str(excinfo.value ) )

													_test_prompt_error_raised_helper('''<|startofprev|> test'''							,  '''<|startofprev|>''' )
													_test_prompt_error_raised_helper('''test <|notimestamps|>'''							,  '''<|notimestamps|>''' )
													_test_prompt_error_raised_helper('''test <|zh|> test <|transcribe|>'''							,  '''<|zh|>''' )






'''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			(  _a						:						int					,  _a						:						bool = False      ):




    """simple docstring"""



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







def 			_UpperCamelCase			(  _a						:						Optional[RNGType] = None					,  _a						:						Optional[torch.Generator] = None      ):




    """simple docstring"""



    if rng_type == RNGType.TORCH:
        __UpperCamelCase						:						Union[str, Any]		     =	torch.get_rng_state()
    elif rng_type == RNGType.CUDA:
        __UpperCamelCase						:						int		     =	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						:						Optional[Any]		     =	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						:						str		     =	torch.xpu.get_rng_state()
    elif rng_type == RNGType.GENERATOR:
        assert generator is not None, "Need a generator to synchronize its seed."
        __UpperCamelCase						:						List[Any]		     =	generator.get_state()

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

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







def 			_UpperCamelCase			(  _a						:						List[Union[str, RNGType]]					,  _a						:						Optional[torch.Generator] = None      ):




    """simple docstring"""



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



