




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




    """simple docstring"""

    def __init__(       self     :     Any   ,	SCREAMING_SNAKE_CASE__     :     int   ,	SCREAMING_SNAKE_CASE__     :     List[Any]=99   ,	SCREAMING_SNAKE_CASE__     :     int=13   ,	SCREAMING_SNAKE_CASE__     :     int=16   ,	SCREAMING_SNAKE_CASE__     :     Dict=7   ,	SCREAMING_SNAKE_CASE__     :     List[Any]=True   ,	SCREAMING_SNAKE_CASE__     :     List[Any]=True   ,	SCREAMING_SNAKE_CASE__     :     Optional[int]=True   ,	SCREAMING_SNAKE_CASE__     :     Optional[int]=False   ,	SCREAMING_SNAKE_CASE__     :     Union[str, Any]=True   ,	SCREAMING_SNAKE_CASE__     :     Tuple=2   ,	SCREAMING_SNAKE_CASE__     :     List[str]=32   ,	SCREAMING_SNAKE_CASE__     :     Optional[int]=4   ,	SCREAMING_SNAKE_CASE__     :     Union[str, Any]=4   ,	SCREAMING_SNAKE_CASE__     :     Optional[Any]=30   ,	SCREAMING_SNAKE_CASE__     :     str=0   ,	SCREAMING_SNAKE_CASE__     :     Tuple=1   ,	SCREAMING_SNAKE_CASE__     :     Optional[Any]=2   ,	SCREAMING_SNAKE_CASE__     :     List[str]=None   ,	)      ->       Dict:
         lowerCAmelCase__		= parent
         lowerCAmelCase__		= batch_size
         lowerCAmelCase__		= decoder_seq_length
         # For common tests
         lowerCAmelCase__		= self.decoder_seq_length
         lowerCAmelCase__		= is_training
         lowerCAmelCase__		= use_attention_mask
         lowerCAmelCase__		= use_labels

         lowerCAmelCase__		= vocab_size
         lowerCAmelCase__		= d_model
         lowerCAmelCase__		= d_model
         lowerCAmelCase__		= decoder_layers
         lowerCAmelCase__		= decoder_layers
         lowerCAmelCase__		= decoder_ffn_dim
         lowerCAmelCase__		= decoder_attention_heads
         lowerCAmelCase__		= decoder_attention_heads
         lowerCAmelCase__		= eos_token_id
         lowerCAmelCase__		= bos_token_id
         lowerCAmelCase__		= pad_token_id
         lowerCAmelCase__		= decoder_start_token_id
         lowerCAmelCase__		= use_cache
         lowerCAmelCase__		= max_position_embeddings

         lowerCAmelCase__		= None
         lowerCAmelCase__		= decoder_seq_length
         lowerCAmelCase__		= 2
         lowerCAmelCase__		= 1


    def    a      (       self     :     int   )      ->       Any:
         lowerCAmelCase__		= ids_tensor([self.batch_size, self.decoder_seq_length]   ,	self.vocab_size   )

         lowerCAmelCase__		= None
         if self.use_attention_mask:
              lowerCAmelCase__		= ids_tensor([self.batch_size, self.decoder_seq_length]   ,	vocab_size=2   )

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

         lowerCAmelCase__		= 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    a      (       self     :     str   ,	SCREAMING_SNAKE_CASE__     :     Union[str, Any]   ,	SCREAMING_SNAKE_CASE__     :     Tuple   ,	SCREAMING_SNAKE_CASE__     :     Union[str, Any]   ,	SCREAMING_SNAKE_CASE__     :     Tuple   ,	)      ->       str:
         lowerCAmelCase__		= True
         lowerCAmelCase__		= SpeechaTextaDecoder(config=SCREAMING_SNAKE_CASE__   ).to(SCREAMING_SNAKE_CASE__   ).eval()
         lowerCAmelCase__		= input_ids[:2]

         input_ids[input_ids == 0] += 1
         # first forward pass
         lowerCAmelCase__		= model(SCREAMING_SNAKE_CASE__   ,	use_cache=SCREAMING_SNAKE_CASE__   )
         lowerCAmelCase__		= model(SCREAMING_SNAKE_CASE__   )
         lowerCAmelCase__		= 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   )

         lowerCAmelCase__		= outputs["past_key_values"]

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

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

         print(SCREAMING_SNAKE_CASE__   )

         lowerCAmelCase__		= model(SCREAMING_SNAKE_CASE__   )["last_hidden_state"]
         lowerCAmelCase__		= model(SCREAMING_SNAKE_CASE__   ,	past_key_values=SCREAMING_SNAKE_CASE__   )["last_hidden_state"]

         # select random slice
         lowerCAmelCase__		= ids_tensor((1,)   ,	output_from_past.shape[-1]   ).item()
         lowerCAmelCase__		= output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
         lowerCAmelCase__		= 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    a      (       self     :     Dict   )      ->       Any:
         lowerCAmelCase__		= self.prepare_config_and_inputs()
         (
             (
             lowerCAmelCase__
         )	,		(
             lowerCAmelCase__
         )	,		(
             lowerCAmelCase__
         )	,		(
             lowerCAmelCase__
         )	,		
         )		= config_and_inputs

         lowerCAmelCase__		= {
             "input_ids": input_ids,
             "attention_mask": attention_mask,
         }
         return config, inputs_dict






@require_torch
class  __lowerCamelCase       (    UpperCamelCase__	,				UpperCamelCase__	,				UpperCamelCase__	,				unittest.TestCase    ):




    """simple docstring"""

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


    def    a      (       self     :     Optional[Any]   ,	)      ->       Optional[Any]:
         lowerCAmelCase__		= SpeechaTextaStandaloneDecoderModelTester(self   ,	is_training=SCREAMING_SNAKE_CASE__   )
         lowerCAmelCase__		= ConfigTester(self   ,	config_class=SCREAMING_SNAKE_CASE__   )


    def    a      (       self     :     List[Any]   )      ->       Union[str, Any]:
         pass


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


    def    a      (       self     :     Union[str, Any]   )      ->       Dict:
         pass


    def    a      (       self     :     Optional[Any]   )      ->       Any:
         self.config_tester.run_common_tests()


    def    a      (       self     :     Union[str, Any]   )      ->       Union[str, Any]:
         lowerCAmelCase__		= self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_decoder_model_past(*SCREAMING_SNAKE_CASE__   )


    def    a      (       self     :     List[str]   )      ->       int:
         return






import os
import pathlib
import tempfile
import uuid

import numpy as np

from ..utils import is_soundfile_availble, is_torch_available, is_vision_available, logging


UpperCamelCase								=     logging.get_logger(__name__)

if is_vision_available():
     import PIL.Image
     from PIL import Image
     from PIL.Image import Image as ImageType
else:
     UpperCamelCase								=     object

if is_torch_available():
     import torch

if is_soundfile_availble():
     import soundfile as sf





class  __lowerCamelCase       :




    """simple docstring"""

    def __init__(       self     :     Tuple   ,	SCREAMING_SNAKE_CASE__     :     List[str]   )      ->       Union[str, Any]:
         lowerCAmelCase__		= value


    def __str__(       self     :     int   )      ->       List[Any]:
         return self.to_string()


    def    a      (       self     :     str   )      ->       Union[str, Any]:
         logger.error(
             "This is a raw AgentType of unknown type. Display in notebooks and string conversion will be unreliable"   )
         return self._value


    def    a      (       self     :     str   )      ->       str:
         logger.error(
             "This is a raw AgentType of unknown type. Display in notebooks and string conversion will be unreliable"   )
         return str(self._value   )






class  __lowerCamelCase       (    UpperCamelCase__	,				UpperCamelCase__    ):




    """simple docstring"""

    def    a      (       self     :     Optional[Any]   )      ->       int:
         return self._value


    def    a      (       self     :     Optional[Any]   )      ->       Any:
         return self._value






class  __lowerCamelCase       (    UpperCamelCase__	,				UpperCamelCase__    ):




    """simple docstring"""

    def __init__(       self     :     Union[str, Any]   ,	SCREAMING_SNAKE_CASE__     :     Optional[Any]   )      ->       Optional[int]:
         super().__init__(SCREAMING_SNAKE_CASE__   )

         if not is_vision_available():
              raise ImportError("PIL must be installed in order to handle images."   )

         lowerCAmelCase__		= None
         lowerCAmelCase__		= None
         lowerCAmelCase__		= None

         if isinstance(SCREAMING_SNAKE_CASE__   ,	SCREAMING_SNAKE_CASE__   ):
              lowerCAmelCase__		= value
         elif isinstance(SCREAMING_SNAKE_CASE__   ,	(str, pathlib.Path)   ):
              lowerCAmelCase__		= value
         elif isinstance(SCREAMING_SNAKE_CASE__   ,	torch.Tensor   ):
              lowerCAmelCase__		= value
         else:
              raise ValueError(f'Unsupported type for {self.__class__.__name__}: {type(SCREAMING_SNAKE_CASE__   )}'   )


    def    a      (       self     :     Optional[int]   ,	SCREAMING_SNAKE_CASE__     :     Any=None   ,	SCREAMING_SNAKE_CASE__     :     str=None   )      ->       Union[str, Any]:
         from IPython.display import Image, display

         display(Image(self.to_string()   )   )


    def    a      (       self     :     List[Any]   )      ->       int:
         if self._raw is not None:
              return self._raw

         if self._path is not None:
              lowerCAmelCase__		= Image.open(self._path   )
              return self._raw


    def    a      (       self     :     Optional[int]   )      ->       Union[str, Any]:
         if self._path is not None:
              return self._path

         if self._raw is not None:
              lowerCAmelCase__		= tempfile.mkdtemp()
              lowerCAmelCase__		= os.path.join(SCREAMING_SNAKE_CASE__   ,	str(uuid.uuida()   ) + ".png"   )
              self._raw.save(self._path   )

              return self._path

         if self._tensor is not None:
              lowerCAmelCase__		= self._tensor.cpu().detach().numpy()

              # There is likely simpler than load into image into save
              lowerCAmelCase__		= Image.fromarray((array * 255).astype(np.uinta   )   )

              lowerCAmelCase__		= tempfile.mkdtemp()
              lowerCAmelCase__		= os.path.join(SCREAMING_SNAKE_CASE__   ,	str(uuid.uuida()   ) + ".png"   )

              img.save(self._path   )

              return self._path






class  __lowerCamelCase       (    UpperCamelCase__    ):




    """simple docstring"""

    def __init__(       self     :     List[str]   ,	SCREAMING_SNAKE_CASE__     :     Union[str, Any]   ,	SCREAMING_SNAKE_CASE__     :     int=16_000   )      ->       Union[str, Any]:
         super().__init__(SCREAMING_SNAKE_CASE__   )

         if not is_soundfile_availble():
              raise ImportError("soundfile must be installed in order to handle audio."   )

         lowerCAmelCase__		= None
         lowerCAmelCase__		= None

         lowerCAmelCase__		= samplerate

         if isinstance(SCREAMING_SNAKE_CASE__   ,	(str, pathlib.Path)   ):
              lowerCAmelCase__		= value
         elif isinstance(SCREAMING_SNAKE_CASE__   ,	torch.Tensor   ):
              lowerCAmelCase__		= value
         else:
              raise ValueError(f'Unsupported audio type: {type(SCREAMING_SNAKE_CASE__   )}'   )


    def    a      (       self     :     str   ,	SCREAMING_SNAKE_CASE__     :     List[str]=None   ,	SCREAMING_SNAKE_CASE__     :     List[str]=None   )      ->       Any:
         from IPython.display import Audio, display

         display(Audio(self.to_string()   ,	rate=self.samplerate   )   )


    def    a      (       self     :     Optional[Any]   )      ->       List[Any]:
         if self._tensor is not None:
              return self._tensor

         if self._path is not None:
              lowerCAmelCase__	,		lowerCAmelCase__		= sf.read(self._path   )
              lowerCAmelCase__		= torch.tensor(SCREAMING_SNAKE_CASE__   )
              return self._tensor


    def    a      (       self     :     Dict   )      ->       List[str]:
         if self._path is not None:
              return self._path

         if self._tensor is not None:
              lowerCAmelCase__		= tempfile.mkdtemp()
              lowerCAmelCase__		= os.path.join(SCREAMING_SNAKE_CASE__   ,	str(uuid.uuida()   ) + ".wav"   )
              sf.write(self._path   ,	self._tensor   ,	samplerate=self.samplerate   )
              return self._path






UpperCamelCase								=     {'text': AgentText, 'image': AgentImage, 'audio': AgentAudio}
UpperCamelCase								=     {str: AgentText}

if is_vision_available():
     UpperCamelCase								=     AgentImage



def 							_A   (						*lowerCAmelCase_     :					List[str]   ,				**lowerCAmelCase_     :					Optional[int]			):
     """simple docstring"""

     lowerCAmelCase__		= [(arg.to_raw() if isinstance(lowerCAmelCase_   ,				lowerCAmelCase_			) else arg) for arg in args]
     lowerCAmelCase__		= {k: (v.to_raw() if isinstance(lowerCAmelCase_   ,				lowerCAmelCase_			) else v) for k, v in kwargs.items()}
     return args, kwargs






def 							_A   (						lowerCAmelCase_     :					Dict   ,				lowerCAmelCase_     :					Optional[int]=None			):
     """simple docstring"""

     if isinstance(lowerCAmelCase_   ,				lowerCAmelCase_			):
          lowerCAmelCase__		= {}
          for i, (k, v) in enumerate(outputs.items()			):
               if output_types is not None:
                    # If the class has defined outputs, we can map directly according to the class definition
                    if output_types[i] in AGENT_TYPE_MAPPING:
                         lowerCAmelCase__		= AGENT_TYPE_MAPPING[output_types[i]](lowerCAmelCase_			)
                    else:
                         lowerCAmelCase__		= AgentType(lowerCAmelCase_			)

               else:
                    # If the class does not have defined output, then we map according to the type
                    for _k, _v in INSTANCE_TYPE_MAPPING.items():
                         if isinstance(lowerCAmelCase_   ,				_k			):
                              lowerCAmelCase__		= _v(lowerCAmelCase_			)
                    if k not in decoded_outputs:
                         lowerCAmelCase__		= AgentType[v]

     elif isinstance(lowerCAmelCase_   ,				(list, tuple)			):
          lowerCAmelCase__		= type(lowerCAmelCase_			)()
          for i, v in enumerate(lowerCAmelCase_			):
               if output_types is not None:
                    # If the class has defined outputs, we can map directly according to the class definition
                    if output_types[i] in AGENT_TYPE_MAPPING:
                         decoded_outputs.append(AGENT_TYPE_MAPPING[output_types[i]](lowerCAmelCase_			)			)
                    else:
                         decoded_outputs.append(AgentType(lowerCAmelCase_			)			)
               else:
                    # If the class does not have defined output, then we map according to the type
                    lowerCAmelCase__		= False
                    for _k, _v in INSTANCE_TYPE_MAPPING.items():
                         if isinstance(lowerCAmelCase_   ,				_k			):
                              decoded_outputs.append(_v(lowerCAmelCase_			)			)
                              lowerCAmelCase__		= True

                    if not found:
                         decoded_outputs.append(AgentType(lowerCAmelCase_			)			)

     else:
          if output_types[0] in AGENT_TYPE_MAPPING:
               # If the class has defined outputs, we can map directly according to the class definition
               lowerCAmelCase__		= AGENT_TYPE_MAPPING[output_types[0]](lowerCAmelCase_			)

          else:
               # If the class does not have defined output, then we map according to the type
               for _k, _v in INSTANCE_TYPE_MAPPING.items():
                    if isinstance(lowerCAmelCase_   ,				_k			):
                         return _v(lowerCAmelCase_			)
               return AgentType(lowerCAmelCase_			)

     return decoded_outputs
