


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 	_a							(      SCREAMING_SNAKE_CASE__					,     unittest.TestCase      ):
       _lowercase     :      Optional[Any]  				=			TextToVideoSDPipeline
       _lowercase     :      str  				=			TEXT_TO_IMAGE_PARAMS
       _lowercase     :      List[str]  				=			TEXT_TO_IMAGE_BATCH_PARAMS
       # No `output_type`.
       _lowercase     :      List[Any]  				=			frozenset(
           [
               '''num_inference_steps''',
               '''generator''',
               '''latents''',
               '''return_dict''',
               '''callback''',
               '''callback_steps''',
           ]      )

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





         """simple docstring"""







         torch.manual_seed(0							)
         lowercase__						=				UNetaDConditionModel(
             block_out_channels=(32, 64, 64, 64)   ,						layers_per_block=2   ,						sample_size=32   ,						in_channels=4   ,						out_channels=4   ,						down_block_types=('''CrossAttnDownBlock3D''', '''CrossAttnDownBlock3D''', '''CrossAttnDownBlock3D''', '''DownBlock3D''')   ,						up_block_types=('''UpBlock3D''', '''CrossAttnUpBlock3D''', '''CrossAttnUpBlock3D''', '''CrossAttnUpBlock3D''')   ,						cross_attention_dim=32   ,						attention_head_dim=4   ,						)
         lowercase__						=				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							)
         lowercase__						=				AutoencoderKL(
             block_out_channels=[32, 64]   ,						in_channels=3   ,						out_channels=3   ,						down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']   ,						up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']   ,						latent_channels=4   ,						sample_size=128   ,						)
         torch.manual_seed(0							)
         lowercase__						=				CLIPTextConfig(
             bos_token_id=0   ,						eos_token_id=2   ,						hidden_size=32   ,						intermediate_size=37   ,						layer_norm_eps=1E-0_5   ,						num_attention_heads=4   ,						num_hidden_layers=5   ,						pad_token_id=1   ,						vocab_size=1_000   ,						hidden_act='''gelu'''   ,						projection_dim=512   ,						)
         lowercase__						=				CLIPTextModel(_SCREAMING_SNAKE_CASE							)
         lowercase__						=				CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''							)

         lowercase__						=				{
             "unet": unet,
             "scheduler": scheduler,
             "vae": vae,
             "text_encoder": text_encoder,
             "tokenizer": tokenizer,
         }
         return components

       def 				lowerCamelCase_			(							self:     Any   ,						UpperCamelCase_:     Optional[Any]   ,						UpperCamelCase_:     Any=0							)      ->       Optional[Any]:





         """simple docstring"""







         if str(_SCREAMING_SNAKE_CASE							).startswith('''mps'''							):
           lowercase__						=				torch.manual_seed(_SCREAMING_SNAKE_CASE							)
         else:
           lowercase__						=				torch.Generator(device=_SCREAMING_SNAKE_CASE							).manual_seed(_SCREAMING_SNAKE_CASE							)
         lowercase__						=				{
             "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 				lowerCamelCase_			(							self:     Union[str, Any]							)      ->       int:





         """simple docstring"""







         lowercase__						=				"cpu"  # ensure determinism for the device-dependent torch.Generator
         lowercase__						=				self.get_dummy_components()
         lowercase__						=				TextToVideoSDPipeline(**_SCREAMING_SNAKE_CASE							)
         lowercase__						=				sd_pipe.to(_SCREAMING_SNAKE_CASE							)
         sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE							)

         lowercase__						=				self.get_dummy_inputs(_SCREAMING_SNAKE_CASE							)
         lowercase__						=				"np"
         lowercase__						=				sd_pipe(**_SCREAMING_SNAKE_CASE							).frames
         lowercase__						=				frames[0][-3:, -3:, -1]

         assert frames[0].shape == (64, 64, 3)
         lowercase__						=				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 				lowerCamelCase_			(							self:     str							)      ->       Tuple:





         """simple docstring"""







         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 				lowerCamelCase_			(							self:     Optional[int]							)      ->       int:





         """simple docstring"""







         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 				lowerCamelCase_			(							self:     Dict							)      ->       Optional[int]:





         """simple docstring"""







         pass

       @unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''							)
       def 				lowerCamelCase_			(							self:     Optional[Any]							)      ->       Dict:





         """simple docstring"""







         pass

       @unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''							)
       def 				lowerCamelCase_			(							self:     Union[str, Any]							)      ->       List[str]:





         """simple docstring"""







         pass


       def 				lowerCamelCase_			(							self:     Optional[int]							)      ->       Any:





         """simple docstring"""







         return super().test_progress_bar()


@slow
@skip_mps
class 	_a							(      unittest.TestCase      ):

       def 				lowerCamelCase_			(							self:     Optional[int]							)      ->       Optional[Any]:





         """simple docstring"""







         lowercase__						=				load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'''							)

         lowercase__						=				TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''							)
         lowercase__						=				DPMSolverMultistepScheduler.from_config(pipe.scheduler.config							)
         lowercase__						=				pipe.to('''cuda'''							)

         lowercase__						=				"Spiderman is surfing"
         lowercase__						=				torch.Generator(device='''cpu'''							).manual_seed(0							)

         lowercase__						=				pipe(_SCREAMING_SNAKE_CASE   ,						generator=_SCREAMING_SNAKE_CASE   ,						num_inference_steps=25   ,						output_type='''pt'''							).frames
         lowercase__						=				video_frames.cpu().numpy()

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


       def 				lowerCamelCase_			(							self:     List[str]							)      ->       Optional[Any]:





         """simple docstring"""







         lowercase__						=				load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'''							)

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

         lowercase__						=				"Spiderman is surfing"
         lowercase__						=				torch.Generator(device='''cpu'''							).manual_seed(0							)

         lowercase__						=				pipe(_SCREAMING_SNAKE_CASE   ,						generator=_SCREAMING_SNAKE_CASE   ,						num_inference_steps=2   ,						output_type='''pt'''							).frames
         lowercase__						=				video_frames.cpu().numpy()

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






from __future__ import annotations

from collections import namedtuple
from dataclasses import dataclass


@dataclass
class 	_a							:
							_lowercase     :      int
							_lowercase     :      TreeNode | None  				=			None
							_lowercase     :      TreeNode | None  				=			None
lowerCAmelCase	=							namedtuple('CoinsDistribResult', 'moves excess')
def       _a			(				SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		if root is None:
				return 0

		# Validation
		def count_nodes(SCREAMING_SNAKE_CASE					) -> int:
				if node is None:
						return 0

				return count_nodes(node.left					) + count_nodes(node.right					) + 1

		def count_coins(SCREAMING_SNAKE_CASE					) -> int:
				if node is None:
						return 0

				return count_coins(node.left					) + count_coins(node.right					) + node.data

		if count_nodes(SCREAMING_SNAKE_CASE					) != count_coins(SCREAMING_SNAKE_CASE					):
				raise ValueError('''The nodes number should be same as the number of coins'''					)

		# Main calculation
		def get_distrib(SCREAMING_SNAKE_CASE					) -> CoinsDistribResult:

				if node is None:
						return CoinsDistribResult(0						,    1					)

				lowercase__					,		lowercase__						=				get_distrib(node.left					)
				lowercase__					,		lowercase__						=				get_distrib(node.right					)

				lowercase__						=				1 - left_distrib_excess
				lowercase__						=				1 - right_distrib_excess

				lowercase__						=				(
				    left_distrib_moves
				    + right_distrib_moves
				    + abs(SCREAMING_SNAKE_CASE					)
				    + abs(SCREAMING_SNAKE_CASE					)
				)
				lowercase__						=				node.data - coins_to_left - coins_to_right

				return CoinsDistribResult(SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					)

		return get_distrib(SCREAMING_SNAKE_CASE					)[0]


if __name__ == "__main__":
			import doctest

			doctest.testmod()


