
from __future__ import annotations

import random
import unittest

from transformers import TransfoXLConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
					import tensorflow as tf

					from transformers import (
					    TF_TRANSFO_XL_PRETRAINED_MODEL_ARCHIVE_LIST,
					    TFTransfoXLForSequenceClassification,
					    TFTransfoXLLMHeadModel,
					    TFTransfoXLModel,
					)







class 			__A:




				def __init__(     self		,     _snake_case		,     )    ->      List[str]:


								'''simple docstring'''




								__a										=    parent
								__a										=    13
								__a										=    7
								__a										=    30
								__a										=    self.seq_length + self.mem_len
								__a										=    15
								__a										=    True
								__a										=    True
								__a										=    99
								__a										=    [10, 50, 80]
								__a										=    32
								__a										=    32
								__a										=    4
								__a										=    8
								__a										=    128
								__a										=    2
								__a										=    2
								__a										=    None
								__a										=    1
								__a										=    0
								__a										=    3
								__a										=    self.vocab_size - 1
								__a										=    0.01




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      List[Any]:


								'''simple docstring'''




								__a										=    ids_tensor([self.batch_size, self.seq_length]		,     self.vocab_size				)
								__a										=    ids_tensor([self.batch_size, self.seq_length]		,     self.vocab_size				)

								__a										=    None
								if self.use_labels:
												__a										=    ids_tensor([self.batch_size, self.seq_length]		,     self.vocab_size				)

								__a										=    TransfoXLConfig(
								    vocab_size=self.vocab_size		,     mem_len=self.mem_len		,     clamp_len=self.clamp_len		,     cutoffs=self.cutoffs		,     d_model=self.hidden_size		,     d_embed=self.d_embed		,     n_head=self.num_attention_heads		,     d_head=self.d_head		,     d_inner=self.d_inner		,     div_val=self.div_val		,     n_layer=self.num_hidden_layers		,     eos_token_id=self.eos_token_id		,     pad_token_id=self.vocab_size - 1		,     init_range=self.init_range		,     num_labels=self.num_labels		,     )

								return (config, input_ids_a, input_ids_a, lm_labels)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Tuple:


								'''simple docstring'''




								random.seed(self.seed				)
								tf.random.set_seed(self.seed				)




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case		,     _snake_case		,     _snake_case				)    ->      List[Any]:


								'''simple docstring'''




								__a										=    TFTransfoXLModel(SCREAMING_SNAKE_CASE__				)

								__a										=    model(SCREAMING_SNAKE_CASE__				).to_tuple()

								__a										=    {'input_ids': input_ids_a, 'mems': mems_a}

								__a										=    model(SCREAMING_SNAKE_CASE__				).to_tuple()

								self.parent.assertEqual(hidden_states_a.shape		,     (self.batch_size, self.seq_length, self.hidden_size)				)
								self.parent.assertEqual(hidden_states_a.shape		,     (self.batch_size, self.seq_length, self.hidden_size)				)
								self.parent.assertListEqual(
								    [mem.shape for mem in mems_a]		,     [(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers		,     )
								self.parent.assertListEqual(
								    [mem.shape for mem in mems_a]		,     [(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers		,     )




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case		,     _snake_case		,     _snake_case				)    ->      str:


								'''simple docstring'''




								__a										=    TFTransfoXLLMHeadModel(SCREAMING_SNAKE_CASE__				)

								__a										=    model(SCREAMING_SNAKE_CASE__				).to_tuple()

								__a										=    {'input_ids': input_ids_a, 'labels': lm_labels}
								__a										=    model(SCREAMING_SNAKE_CASE__				).to_tuple()

								__a										=    model([input_ids_a, mems_a]				).to_tuple()

								__a										=    {'input_ids': input_ids_a, 'mems': mems_a, 'labels': lm_labels}

								__a										=    model(SCREAMING_SNAKE_CASE__				).to_tuple()

								self.parent.assertEqual(lm_logits_a.shape		,     (self.batch_size, self.seq_length, self.vocab_size)				)
								self.parent.assertListEqual(
								    [mem.shape for mem in mems_a]		,     [(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers		,     )

								self.parent.assertEqual(lm_logits_a.shape		,     (self.batch_size, self.seq_length, self.vocab_size)				)
								self.parent.assertListEqual(
								    [mem.shape for mem in mems_a]		,     [(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers		,     )




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case		,     _snake_case		,     _snake_case				)    ->      Dict:


								'''simple docstring'''




								__a										=    TFTransfoXLForSequenceClassification(SCREAMING_SNAKE_CASE__				)
								__a										=    model(SCREAMING_SNAKE_CASE__				)
								self.parent.assertEqual(result.logits.shape		,     (self.batch_size, self.num_labels)				)




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


								'''simple docstring'''




								__a										=    self.prepare_config_and_inputs()
								(__a)										=    config_and_inputs
								__a										=    {'input_ids': input_ids_a}
								return config, inputs_dict







@require_tf
class 			__A( lowercase__  ,   lowercase__  ,   unittest.TestCase			):
				snake_case_						   =    (
				    (TFTransfoXLModel, TFTransfoXLLMHeadModel, TFTransfoXLForSequenceClassification) if is_tf_available() else ()
				)
				snake_case_						   =    () if is_tf_available() else ()
				snake_case_						   =    (
				    {
				        '''feature-extraction''': TFTransfoXLModel,
				        '''text-classification''': TFTransfoXLForSequenceClassification,
				        '''text-generation''': TFTransfoXLLMHeadModel,
				        '''zero-shot''': TFTransfoXLForSequenceClassification,
				    }
				    if is_tf_available()
				    else {}
				)
				# TODO: add this test when TFTransfoXLLMHead has a linear output layer implemented
				snake_case_						   =    False
				snake_case_						   =    False
				snake_case_						   =    False
				snake_case_						   =    False




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case		,     _snake_case		,     _snake_case		,     _snake_case				)    ->      str:


								'''simple docstring'''




								if pipeline_test_casse_name == "TextGenerationPipelineTests":
												# Get `ValueError: AttributeError: 'NoneType' object has no attribute 'new_ones'` or `AssertionError`.
												# `TransfoXLConfig` was never used in pipeline tests: cannot create a simple
												# tokenizer.
												return True

								return False




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[Any]:


								'''simple docstring'''




								__a										=    TFTransfoXLModelTester(self				)
								__a										=    ConfigTester(self		,     config_class=SCREAMING_SNAKE_CASE__		,     d_embed=37				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[int]:


								'''simple docstring'''




								self.config_tester.run_common_tests()




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[int]:


								'''simple docstring'''




								self.model_tester.set_seed()
								__a										=    self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_transfo_xl_model(*SCREAMING_SNAKE_CASE__				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      int:


								'''simple docstring'''




								self.model_tester.set_seed()
								__a										=    self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_transfo_xl_lm_head(*SCREAMING_SNAKE_CASE__				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      str:


								'''simple docstring'''




								__a										=    self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_transfo_xl_for_sequence_classification(*SCREAMING_SNAKE_CASE__				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      str:


								'''simple docstring'''




								__a										=    self.model_tester.prepare_config_and_inputs_for_common()
								__a										=    [TFTransfoXLForSequenceClassification]

								for model_class in self.all_model_classes:
												__a										=    model_class(SCREAMING_SNAKE_CASE__				)
												assert isinstance(model.get_input_embeddings()		,     tf.keras.layers.Layer				)
												if model_class in list_other_models_with_output_ebd:
																__a										=    model.get_output_embeddings()
																assert isinstance(SCREAMING_SNAKE_CASE__		,     tf.keras.layers.Layer				)
																__a										=    model.get_bias()
																assert name is None
												else:
																__a										=    model.get_output_embeddings()
																assert x is None
																__a										=    model.get_bias()
																assert name is None




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Tuple:


								'''simple docstring'''




								pass




				@slow
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[Any]:


								'''simple docstring'''




								for model_name in TF_TRANSFO_XL_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
												__a										=    TFTransfoXLModel.from_pretrained(SCREAMING_SNAKE_CASE__				)
												self.assertIsNotNone(SCREAMING_SNAKE_CASE__				)




				@unittest.skip(reason='''This model doesn\'t play well with fit() due to not returning a single loss.'''				)
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[Any]:


								'''simple docstring'''




								pass







@require_tf
class 			__A( unittest.TestCase			):




				@unittest.skip('''Skip test until #12651 is resolved.'''				)
				@slow
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Any:


								'''simple docstring'''




								__a										=    TFTransfoXLLMHeadModel.from_pretrained('''transfo-xl-wt103'''				)
								# fmt: off
								__a										=    tf.convert_to_tensor([[33,1_297,2,1,1_009,4,1_109,11_739,4_762,358,5,25,245,22,1_706,17,20_098,5,3_215,21,37,1_110,3,13,1_041,4,24,603,490,2,71_477,20_098,104_447,2,20_961,1,2_604,4,1,329,3,6_224,831,16_002,2,8,603,78_967,29_546,23,803,20,25,416,5,8,232,4,277,6,1_855,4_601,3,29_546,54,8,3_609,5,57_211,49,4,1,277,18,8,1_755,15_691,3,341,25,416,693,42_573,71,17,401,94,31,17_919,2,29_546,7_873,18,1,435,23,11_011,755,5,5_167,3,7_983,98,84,2,29_546,3_267,8,3_609,4,1,4_865,1_075,2,6_087,71,6,346,8,5_854,3,29_546,824,1_400,1_868,2,19,160,2,311,8,5_496,2,20_920,17,25,15_097,3,24,24,0]]		,     dtype=tf.intaa				)  # noqa: E231
								# fmt: on
								#  In 1991 , the remains of Russian Tsar Nicholas II and his family
								#  ( except for Alexei and Maria ) are discovered .
								#  The voice of Nicholas's young son , Tsarevich Alexei Nikolaevich , narrates the
								#  remainder of the story . 1883 Western Siberia ,
								#  a young Grigori Rasputin is asked by his father and a group of men to perform magic .
								#  Rasputin has a vision and denounces one of the men as a horse thief . Although his
								#  father initially slaps him for making such an accusation , Rasputin watches as the
								#  man is chased outside and beaten . Twenty years later , Rasputin sees a vision of
								#  the Virgin Mary , prompting him to become a priest . Rasputin quickly becomes famous ,
								#  with people , even a bishop , begging for his blessing . <eod> </s> <eos>

								# fmt: off
								__a										=    [33,1_297,2,1,1_009,4,1_109,11_739,4_762,358,5,25,245,22,1_706,17,20_098,5,3_215,21,37,1_110,3,13,1_041,4,24,603,490,2,71_477,20_098,104_447,2,20_961,1,2_604,4,1,329,3,6_224,831,16_002,2,8,603,78_967,29_546,23,803,20,25,416,5,8,232,4,277,6,1_855,4_601,3,29_546,54,8,3_609,5,57_211,49,4,1,277,18,8,1_755,15_691,3,341,25,416,693,42_573,71,17,401,94,31,17_919,2,29_546,7_873,18,1,435,23,11_011,755,5,5_167,3,7_983,98,84,2,29_546,3_267,8,3_609,4,1,4_865,1_075,2,6_087,71,6,346,8,5_854,3,29_546,824,1_400,1_868,2,19,160,2,311,8,5_496,2,20_920,17,25,15_097,3,24,24,0,33,1,1_857,2,1,1_009,4,1_109,11_739,4_762,358,5,25,245,28,1_110,3,13,1_041,4,24,603,490,2,71_477,20_098,104_447,2,20_961,1,2_604,4,1,329,3,0]  # noqa: E231
								# fmt: on
								#  In 1991, the remains of Russian Tsar Nicholas II and his family (
								#  except for Alexei and Maria ) are discovered. The voice of young son,
								#  Tsarevich Alexei Nikolaevich, narrates the remainder of the story.
								#  1883 Western Siberia, a young Grigori Rasputin is asked by his father
								#  and a group of men to perform magic. Rasputin has a vision and
								#  denounces one of the men as a horse thief. Although his father initially
								#  slaps him for making such an accusation, Rasputin watches as the man
								#  is chased outside and beaten. Twenty years later, Rasputin sees a vision
								#  of the Virgin Mary, prompting him to become a priest.
								#  Rasputin quickly becomes famous, with people, even a bishop, begging for
								#  his blessing. <unk> <unk> <eos> In the 1990s, the remains of Russian Tsar
								# Nicholas II and his family were discovered. The voice of <unk> young son,
								# Tsarevich Alexei Nikolaevich, narrates the remainder of the story.<eos>

								__a										=    model.generate(SCREAMING_SNAKE_CASE__		,     max_length=200		,     do_sample=SCREAMING_SNAKE_CASE__				)
								self.assertListEqual(output_ids[0].numpy().tolist()		,     SCREAMING_SNAKE_CASE__				)



import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
@skip_mps
class  SCREAMING_SNAKE_CASE__  (				lowercase__						,    unittest.TestCase			):
      snake_case__				:     Optional[Any]        =						TextToVideoSDPipeline
      snake_case__				:     Optional[int]        =						TEXT_TO_IMAGE_PARAMS
      snake_case__				:     str        =						TEXT_TO_IMAGE_BATCH_PARAMS
      # No `output_type`.
      snake_case__				:     Optional[Any]        =						frozenset(
          [
              '''num_inference_steps''',
              '''generator''',
              '''latents''',
              '''return_dict''',
              '''callback''',
              '''callback_steps''',
          ]			)
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			Union[str, Any]:
             torch.manual_seed(0     )
             a_					:		Optional[int]							=  UNetaDConditionModel(
                 block_out_channels=(3_2, 6_4, 6_4, 6_4)     ,							layers_per_block=2     ,							sample_size=3_2     ,							in_channels=4     ,							out_channels=4     ,							down_block_types=('CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'DownBlock3D')     ,							up_block_types=('UpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D')     ,							cross_attention_dim=3_2     ,							attention_head_dim=4     ,							)
             a_					:		int							=  DDIMScheduler(
                 beta_start=0.00085     ,							beta_end=0.012     ,							beta_schedule='scaled_linear'     ,							clip_sample=SCREAMING_SNAKE_CASE__     ,							set_alpha_to_one=SCREAMING_SNAKE_CASE__     ,							)
             torch.manual_seed(0     )
             a_					:		int							=  AutoencoderKL(
                 block_out_channels=[3_2, 6_4]     ,							in_channels=3     ,							out_channels=3     ,							down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']     ,							up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']     ,							latent_channels=4     ,							sample_size=1_2_8     ,							)
             torch.manual_seed(0     )
             a_					:		Optional[int]							=  CLIPTextConfig(
                 bos_token_id=0     ,							eos_token_id=2     ,							hidden_size=3_2     ,							intermediate_size=3_7     ,							layer_norm_eps=1E-05     ,							num_attention_heads=4     ,							num_hidden_layers=5     ,							pad_token_id=1     ,							vocab_size=1_0_0_0     ,							hidden_act='gelu'     ,							projection_dim=5_1_2     ,							)
             a_					:		Dict							=  CLIPTextModel(SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

             a_					:		Union[str, Any]							=  {
                 'unet': unet,
                 'scheduler': scheduler,
                 'vae': vae,
                 'text_encoder': text_encoder,
                 'tokenizer': tokenizer,
             }
             return components
      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     ,							SCREAMING_SNAKE_CASE__				:       str     ,							SCREAMING_SNAKE_CASE__				:       List[Any]=0     )		->			List[str]:
             if str(SCREAMING_SNAKE_CASE__     ).startswith('mps'     ):
                    a_					:		Dict							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             else:
                    a_					:		Optional[Any]							=  torch.Generator(device=SCREAMING_SNAKE_CASE__     ).manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		int							=  {
                 'prompt': 'A painting of a squirrel eating a burger',
                 'generator': generator,
                 'num_inference_steps': 2,
                 'guidance_scale': 6.0,
                 'output_type': 'pt',
             }
             return inputs
      def        SCREAMING_SNAKE_CASE							(       self				:       Union[str, Any]     )		->			Tuple:
             a_					:		List[Any]							=  'cpu'  # ensure determinism for the device-dependent torch.Generator
             a_					:		Dict							=  self.get_dummy_components()
             a_					:		str							=  TextToVideoSDPipeline(**SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Tuple							=  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__     )
             a_					:		Dict							=  'np'
             a_					:		Dict							=  sd_pipe(**SCREAMING_SNAKE_CASE__     ).frames
             a_					:		int							=  frames[0][-3:, -3:, -1]

             assert frames[0].shape == (6_4, 6_4, 3)
             a_					:		Union[str, Any]							=  np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0]     )

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-2
      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			List[Any]:
             self._test_attention_slicing_forward_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__     ,							expected_max_diff=3E-3     )
      @unittest.skipIf(
          torch_device != 'cuda' or not is_xformers_available()     ,							reason='XFormers attention is only available with CUDA and `xformers` installed'     ,							)
      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			str:
             self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__     ,							expected_max_diff=1E-2     )
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Union[str, Any]:
             pass
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			List[str]:
             pass
      @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       int     )		->			List[Any]:
             pass

      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Union[str, Any]:
             return super().test_progress_bar()
@slow
@skip_mps
class  SCREAMING_SNAKE_CASE__  (				unittest.TestCase			):
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[Any]     )		->			List[Any]:
             a_					:		str							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'     )

             a_					:		Any							=  TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'     )
             a_					:		Optional[Any]							=  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config     )
             a_					:		Optional[Any]							=  pipe.to('cuda'     )

             a_					:		Any							=  'Spiderman is surfing'
             a_					:		List[Any]							=  torch.Generator(device='cpu'     ).manual_seed(0     )

             a_					:		Optional[Any]							=  pipe(SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=2_5     ,							output_type='pt'     ).frames
             a_					:		str							=  video_frames.cpu().numpy()

             assert np.abs(expected_video - video     ).mean() < 5E-2

      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			Any:
             a_					:		Dict							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'     )

             a_					:		Tuple							=  TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'     )
             a_					:		Tuple							=  pipe.to('cuda'     )

             a_					:		Any							=  'Spiderman is surfing'
             a_					:		List[str]							=  torch.Generator(device='cpu'     ).manual_seed(0     )

             a_					:		List[Any]							=  pipe(SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=2     ,							output_type='pt'     ).frames
             a_					:		List[str]							=  video_frames.cpu().numpy()

             assert np.abs(expected_video - video     ).mean() < 5E-2


