import numpy as np
import torch
from torch.utils.data import Dataset

from utils import logger


class 	lowerCamelCase_     (   UpperCAmelCase_ ):






    '''simple docstring'''






    def __init__(	self , __lowercase , __lowercase)       ->	Optional[int]:
        __UpperCamelCase     :Any			  =		params

        __UpperCamelCase     :Optional[int]			  =		np.array(__lowercase)
        __UpperCamelCase     :Union[str, Any]			  =		np.array([len(__lowercase) for t in data])

        self.check()
        self.remove_long_sequences()
        self.remove_empty_sequences()
        self.remove_unknown_sequences()
        self.check()
        self.print_statistics()







    def __getitem__(	self , __lowercase)       ->	Union[str, Any]:
        return (self.token_ids[index], self.lengths[index])







    def __len__(	self)       ->	Optional[Any]:
        return len(self.lengths)







    def 			UpperCamelCase__   (	self)       ->	List[Any]:
        assert len(self.token_ids) == len(self.lengths)
        assert all(self.lengths[i] == len(self.token_ids[i]) for i in range(len(self.lengths)))







    def 			UpperCamelCase__   (	self)       ->	Dict:
        __UpperCamelCase     :Optional[Any]			  =		self.params.max_model_input_size
        __UpperCamelCase     :int			  =		self.lengths > max_len
        logger.info(f"""Splitting {sum(__lowercase)} too long sequences.""")

        def divide_chunks(__lowercase , __lowercase):
            return [l[i : i + n] for i in range(0 , len(__lowercase) , __lowercase)]

        __UpperCamelCase     :str			  =		[]
        __UpperCamelCase     :Union[str, Any]			  =		[]
        if self.params.mlm:
            __UpperCamelCase  ,       __UpperCamelCase     :Any			  =		self.params.special_tok_ids['''cls_token'''], self.params.special_tok_ids['''sep_token''']
        else:
            __UpperCamelCase  ,       __UpperCamelCase     :str			  =		self.params.special_tok_ids['''bos_token'''], self.params.special_tok_ids['''eos_token''']

        for seq_, len_ in zip(self.token_ids , self.lengths):
            assert (seq_[0] == cls_id) and (seq_[-1] == sep_id), seq_
            if len_ <= max_len:
                new_tok_ids.append(seq_)
                new_lengths.append(len_)
            else:
                __UpperCamelCase     :Any			  =		[]
                for sub_s in divide_chunks(seq_ , max_len - 2):
                    if sub_s[0] != cls_id:
                        __UpperCamelCase     :Dict			  =		np.insert(__lowercase , 0 , __lowercase)
                    if sub_s[-1] != sep_id:
                        __UpperCamelCase     :Optional[int]			  =		np.insert(__lowercase , len(__lowercase) , __lowercase)
                    assert len(__lowercase) <= max_len
                    assert (sub_s[0] == cls_id) and (sub_s[-1] == sep_id), sub_s
                    sub_seqs.append(__lowercase)

                new_tok_ids.extend(__lowercase)
                new_lengths.extend([len(__lowercase) for l in sub_seqs])

        __UpperCamelCase     :str			  =		np.array(__lowercase)
        __UpperCamelCase     :str			  =		np.array(__lowercase)







    def 			UpperCamelCase__   (	self)       ->	Dict:
        __UpperCamelCase     :Union[str, Any]			  =		len(self)
        __UpperCamelCase     :List[Any]			  =		self.lengths > 11
        __UpperCamelCase     :Optional[Any]			  =		self.token_ids[indices]
        __UpperCamelCase     :List[str]			  =		self.lengths[indices]
        __UpperCamelCase     :Optional[Any]			  =		len(self)
        logger.info(f"""Remove {init_size - new_size} too short (<=11 tokens) sequences.""")







    def 			UpperCamelCase__   (	self)       ->	Union[str, Any]:
        if "unk_token" not in self.params.special_tok_ids:
            return
        else:
            __UpperCamelCase     :Dict			  =		self.params.special_tok_ids['''unk_token''']
        __UpperCamelCase     :int			  =		len(self)
        __UpperCamelCase     :str			  =		np.array([np.count_nonzero(a == unk_token_id) for a in self.token_ids])
        __UpperCamelCase     :Dict			  =		(unk_occs / self.lengths) < 0.5
        __UpperCamelCase     :str			  =		self.token_ids[indices]
        __UpperCamelCase     :int			  =		self.lengths[indices]
        __UpperCamelCase     :List[Any]			  =		len(self)
        logger.info(f"""Remove {init_size - new_size} sequences with a high level of unknown tokens (50%).""")







    def 			UpperCamelCase__   (	self)       ->	int:
        if not self.params.is_master:
            return
        logger.info(f"""{len(self)} sequences""")
        # data_len = sum(self.lengths)
        # nb_unique_tokens = len(Counter(list(chain(*self.token_ids))))
        # logger.info(f'{data_len} tokens ({nb_unique_tokens} unique)')

        # unk_idx = self.params.special_tok_ids['unk_token']
        # nb_unknown = sum([(t==unk_idx).sum() for t in self.token_ids])
        # logger.info(f'{nb_unknown} unknown tokens (covering {100*nb_unknown/data_len:.2f}% of the data)')







    def 			UpperCamelCase__   (	self , __lowercase)       ->	List[Any]:
        __UpperCamelCase     :List[str]			  =		[t[0] for t in batch]
        __UpperCamelCase     :Optional[int]			  =		[t[1] for t in batch]
        assert len(__lowercase) == len(__lowercase)

        # Max for paddings
        __UpperCamelCase     :List[Any]			  =		max(__lowercase)

        # Pad token ids
        if self.params.mlm:
            __UpperCamelCase     :Optional[int]			  =		self.params.special_tok_ids['''pad_token''']
        else:
            __UpperCamelCase     :Any			  =		self.params.special_tok_ids['''unk_token''']
        __UpperCamelCase     :str			  =		[list(t.astype(__lowercase)) + [pad_idx] * (max_seq_len_ - len(__lowercase)) for t in token_ids]
        assert len(tk_) == len(__lowercase)
        assert all(len(__lowercase) == max_seq_len_ for t in tk_)

        __UpperCamelCase     :Optional[int]			  =		torch.tensor(tk_)  # (bs, max_seq_len_)
        __UpperCamelCase     :List[Any]			  =		torch.tensor(__lowercase)  # (bs)
        return tk_t, lg_t









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


