





"""simple docstring"""





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    _lowerCAmelCase  (	a    ,  unittest.TestCase			):
       """simple docstring"""




       __magic_name__	:Dict						    =   TextToVideoSDPipeline
       __magic_name__	:Any						    =   TEXT_TO_IMAGE_PARAMS
       __magic_name__	:Any						    =   TEXT_TO_IMAGE_BATCH_PARAMS
       # No `output_type`.
       __magic_name__	:Union[str, Any]						    =   frozenset(
           [
               """num_inference_steps""",
               """generator""",
               """latents""",
               """return_dict""",
               """callback""",
               """callback_steps""",
           ]			)






       def 	snake_case		(       self ):



         '''simple docstring'''

         torch.manual_seed(0 )
         lowerCAmelCase__     :Any											=				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       ,					)
         lowerCAmelCase__     :List[str]											=				DDIMScheduler(
             beta_start=0.0_00_85       ,					beta_end=0.0_12       ,					beta_schedule='scaled_linear'       ,					clip_sample=__UpperCAmelCase       ,					set_alpha_to_one=__UpperCAmelCase       ,					)
         torch.manual_seed(0 )
         lowerCAmelCase__     :str											=				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 )
         lowerCAmelCase__     :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       ,					)
         lowerCAmelCase__     :Tuple											=				CLIPTextModel(__UpperCAmelCase )
         lowerCAmelCase__     :Tuple											=				CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

         lowerCAmelCase__     :int											=				{
             'unet': unet,
             'scheduler': scheduler,
             'vae': vae,
             'text_encoder': text_encoder,
             'tokenizer': tokenizer,
         }
         return components






       def 	snake_case		(       self       ,					__UpperCAmelCase       ,					__UpperCAmelCase=0 ):



         '''simple docstring'''

         if str(__UpperCAmelCase ).startswith('mps' ):
           lowerCAmelCase__     :Any											=				torch.manual_seed(__UpperCAmelCase )
         else:
           lowerCAmelCase__     :List[Any]											=				torch.Generator(device=__UpperCAmelCase ).manual_seed(__UpperCAmelCase )
         lowerCAmelCase__     :List[Any]											=				{
             '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 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Optional[int]											=				'cpu'  # ensure determinism for the device-dependent torch.Generator
         lowerCAmelCase__     :Any											=				self.get_dummy_components()
         lowerCAmelCase__     :Tuple											=				TextToVideoSDPipeline(**__UpperCAmelCase )
         lowerCAmelCase__     :Optional[Any]											=				sd_pipe.to(__UpperCAmelCase )
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase )

         lowerCAmelCase__     :Optional[Any]											=				self.get_dummy_inputs(__UpperCAmelCase )
         lowerCAmelCase__     :Optional[int]											=				'np'
         lowerCAmelCase__     :int											=				sd_pipe(**__UpperCAmelCase ).frames
         lowerCAmelCase__     :int											=				frames[0][-3:, -3:, -1]

         assert frames[0].shape == (6_4, 6_4, 3)
         lowerCAmelCase__     :int											=				np.array([1_58.0, 1_60.0, 1_53.0, 1_25.0, 1_00.0, 1_21.0, 1_11.0, 93.0, 1_13.0] )

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1E-2






       def 	snake_case		(       self ):



         '''simple docstring'''

         self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__UpperCAmelCase       ,					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 	snake_case		(       self ):



         '''simple docstring'''

         self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__UpperCAmelCase       ,					expected_max_diff=1E-2 )






       @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.' )
       def 	snake_case		(       self ):



         '''simple docstring'''

         pass






       @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.' )
       def 	snake_case		(       self ):



         '''simple docstring'''

         pass






       @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.' )
       def 	snake_case		(       self ):



         '''simple docstring'''

         pass






       def 	snake_case		(       self ):



         '''simple docstring'''

         return super().test_progress_bar()






@slow
@skip_mps
class    _lowerCAmelCase  (	unittest.TestCase			):
       """simple docstring"""




       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Union[str, Any]											=				load_numpy(
             'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy' )

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

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

         lowerCAmelCase__     :str											=				pipe(__UpperCAmelCase       ,					generator=__UpperCAmelCase       ,					num_inference_steps=2_5       ,					output_type='pt' ).frames
         lowerCAmelCase__     :int											=				video_frames.cpu().numpy()

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






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Union[str, Any]											=				load_numpy(
             'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy' )

         lowerCAmelCase__     :List[str]											=				TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b' )
         lowerCAmelCase__     :Tuple											=				pipe.to('cuda' )

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

         lowerCAmelCase__     :Tuple											=				pipe(__UpperCAmelCase       ,					generator=__UpperCAmelCase       ,					num_inference_steps=2       ,					output_type='pt' ).frames
         lowerCAmelCase__     :List[str]											=				video_frames.cpu().numpy()

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





import numpy

# List of input, output pairs
a_								=       (
    ((5, 2, 3), 15),
    ((6, 5, 9), 25),
    ((11, 12, 13), 41),
    ((1, 1, 1), 8),
    ((11, 12, 13), 41),
)
a_								=       (((515, 22, 13), 555), ((61, 35, 49), 150))
a_								=       [2, 4, 1, 5]
a_								=       len(train_data)
a_								=       0.0_09





def 		a__      (		_UpperCamelCase      :			str			,_UpperCamelCase      :			List[Any]="train"				):
   return calculate_hypothesis_value(_UpperCamelCase			,_UpperCamelCase				) - output(
       _UpperCamelCase			,_UpperCamelCase				)





def 		a__      (		_UpperCamelCase      :			Optional[Any]				):
   __lowerCamelCase    				=     0
   for i in range(len(_UpperCamelCase				) - 1				):
      hyp_val += data_input_tuple[i] * parameter_vector[i + 1]
   hyp_val += parameter_vector[0]
   return hyp_val





def 		a__      (		_UpperCamelCase      :			Tuple			,_UpperCamelCase      :			List[Any]				):
   if data_set == "train":
      return train_data[example_no][1]
   elif data_set == "test":
      return test_data[example_no][1]
   return None





def 		a__      (		_UpperCamelCase      :			List[Any]			,_UpperCamelCase      :			Union[str, Any]				):
   if data_set == "train":
      return _hypothesis_value(train_data[example_no][0]				)
   elif data_set == "test":
      return _hypothesis_value(test_data[example_no][0]				)
   return None





def 		a__      (		_UpperCamelCase      :			Dict			,_UpperCamelCase      :			Dict=m				):
   __lowerCamelCase    				=     0
   for i in range(_UpperCamelCase				):
      if index == -1:
         summation_value += _error(_UpperCamelCase				)
      else:
         summation_value += _error(_UpperCamelCase				) * train_data[i][0][index]
   return summation_value





def 		a__      (		_UpperCamelCase      :			Optional[int]				):
   __lowerCamelCase    				=     summation_of_cost_derivative(_UpperCamelCase			,_UpperCamelCase				) / m
   return cost_derivative_value





def 		a__      (		):
   global parameter_vector
   # Tune these values to set a tolerance value for predicted output
   __lowerCamelCase    				=     0.000_002
   __lowerCamelCase    				=     0
   __lowerCamelCase    				=     0
   while True:
      j += 1
      __lowerCamelCase    				=     [0, 0, 0, 0]
      for i in range(0			,len(_UpperCamelCase				)				):
         __lowerCamelCase    				=     get_cost_derivative(i - 1				)
         __lowerCamelCase    				=     (
             parameter_vector[i] - LEARNING_RATE * cost_derivative
         )
      if numpy.allclose(
          _UpperCamelCase			,_UpperCamelCase			,atol=_UpperCamelCase			,rtol=_UpperCamelCase			,):
         break
      __lowerCamelCase    				=     temp_parameter_vector
   print(('''Number of iterations:''', j)				)





def 		a__      (		):
   for i in range(len(_UpperCamelCase				)				):
      print(('''Actual output value:''', output(_UpperCamelCase			,'''test'''				))				)
      print(('''Hypothesis output:''', calculate_hypothesis_value(_UpperCamelCase			,'''test'''				))				)


if __name__ == "__main__":
  run_gradient_descent()
  print("""\nTesting gradient descent for a linear hypothesis function.\n""")
  test_gradient_descent()






