


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    A_   :







 """simple docstring"""





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

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

      UpperCAmelCase						:				Optional[Any]		  =			None
      UpperCAmelCase						:				List[str]		  =			decoder_seq_length
      UpperCAmelCase						:				Optional[Any]		  =			2
      UpperCAmelCase						:				Optional[Any]		  =			1
 def   _UpperCamelCase    (			self					: Union[str, Any]				)    ->			int:
      UpperCAmelCase						:				List[Any]		  =			ids_tensor([self.batch_size, self.decoder_seq_length]       ,    self.vocab_size				)

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

      UpperCAmelCase						:				Optional[int]		  =			None
      if self.use_labels:
           UpperCAmelCase						:				Optional[int]		  =			ids_tensor([self.batch_size, self.decoder_seq_length]       ,    self.vocab_size				)

      UpperCAmelCase						:				List[Any]		  =			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   _UpperCamelCase    (			self					: Union[str, Any]       ,    _a					: Optional[int]       ,    _a					: List[Any]       ,    _a					: Union[str, Any]       ,    _a					: Tuple       ,    )    ->			Dict:
      UpperCAmelCase						:				Dict		  =			True
      UpperCAmelCase						:				Optional[int]		  =			SpeechaTextaDecoder(config=_a				).to(_a				).eval()
      UpperCAmelCase						:				int		  =			input_ids[:2]

      input_ids[input_ids == 0] += 1
      # first forward pass
      UpperCAmelCase						:				Any		  =			model(_a       ,    use_cache=_a				)
      UpperCAmelCase						:				List[str]		  =			model(_a				)
      UpperCAmelCase						:				Dict		  =			model(_a       ,    use_cache=_a				)

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

      UpperCAmelCase						:				Tuple		  =			outputs["past_key_values"]

      # create hypothetical next token and extent to next_input_ids
      UpperCAmelCase						:				Tuple		  =			ids_tensor((2, 1)       ,    config.vocab_size - 1				) + 1

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

      print(_a				)

      UpperCAmelCase						:				int		  =			model(_a				)["last_hidden_state"]
      UpperCAmelCase						:				List[Any]		  =			model(_a       ,    past_key_values=_a				)["last_hidden_state"]

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

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

 def   _UpperCamelCase    (			self					: str				)    ->			List[Any]:
      UpperCAmelCase						:				str		  =			self.prepare_config_and_inputs()
      (
          (
          UpperCAmelCase
      )  , (
          UpperCAmelCase
      )  , (
          UpperCAmelCase
      )  , (
          UpperCAmelCase
      )  , 
      )						:				str		  =			config_and_inputs

      UpperCAmelCase						:				Any		  =			{
          "input_ids": input_ids,
          "attention_mask": attention_mask,
      }
      return config, inputs_dict




@require_torch
class    A_   ( a_      ,						a_      ,						a_      ,						unittest.TestCase     ):







 """simple docstring"""





 _UpperCamelCase				: str      =					(SpeechaTextaDecoder, SpeechaTextaForCausalLM) if is_torch_available() else ()
 _UpperCamelCase				: List[str]      =					(SpeechaTextaForCausalLM,) if is_torch_available() else ()
 _UpperCamelCase				: List[str]      =					{'''text-generation''': SpeechaTextaForCausalLM} if is_torch_available() else {}
 _UpperCamelCase				: List[str]      =					True
 _UpperCamelCase				: Optional[Any]      =					False
 def   _UpperCamelCase    (			self					: int       ,    )    ->			Optional[int]:
      UpperCAmelCase						:				List[Any]		  =			SpeechaTextaStandaloneDecoderModelTester(self       ,    is_training=_a				)
      UpperCAmelCase						:				List[Any]		  =			ConfigTester(self       ,    config_class=_a				)
 def   _UpperCamelCase    (			self					: List[Any]				)    ->			Union[str, Any]:
      pass
 def   _UpperCamelCase    (			self					: Optional[int]				)    ->			int:
      pass
 def   _UpperCamelCase    (			self					: Tuple				)    ->			List[str]:
      pass
 def   _UpperCamelCase    (			self					: List[Any]				)    ->			List[Any]:
      self.config_tester.run_common_tests()
 def   _UpperCamelCase    (			self					: Any				)    ->			Any:
      UpperCAmelCase						:				int		  =			self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_decoder_model_past(*_a				)

 def   _UpperCamelCase    (			self					: List[Any]				)    ->			Union[str, Any]:
      return







import torch

from diffusers import EulerAncestralDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest




class    A_   ( a_     ):







 """simple docstring"""





 _UpperCamelCase				: Any      =					(EulerAncestralDiscreteScheduler,)
 _UpperCamelCase				: Union[str, Any]      =					10
 def   _UpperCamelCase    (			self					: int       ,    **_a					: str				)    ->			Dict:
      UpperCAmelCase						:				List[Any]		  =			{
          "num_train_timesteps": 1_1_0_0,
          "beta_start": 0.0_0_0_1,
          "beta_end": 0.0_2,
          "beta_schedule": "linear",
      }

      config.update(**_a				)
      return config
 def   _UpperCamelCase    (			self					: str				)    ->			Union[str, Any]:
      for timesteps in [1_0, 5_0, 1_0_0, 1_0_0_0]:
           self.check_over_configs(num_train_timesteps=_a				)
 def   _UpperCamelCase    (			self					: Tuple				)    ->			int:
      for beta_start, beta_end in zip([0.0_0_0_0_1, 0.0_0_0_1, 0.0_0_1]       ,    [0.0_0_0_2, 0.0_0_2, 0.0_2]				):
           self.check_over_configs(beta_start=_a       ,    beta_end=_a				)
 def   _UpperCamelCase    (			self					: Optional[Any]				)    ->			List[str]:
      for schedule in ["linear", "scaled_linear"]:
           self.check_over_configs(beta_schedule=_a				)
 def   _UpperCamelCase    (			self					: Dict				)    ->			str:
      for prediction_type in ["epsilon", "v_prediction"]:
           self.check_over_configs(prediction_type=_a				)
 def   _UpperCamelCase    (			self					: Any				)    ->			Optional[Any]:
      UpperCAmelCase						:				int		  =			self.scheduler_classes[0]
      UpperCAmelCase						:				Optional[int]		  =			self.get_scheduler_config()
      UpperCAmelCase						:				List[Any]		  =			scheduler_class(**_a				)

      scheduler.set_timesteps(self.num_inference_steps				)

      UpperCAmelCase						:				Dict		  =			torch.manual_seed(0				)

      UpperCAmelCase						:				Union[str, Any]		  =			self.dummy_model()
      UpperCAmelCase						:				Dict		  =			self.dummy_sample_deter * scheduler.init_noise_sigma.cpu()
      UpperCAmelCase						:				Optional[Any]		  =			sample.to(_a				)

      for i, t in enumerate(scheduler.timesteps				):
           UpperCAmelCase						:				List[Any]		  =			scheduler.scale_model_input(_a       ,    _a				)

           UpperCAmelCase						:				str		  =			model(_a       ,    _a				)

           UpperCAmelCase						:				int		  =			scheduler.step(_a       ,    _a       ,    _a       ,    generator=_a				)
           UpperCAmelCase						:				Optional[int]		  =			output.prev_sample

      UpperCAmelCase						:				Tuple		  =			torch.sum(torch.abs(_a				)				)
      UpperCAmelCase						:				str		  =			torch.mean(torch.abs(_a				)				)

      assert abs(result_sum.item() - 1_5_2.3_1_9_2				) < 1E-2
      assert abs(result_mean.item() - 0.1_9_8_3				) < 1E-3
 def   _UpperCamelCase    (			self					: Optional[int]				)    ->			Tuple:
      UpperCAmelCase						:				Optional[int]		  =			self.scheduler_classes[0]
      UpperCAmelCase						:				Tuple		  =			self.get_scheduler_config(prediction_type="v_prediction"				)
      UpperCAmelCase						:				str		  =			scheduler_class(**_a				)

      scheduler.set_timesteps(self.num_inference_steps				)

      UpperCAmelCase						:				Dict		  =			torch.manual_seed(0				)

      UpperCAmelCase						:				List[Any]		  =			self.dummy_model()
      UpperCAmelCase						:				Optional[int]		  =			self.dummy_sample_deter * scheduler.init_noise_sigma
      UpperCAmelCase						:				Any		  =			sample.to(_a				)

      for i, t in enumerate(scheduler.timesteps				):
           UpperCAmelCase						:				Union[str, Any]		  =			scheduler.scale_model_input(_a       ,    _a				)

           UpperCAmelCase						:				Union[str, Any]		  =			model(_a       ,    _a				)

           UpperCAmelCase						:				Union[str, Any]		  =			scheduler.step(_a       ,    _a       ,    _a       ,    generator=_a				)
           UpperCAmelCase						:				Optional[int]		  =			output.prev_sample

      UpperCAmelCase						:				Dict		  =			torch.sum(torch.abs(_a				)				)
      UpperCAmelCase						:				Any		  =			torch.mean(torch.abs(_a				)				)

      assert abs(result_sum.item() - 1_0_8.4_4_3_9				) < 1E-2
      assert abs(result_mean.item() - 0.1_4_1_2				) < 1E-3

 def   _UpperCamelCase    (			self					: List[str]				)    ->			Optional[Any]:
      UpperCAmelCase						:				str		  =			self.scheduler_classes[0]
      UpperCAmelCase						:				str		  =			self.get_scheduler_config()
      UpperCAmelCase						:				Union[str, Any]		  =			scheduler_class(**_a				)

      scheduler.set_timesteps(self.num_inference_steps       ,    device=_a				)
      UpperCAmelCase						:				str		  =			torch.manual_seed(0				)

      UpperCAmelCase						:				Optional[int]		  =			self.dummy_model()
      UpperCAmelCase						:				Any		  =			self.dummy_sample_deter * scheduler.init_noise_sigma.cpu()
      UpperCAmelCase						:				Optional[Any]		  =			sample.to(_a				)

      for t in scheduler.timesteps:
           UpperCAmelCase						:				Tuple		  =			scheduler.scale_model_input(_a       ,    _a				)

           UpperCAmelCase						:				int		  =			model(_a       ,    _a				)

           UpperCAmelCase						:				Tuple		  =			scheduler.step(_a       ,    _a       ,    _a       ,    generator=_a				)
           UpperCAmelCase						:				str		  =			output.prev_sample

      UpperCAmelCase						:				Tuple		  =			torch.sum(torch.abs(_a				)				)
      UpperCAmelCase						:				Optional[Any]		  =			torch.mean(torch.abs(_a				)				)

      assert abs(result_sum.item() - 1_5_2.3_1_9_2				) < 1E-2
      assert abs(result_mean.item() - 0.1_9_8_3				) < 1E-3



