




"""simple docstring"""





import unittest

from transformers import SpeechaTextaConfig
from transformers.testing_utils import is_torch_available, require_torch, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
							import torch

							from transformers.models.speech_to_text_a.modeling_speech_to_text_a import (
							    SpeechaTextaDecoder,
							    SpeechaTextaForCausalLM,
							)

@require_torch
class       __lowercase       :


				'''simple docstring'''


				def __init__(   self   :     int      ,SCREAMING_SNAKE_CASE_   :     Optional[Any]      ,SCREAMING_SNAKE_CASE_   :     Optional[Any]=9_9      ,SCREAMING_SNAKE_CASE_   :     List[str]=1_3      ,SCREAMING_SNAKE_CASE_   :     str=1_6      ,SCREAMING_SNAKE_CASE_   :     Tuple=7      ,SCREAMING_SNAKE_CASE_   :     Union[str, Any]=True      ,SCREAMING_SNAKE_CASE_   :     List[Any]=True      ,SCREAMING_SNAKE_CASE_   :     Tuple=True      ,SCREAMING_SNAKE_CASE_   :     Optional[Any]=False      ,SCREAMING_SNAKE_CASE_   :     Tuple=True      ,SCREAMING_SNAKE_CASE_   :     Dict=2      ,SCREAMING_SNAKE_CASE_   :     Tuple=3_2      ,SCREAMING_SNAKE_CASE_   :     List[str]=4      ,SCREAMING_SNAKE_CASE_   :     str=4      ,SCREAMING_SNAKE_CASE_   :     List[str]=3_0      ,SCREAMING_SNAKE_CASE_   :     Dict=0      ,SCREAMING_SNAKE_CASE_   :     List[Any]=1      ,SCREAMING_SNAKE_CASE_   :     Optional[Any]=2      ,SCREAMING_SNAKE_CASE_   :     Union[str, Any]=None      ,):
										snake_case__       :					Tuple										=					parent
										snake_case__       :					Optional[int]										=					batch_size
										snake_case__       :					str										=					decoder_seq_length
										# For common tests
										snake_case__       :					Any										=					self.decoder_seq_length
										snake_case__       :					Optional[int]										=					is_training
										snake_case__       :					Optional[int]										=					use_attention_mask
										snake_case__       :					Optional[Any]										=					use_labels

										snake_case__       :					Union[str, Any]										=					vocab_size
										snake_case__       :					Any										=					d_model
										snake_case__       :					Tuple										=					d_model
										snake_case__       :					Optional[int]										=					decoder_layers
										snake_case__       :					str										=					decoder_layers
										snake_case__       :					Any										=					decoder_ffn_dim
										snake_case__       :					List[Any]										=					decoder_attention_heads
										snake_case__       :					str										=					decoder_attention_heads
										snake_case__       :					Union[str, Any]										=					eos_token_id
										snake_case__       :					List[Any]										=					bos_token_id
										snake_case__       :					Any										=					pad_token_id
										snake_case__       :					List[str]										=					decoder_start_token_id
										snake_case__       :					Tuple										=					use_cache
										snake_case__       :					Optional[int]										=					max_position_embeddings

										snake_case__       :					str										=					None
										snake_case__       :					Optional[int]										=					decoder_seq_length
										snake_case__       :					Union[str, Any]										=					2
										snake_case__       :					str										=					1

				def       _lowerCamelCase					(   self   :     Union[str, Any]      ):
										snake_case__       :					List[str]										=					ids_tensor([self.batch_size, self.decoder_seq_length]      ,self.vocab_size      )

										snake_case__       :					Union[str, Any]										=					None
										if self.use_attention_mask:
																snake_case__       :					Optional[int]										=					ids_tensor([self.batch_size, self.decoder_seq_length]      ,vocab_size=2      )

										snake_case__       :					List[Any]										=					None
										if self.use_labels:
																snake_case__       :					List[Any]										=					ids_tensor([self.batch_size, self.decoder_seq_length]      ,self.vocab_size      )

										snake_case__       :					Dict										=					SpeechaTextaConfig(
										    vocab_size=self.vocab_size      ,d_model=self.d_model      ,decoder_layers=self.decoder_layers      ,decoder_ffn_dim=self.decoder_ffn_dim      ,decoder_attention_heads=self.decoder_attention_heads      ,eos_token_id=self.eos_token_id      ,bos_token_id=self.bos_token_id      ,use_cache=self.use_cache      ,pad_token_id=self.pad_token_id      ,decoder_start_token_id=self.decoder_start_token_id      ,max_position_embeddings=self.max_position_embeddings      ,)

										return (
										    config,
										    input_ids,
										    attention_mask,
										    lm_labels,
										)

				def       _lowerCamelCase					(   self   :     str      ,SCREAMING_SNAKE_CASE_   :     Dict      ,SCREAMING_SNAKE_CASE_   :     int      ,SCREAMING_SNAKE_CASE_   :     Dict      ,SCREAMING_SNAKE_CASE_   :     int      ,):
										snake_case__       :					Optional[int]										=					True
										snake_case__       :					Dict										=					SpeechaTextaDecoder(config=SCREAMING_SNAKE_CASE_      ).to(SCREAMING_SNAKE_CASE_      ).eval()
										snake_case__       :					List[str]										=					input_ids[:2]

										input_ids[input_ids == 0] += 1
										# first forward pass
										snake_case__       :					Dict										=					model(SCREAMING_SNAKE_CASE_      ,use_cache=SCREAMING_SNAKE_CASE_      )
										snake_case__       :					Any										=					model(SCREAMING_SNAKE_CASE_      )
										snake_case__       :					Tuple										=					model(SCREAMING_SNAKE_CASE_      ,use_cache=SCREAMING_SNAKE_CASE_      )

										self.parent.assertTrue(len(SCREAMING_SNAKE_CASE_      ) == len(SCREAMING_SNAKE_CASE_      )      )
										self.parent.assertTrue(len(SCREAMING_SNAKE_CASE_      ) == len(SCREAMING_SNAKE_CASE_      ) + 1      )

										snake_case__       :					List[Any]										=					outputs['''past_key_values''']

										# create hypothetical next token and extent to next_input_ids
										snake_case__       :					List[str]										=					ids_tensor((2, 1)      ,config.vocab_size - 1      ) + 1

										# append to next input_ids and
										snake_case__       :					int										=					torch.cat([input_ids, next_tokens]      ,dim=-1      )

										print(SCREAMING_SNAKE_CASE_      )

										snake_case__       :					List[Any]										=					model(SCREAMING_SNAKE_CASE_      )['''last_hidden_state''']
										snake_case__       :					Optional[Any]										=					model(SCREAMING_SNAKE_CASE_      ,past_key_values=SCREAMING_SNAKE_CASE_      )['''last_hidden_state''']

										# select random slice
										snake_case__       :					int										=					ids_tensor((1,)      ,output_from_past.shape[-1]      ).item()
										snake_case__       :					str										=					output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
										snake_case__       :					Optional[int]										=					output_from_past[:, 0, random_slice_idx].detach()

										# test that outputs are equal for slice
										assert torch.allclose(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_      ,atol=1E-3      )





				def       _lowerCamelCase					(   self   :     Tuple      ):
										snake_case__       :					str										=					self.prepare_config_and_inputs()
										(
										    snake_case__	
										)       :					Optional[int]										=					config_and_inputs

										snake_case__       :					Optional[Any]										=					{
										    '''input_ids''': input_ids,
										    '''attention_mask''': attention_mask,
										}
										return config, inputs_dict







@require_torch
class       __lowercase       (			snake_case_						,   snake_case_						,   snake_case_						,   unittest.TestCase	):


				'''simple docstring'''


				_lowerCAmelCase				=(SpeechaTextaDecoder, SpeechaTextaForCausalLM) if is_torch_available() else ()
				_lowerCAmelCase				=(SpeechaTextaForCausalLM,) if is_torch_available() else ()
				_lowerCAmelCase				={"text-generation": SpeechaTextaForCausalLM} if is_torch_available() else {}
				_lowerCAmelCase				=True
				_lowerCAmelCase				=False

				def       _lowerCamelCase					(   self   :     str      ,):
										snake_case__       :					str										=					SpeechaTextaStandaloneDecoderModelTester(self      ,is_training=SCREAMING_SNAKE_CASE_      )
										snake_case__       :					Optional[int]										=					ConfigTester(self      ,config_class=SCREAMING_SNAKE_CASE_      )

				def       _lowerCamelCase					(   self   :     Optional[Any]      ):
										pass

				def       _lowerCamelCase					(   self   :     Optional[Any]      ):
										pass

				def       _lowerCamelCase					(   self   :     List[str]      ):
										pass

				def       _lowerCamelCase					(   self   :     Dict      ):
										self.config_tester.run_common_tests()

				def       _lowerCamelCase					(   self   :     Union[str, Any]      ):
										snake_case__       :					Optional[int]										=					self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_decoder_model_past(*SCREAMING_SNAKE_CASE_      )





				def       _lowerCamelCase					(   self   :     Union[str, Any]      ):
										return








"""simple docstring"""





import os
import random
from pathlib import Path
from typing import List

import numpy as np
import torch
from torch.cuda.amp import GradScaler

from .utils import (
    MODEL_NAME,
    OPTIMIZER_NAME,
    RNG_STATE_NAME,
    SCALER_NAME,
    SCHEDULER_NAME,
    get_pretty_name,
    is_tpu_available,
    is_xpu_available,
    save,
)


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

from .logging import get_logger
from .state import PartialState


_lowercase	      =    get_logger(__name__)

def 		lowerCAmelCase							( lowerCamelCase      ,			lowerCamelCase      ,			lowerCamelCase      ,			lowerCamelCase      ,			lowerCamelCase      ,			lowerCamelCase = None      ,			):
						'''simple docstring'''


						for i, state in enumerate(lowerCamelCase		):
												snake_case__       :					List[Any]										=					f'''{MODEL_NAME}.bin''' if i == 0 else f'''{MODEL_NAME}_{i}.bin'''
												snake_case__       :					Tuple										=					os.path.join(lowerCamelCase      ,			lowerCamelCase		)
												save(lowerCamelCase      ,			lowerCamelCase		)
												logger.info(f'''Model weights saved in {output_model_file}'''		)
						# Optimizer states
						for i, opt in enumerate(lowerCamelCase		):
												snake_case__       :					str										=					opt.state_dict()
												snake_case__       :					Optional[Any]										=					f'''{OPTIMIZER_NAME}.bin''' if i == 0 else f'''{OPTIMIZER_NAME}_{i}.bin'''
												snake_case__       :					str										=					os.path.join(lowerCamelCase      ,			lowerCamelCase		)
												save(lowerCamelCase      ,			lowerCamelCase		)
												logger.info(f'''Optimizer state saved in {output_optimizer_file}'''		)
						# Scheduler states
						for i, scheduler in enumerate(lowerCamelCase		):
												snake_case__       :					int										=					scheduler.state_dict()
												snake_case__       :					List[Any]										=					f'''{SCHEDULER_NAME}.bin''' if i == 0 else f'''{SCHEDULER_NAME}_{i}.bin'''
												snake_case__       :					int										=					os.path.join(lowerCamelCase      ,			lowerCamelCase		)
												save(lowerCamelCase      ,			lowerCamelCase		)
												logger.info(f'''Scheduler state saved in {output_scheduler_file}'''		)
						# GradScaler state
						if scaler is not None:
												snake_case__       :					Tuple										=					scaler.state_dict()
												snake_case__       :					Tuple										=					os.path.join(lowerCamelCase      ,			lowerCamelCase		)
												torch.save(lowerCamelCase      ,			lowerCamelCase		)
												logger.info(f'''Gradient scaler state saved in {output_scaler_file}'''		)
						# Random number generator states
						snake_case__       :					Dict										=					{}
						snake_case__       :					Dict										=					f'''{RNG_STATE_NAME}_{process_index}.pkl'''
						snake_case__       :					Dict										=					random.getstate()
						snake_case__       :					List[str]										=					np.random.get_state()
						snake_case__       :					str										=					torch.get_rng_state()
						if is_xpu_available():
												snake_case__       :					List[Any]										=					torch.xpu.get_rng_state_all()
						else:
												snake_case__       :					Union[str, Any]										=					torch.cuda.get_rng_state_all()
						if is_tpu_available():
												snake_case__       :					int										=					xm.get_rng_state()
						snake_case__       :					int										=					os.path.join(lowerCamelCase      ,			lowerCamelCase		)
						torch.save(lowerCamelCase      ,			lowerCamelCase		)
						logger.info(f'''Random states saved in {output_states_file}'''		)
						return output_dir

def 		lowerCAmelCase							( lowerCamelCase      ,			lowerCamelCase      ,			lowerCamelCase      ,			lowerCamelCase      ,			lowerCamelCase      ,			lowerCamelCase=None      ,			lowerCamelCase=None      ,			**lowerCamelCase      ,			):
						'''simple docstring'''


						if map_location not in [None, "cpu", "on_device"]:
												raise TypeError(
												    '''Unsupported optimizer map location passed, please choose one of `None`, `\'cpu\'`, or `\'on_device\'`'''		)
						if map_location is None:
												snake_case__       :					str										=					'''cpu'''
						elif map_location == "on_device":
												snake_case__       :					List[Any]										=					PartialState().device
						# Model states
						for i, model in enumerate(lowerCamelCase		):
												snake_case__       :					Optional[int]										=					f'''{MODEL_NAME}.bin''' if i == 0 else f'''{MODEL_NAME}_{i}.bin'''
												snake_case__       :					int										=					os.path.join(lowerCamelCase      ,			lowerCamelCase		)
												models[i].load_state_dict(torch.load(lowerCamelCase      ,			map_location=lowerCamelCase		)      ,			**lowerCamelCase		)
						logger.info('''All model weights loaded successfully'''		)

						# Optimizer states
						for i, opt in enumerate(lowerCamelCase		):
												snake_case__       :					Any										=					f'''{OPTIMIZER_NAME}.bin''' if i == 0 else f'''{OPTIMIZER_NAME}_{i}.bin'''
												snake_case__       :					Optional[Any]										=					os.path.join(lowerCamelCase      ,			lowerCamelCase		)
												snake_case__       :					Any										=					torch.load(lowerCamelCase      ,			map_location=lowerCamelCase		)
												optimizers[i].load_state_dict(lowerCamelCase		)
						logger.info('''All optimizer states loaded successfully'''		)

						# Scheduler states
						for i, scheduler in enumerate(lowerCamelCase		):
												snake_case__       :					Optional[int]										=					f'''{SCHEDULER_NAME}.bin''' if i == 0 else f'''{SCHEDULER_NAME}_{i}.bin'''
												snake_case__       :					Optional[Any]										=					os.path.join(lowerCamelCase      ,			lowerCamelCase		)
												scheduler.load_state_dict(torch.load(lowerCamelCase		)		)
						logger.info('''All scheduler states loaded successfully'''		)

						# GradScaler state
						if scaler is not None:
												snake_case__       :					Optional[Any]										=					os.path.join(lowerCamelCase      ,			lowerCamelCase		)
												scaler.load_state_dict(torch.load(lowerCamelCase		)		)
												logger.info('''GradScaler state loaded successfully'''		)

						# Random states
						try:
												snake_case__       :					Optional[int]										=					torch.load(os.path.join(lowerCamelCase      ,			f'''{RNG_STATE_NAME}_{process_index}.pkl'''		)		)
												random.setstate(states['''random_state''']		)
												np.random.set_state(states['''numpy_random_seed''']		)
												torch.set_rng_state(states['''torch_manual_seed''']		)
												if is_xpu_available():
																		torch.xpu.set_rng_state_all(states['''torch_xpu_manual_seed''']		)
												else:
																		torch.cuda.set_rng_state_all(states['''torch_cuda_manual_seed''']		)
												if is_tpu_available():
																		xm.set_rng_state(states['''xm_seed''']		)
												logger.info('''All random states loaded successfully'''		)
						except Exception:
												logger.info('''Could not load random states'''		)

def 		lowerCAmelCase							( lowerCamelCase      ,			lowerCamelCase      ,			lowerCamelCase = 0		):
						'''simple docstring'''


						snake_case__       :					Union[str, Any]										=					Path(lowerCamelCase		) / f'''custom_checkpoint_{index}.pkl'''
						logger.info(f'''Saving the state of {get_pretty_name(lowerCamelCase		)} to {save_location}'''		)
						torch.save(obj.state_dict()      ,			lowerCamelCase		)

def 		lowerCAmelCase							( lowerCamelCase      ,			lowerCamelCase      ,			lowerCamelCase = 0		):
						'''simple docstring'''


						snake_case__       :					Any										=					f'''{path}/custom_checkpoint_{index}.pkl'''
						logger.info(f'''Loading the state of {get_pretty_name(lowerCamelCase		)} from {load_location}'''		)
						obj.load_state_dict(torch.load(lowerCamelCase      ,			map_location='''cpu'''		)		)


