"""simple docstring"""






from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block


@dataclass
class 					UpperCAmelCase		(_UpperCAmelCase						):



    """simple docstring"""

    _UpperCAmelCase			:torch.FloatTensor



class 					UpperCAmelCase		(_UpperCAmelCase					,_UpperCAmelCase						):



    """simple docstring"""

    @register_to_config
    def __init__(     self					,    _UpperCAmelCase = None					,    _UpperCAmelCase = 3					,    _UpperCAmelCase = 3					,    _UpperCAmelCase = False					,    _UpperCAmelCase = "positional"					,    _UpperCAmelCase = 0					,    _UpperCAmelCase = True					,    _UpperCAmelCase = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")					,    _UpperCAmelCase = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")					,    _UpperCAmelCase = (224, 448, 672, 896)					,    _UpperCAmelCase = 2					,    _UpperCAmelCase = 1					,    _UpperCAmelCase = 1					,    _UpperCAmelCase = "conv"					,    _UpperCAmelCase = "conv"					,    _UpperCAmelCase = "silu"					,    _UpperCAmelCase = 8					,    _UpperCAmelCase = 32					,    _UpperCAmelCase = 1e-5					,    _UpperCAmelCase = "default"					,    _UpperCAmelCase = True					,    _UpperCAmelCase = None					,    _UpperCAmelCase = None					,    ):
       super().__init__()

       lowercase__:     Optional[int]      =   sample_size
       lowercase__:     List[str]      =   block_out_channels[0] * 4

       # Check inputs
       if len(_UpperCAmelCase							) != len(_UpperCAmelCase							):
          raise ValueError(
              F"""Must provide the same number of `down_block_types` as `up_block_types`. `down_block_types`: {down_block_types}. `up_block_types`: {up_block_types}."""							)

       if len(_UpperCAmelCase							) != len(_UpperCAmelCase							):
          raise ValueError(
              F"""Must provide the same number of `block_out_channels` as `down_block_types`. `block_out_channels`: {block_out_channels}. `down_block_types`: {down_block_types}."""							)

       # input
       lowercase__:     int      =   nn.Convad(_UpperCAmelCase					,    block_out_channels[0]					,    kernel_size=3					,    padding=(1, 1)							)

       # time
       if time_embedding_type == "fourier":
          lowercase__:     Optional[Any]      =   GaussianFourierProjection(embedding_size=block_out_channels[0]					,    scale=16							)
          lowercase__:     str      =   2 * block_out_channels[0]
       elif time_embedding_type == "positional":
          lowercase__:     Optional[Any]      =   Timesteps(block_out_channels[0]					,    _UpperCAmelCase					,    _UpperCAmelCase							)
          lowercase__:     str      =   block_out_channels[0]

       lowercase__:     int      =   TimestepEmbedding(_UpperCAmelCase					,    _UpperCAmelCase							)

       # class embedding
       if class_embed_type is None and num_class_embeds is not None:
          lowercase__:     List[Any]      =   nn.Embedding(_UpperCAmelCase					,    _UpperCAmelCase							)
       elif class_embed_type == "timestep":
          lowercase__:     Union[str, Any]      =   TimestepEmbedding(_UpperCAmelCase					,    _UpperCAmelCase							)
       elif class_embed_type == "identity":
          lowercase__:     Optional[Any]      =   nn.Identity(_UpperCAmelCase					,    _UpperCAmelCase							)
       else:
          lowercase__:     str      =   None

       lowercase__:     Optional[Any]      =   nn.ModuleList([]							)
       lowercase__:     Optional[int]      =   None
       lowercase__:     Optional[Any]      =   nn.ModuleList([]							)

       # down
       lowercase__:     Optional[int]      =   block_out_channels[0]
       for i, down_block_type in enumerate(_UpperCAmelCase							):
          lowercase__:     Dict      =   output_channel
          lowercase__:     Optional[int]      =   block_out_channels[i]
          lowercase__:     List[str]      =   i == len(_UpperCAmelCase							) - 1

          lowercase__:     Union[str, Any]      =   get_down_block(
              _UpperCAmelCase					,    num_layers=_UpperCAmelCase					,    in_channels=_UpperCAmelCase					,    out_channels=_UpperCAmelCase					,    temb_channels=_UpperCAmelCase					,    add_downsample=not is_final_block					,    resnet_eps=_UpperCAmelCase					,    resnet_act_fn=_UpperCAmelCase					,    resnet_groups=_UpperCAmelCase					,    attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel					,    downsample_padding=_UpperCAmelCase					,    resnet_time_scale_shift=_UpperCAmelCase					,    downsample_type=_UpperCAmelCase					,    )
          self.down_blocks.append(_UpperCAmelCase							)

       # mid
       lowercase__:     Dict      =   UNetMidBlockaD(
           in_channels=block_out_channels[-1]					,    temb_channels=_UpperCAmelCase					,    resnet_eps=_UpperCAmelCase					,    resnet_act_fn=_UpperCAmelCase					,    output_scale_factor=_UpperCAmelCase					,    resnet_time_scale_shift=_UpperCAmelCase					,    attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1]					,    resnet_groups=_UpperCAmelCase					,    add_attention=_UpperCAmelCase					,    )

       # up
       lowercase__:     Dict      =   list(reversed(_UpperCAmelCase							)							)
       lowercase__:     Optional[Any]      =   reversed_block_out_channels[0]
       for i, up_block_type in enumerate(_UpperCAmelCase							):
          lowercase__:     Any      =   output_channel
          lowercase__:     Tuple      =   reversed_block_out_channels[i]
          lowercase__:     List[Any]      =   reversed_block_out_channels[min(i + 1					,    len(_UpperCAmelCase							) - 1							)]

          lowercase__:     Union[str, Any]      =   i == len(_UpperCAmelCase							) - 1

          lowercase__:     int      =   get_up_block(
              _UpperCAmelCase					,    num_layers=layers_per_block + 1					,    in_channels=_UpperCAmelCase					,    out_channels=_UpperCAmelCase					,    prev_output_channel=_UpperCAmelCase					,    temb_channels=_UpperCAmelCase					,    add_upsample=not is_final_block					,    resnet_eps=_UpperCAmelCase					,    resnet_act_fn=_UpperCAmelCase					,    resnet_groups=_UpperCAmelCase					,    attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel					,    resnet_time_scale_shift=_UpperCAmelCase					,    upsample_type=_UpperCAmelCase					,    )
          self.up_blocks.append(_UpperCAmelCase							)
          lowercase__:     Optional[Any]      =   output_channel

       # out
       lowercase__:     List[Any]      =   norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4					,    32							)
       lowercase__:     Union[str, Any]      =   nn.GroupNorm(num_channels=block_out_channels[0]					,    num_groups=_UpperCAmelCase					,    eps=_UpperCAmelCase							)
       lowercase__:     List[str]      =   nn.SiLU()
       lowercase__:     Union[str, Any]      =   nn.Convad(block_out_channels[0]					,    _UpperCAmelCase					,    kernel_size=3					,    padding=1							)






    def _snake_case    (     self					,    _UpperCAmelCase					,    _UpperCAmelCase					,    _UpperCAmelCase = None					,    _UpperCAmelCase = True					,    ):






       if self.config.center_input_sample:
          lowercase__:     int      =   2 * sample - 1.0

       # 1. time
       lowercase__:     List[Any]      =   timestep
       if not torch.is_tensor(_UpperCAmelCase							):
          lowercase__:     Tuple      =   torch.tensor([timesteps]					,    dtype=torch.long					,    device=sample.device							)
       elif torch.is_tensor(_UpperCAmelCase							) and len(timesteps.shape							) == 0:
          lowercase__:     Tuple      =   timesteps[None].to(sample.device							)

       # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
       lowercase__:     Any      =   timesteps * torch.ones(sample.shape[0]					,    dtype=timesteps.dtype					,    device=timesteps.device							)

       lowercase__:     List[str]      =   self.time_proj(_UpperCAmelCase							)

       # timesteps does not contain any weights and will always return f32 tensors
       # but time_embedding might actually be running in fp16. so we need to cast here.
       # there might be better ways to encapsulate this.
       lowercase__:     List[str]      =   t_emb.to(dtype=self.dtype							)
       lowercase__:     Union[str, Any]      =   self.time_embedding(_UpperCAmelCase							)

       if self.class_embedding is not None:
          if class_labels is None:
             raise ValueError('''class_labels should be provided when doing class conditioning'''							)

          if self.config.class_embed_type == "timestep":
             lowercase__:     str      =   self.time_proj(_UpperCAmelCase							)

          lowercase__:     Any      =   self.class_embedding(_UpperCAmelCase							).to(dtype=self.dtype							)
          lowercase__:     List[str]      =   emb + class_emb

       # 2. pre-process
       lowercase__:     Optional[int]      =   sample
       lowercase__:     str      =   self.conv_in(_UpperCAmelCase							)

       # 3. down
       lowercase__:     List[Any]      =   (sample,)
       for downsample_block in self.down_blocks:
          if hasattr(_UpperCAmelCase					,    '''skip_conv'''							):
             lowercase__:     Union[str, Any]      =   downsample_block(
                 hidden_states=_UpperCAmelCase					,    temb=_UpperCAmelCase					,    skip_sample=_UpperCAmelCase							)
          else:
             lowercase__:     Union[str, Any]      =   downsample_block(hidden_states=_UpperCAmelCase					,    temb=_UpperCAmelCase							)

          down_block_res_samples += res_samples

       # 4. mid
       lowercase__:     Dict      =   self.mid_block(_UpperCAmelCase					,    _UpperCAmelCase							)

       # 5. up
       lowercase__:     List[Any]      =   None
       for upsample_block in self.up_blocks:
          lowercase__:     Dict      =   down_block_res_samples[-len(upsample_block.resnets							) :]
          lowercase__:     Dict      =   down_block_res_samples[: -len(upsample_block.resnets							)]

          if hasattr(_UpperCAmelCase					,    '''skip_conv'''							):
             lowercase__:     int      =   upsample_block(_UpperCAmelCase					,    _UpperCAmelCase					,    _UpperCAmelCase					,    _UpperCAmelCase							)
          else:
             lowercase__:     List[Any]      =   upsample_block(_UpperCAmelCase					,    _UpperCAmelCase					,    _UpperCAmelCase							)

        # 6. post-process
       lowercase__:     List[Any]      =   self.conv_norm_out(_UpperCAmelCase							)
       lowercase__:     Dict      =   self.conv_act(_UpperCAmelCase							)
       lowercase__:     Optional[int]      =   self.conv_out(_UpperCAmelCase							)

       if skip_sample is not None:
          sample += skip_sample

       if self.config.time_embedding_type == "fourier":
          lowercase__:     Any      =   timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:]							)))							)
          lowercase__:     str      =   sample / timesteps

       if not return_dict:
          return (sample,)

       return UNetaDOutput(sample=_UpperCAmelCase							)

"""simple docstring"""






import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
       import torch

if is_vision_available():
       from PIL import Image

       from transformers import TvltImageProcessor



def SCREAMING_SNAKE_CASE__    (     __UpperCAmelCase ,					__UpperCAmelCase=1_0 ,					__UpperCAmelCase=1_0 ,					__UpperCAmelCase=False ,					__UpperCAmelCase=False				)						->						List[str]:
   lowercase__:     List[Any]      =   []
   for i in range(image_processor_tester.num_frames				):
      video.append(np.random.randint(2_5_5 ,					size=(image_processor_tester.num_channels, width, height) ,					dtype=np.uinta				)				)

   if not numpify and not torchify:
      # PIL expects the channel dimension as last dimension
      lowercase__:     Union[str, Any]      =   [Image.fromarray(np.moveaxis(__UpperCAmelCase ,					0 ,					-1				)				) for frame in video]

   if torchify:
      lowercase__:     List[str]      =   [torch.from_numpy(__UpperCAmelCase				) for frame in video]

   return video



def SCREAMING_SNAKE_CASE__    (     __UpperCAmelCase ,					__UpperCAmelCase=False ,					__UpperCAmelCase=False ,					__UpperCAmelCase=False				)						->						Tuple:
   assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

   lowercase__:     int      =   []
   for i in range(image_processor_tester.batch_size				):
      if equal_resolution:
         lowercase__:     Any      =   image_processor_tester.max_resolution
      else:
         lowercase__,				lowercase__:     Union[str, Any]      =   np.random.choice(
             np.arange(image_processor_tester.min_resolution ,					image_processor_tester.max_resolution				) ,					2				)
         lowercase__:     str      =   prepare_video(
             image_processor_tester=__UpperCAmelCase ,					width=__UpperCAmelCase ,					height=__UpperCAmelCase ,					numpify=__UpperCAmelCase ,					torchify=__UpperCAmelCase ,					)
      video_inputs.append(__UpperCAmelCase				)

   return video_inputs


class 					UpperCAmelCase		(unittest.TestCase						):



    """simple docstring"""

    def __init__(     self					,    _UpperCAmelCase					,    _UpperCAmelCase=7					,    _UpperCAmelCase=3					,    _UpperCAmelCase=4					,    _UpperCAmelCase=18					,    _UpperCAmelCase=30					,    _UpperCAmelCase=400					,    _UpperCAmelCase=True					,    _UpperCAmelCase=None					,    _UpperCAmelCase=True					,    _UpperCAmelCase=[0.5, 0.5, 0.5]					,    _UpperCAmelCase=[0.5, 0.5, 0.5]					,    _UpperCAmelCase=True					,    _UpperCAmelCase=None					,    ):
       lowercase__:     Any      =   size if size is not None else {'''shortest_edge''': 18}
       lowercase__:     List[Any]      =   crop_size if crop_size is not None else {'''height''': 18, '''width''': 18}

       lowercase__:     List[Any]      =   parent
       lowercase__:     List[str]      =   batch_size
       lowercase__:     Union[str, Any]      =   num_channels
       lowercase__:     List[Any]      =   num_frames
       lowercase__:     Tuple      =   image_size
       lowercase__:     List[Any]      =   min_resolution
       lowercase__:     Union[str, Any]      =   max_resolution
       lowercase__:     Dict      =   do_resize
       lowercase__:     List[Any]      =   size
       lowercase__:     str      =   do_normalize
       lowercase__:     int      =   image_mean
       lowercase__:     Any      =   image_std
       lowercase__:     List[Any]      =   do_center_crop
       lowercase__:     Optional[int]      =   crop_size






    def _snake_case    (     self							):
       return {
           "image_mean": self.image_mean,
           "image_std": self.image_std,
           "do_normalize": self.do_normalize,
           "do_resize": self.do_resize,
           "size": self.size,
           "do_center_crop": self.do_center_crop,
           "crop_size": self.crop_size,
       }



@require_torch
@require_vision
class 					UpperCAmelCase		(_UpperCAmelCase					,unittest.TestCase						):



    """simple docstring"""

    _UpperCAmelCase			:str							=				TvltImageProcessor if is_vision_available() else None






    def _snake_case    (     self							):
       lowercase__:     List[Any]      =   TvltImageProcessorTester(self							)






    @property
    def _snake_case    (     self							):
       return self.image_processor_tester.prepare_image_processor_dict()






    def _snake_case    (     self							):
       lowercase__:     List[str]      =   self.image_processing_class(**self.image_processor_dict							)
       self.assertTrue(hasattr(_UpperCAmelCase					,    '''image_mean'''							)							)
       self.assertTrue(hasattr(_UpperCAmelCase					,    '''image_std'''							)							)
       self.assertTrue(hasattr(_UpperCAmelCase					,    '''do_normalize'''							)							)
       self.assertTrue(hasattr(_UpperCAmelCase					,    '''do_resize'''							)							)
       self.assertTrue(hasattr(_UpperCAmelCase					,    '''do_center_crop'''							)							)
       self.assertTrue(hasattr(_UpperCAmelCase					,    '''size'''							)							)






    def _snake_case    (     self							):
       # Initialize image_processor
       lowercase__:     List[str]      =   self.image_processing_class(**self.image_processor_dict							)
       # create random PIL videos
       lowercase__:     str      =   prepare_video_inputs(self.image_processor_tester					,    equal_resolution=_UpperCAmelCase							)
       for video in video_inputs:
          self.assertIsInstance(_UpperCAmelCase					,    _UpperCAmelCase							)
          self.assertIsInstance(video[0]					,    Image.Image							)

       # Test not batched input
       lowercase__:     Union[str, Any]      =   image_processor(video_inputs[0]					,    return_tensors='''pt'''							).pixel_values
       self.assertEqual(
           encoded_videos.shape					,    (
               1,
               self.image_processor_tester.num_frames,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.crop_size['''height'''],
               self.image_processor_tester.crop_size['''width'''],
           )					,    )

       # Test batched
       lowercase__:     Tuple      =   image_processor(_UpperCAmelCase					,    return_tensors='''pt'''							).pixel_values
       self.assertEqual(
           encoded_videos.shape					,    (
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_frames,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.crop_size['''height'''],
               self.image_processor_tester.crop_size['''width'''],
           )					,    )






    def _snake_case    (     self							):
       # Initialize image_processor
       lowercase__:     str      =   self.image_processing_class(**self.image_processor_dict							)
       # create random numpy tensors
       lowercase__:     List[str]      =   prepare_video_inputs(self.image_processor_tester					,    equal_resolution=_UpperCAmelCase					,    numpify=_UpperCAmelCase							)
       for video in video_inputs:
          self.assertIsInstance(_UpperCAmelCase					,    _UpperCAmelCase							)
          self.assertIsInstance(video[0]					,    np.ndarray							)

       # Test not batched input
       lowercase__:     Optional[Any]      =   image_processor(video_inputs[0]					,    return_tensors='''pt'''							).pixel_values
       self.assertEqual(
           encoded_videos.shape					,    (
               1,
               self.image_processor_tester.num_frames,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.crop_size['''height'''],
               self.image_processor_tester.crop_size['''width'''],
           )					,    )

       # Test batched
       lowercase__:     Tuple      =   image_processor(_UpperCAmelCase					,    return_tensors='''pt'''							).pixel_values
       self.assertEqual(
           encoded_videos.shape					,    (
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_frames,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.crop_size['''height'''],
               self.image_processor_tester.crop_size['''width'''],
           )					,    )






    def _snake_case    (     self							):
       # Initialize image_processor
       lowercase__:     List[Any]      =   self.image_processing_class(**self.image_processor_dict							)
       # create random PyTorch tensors
       lowercase__:     Optional[int]      =   prepare_video_inputs(self.image_processor_tester					,    equal_resolution=_UpperCAmelCase					,    torchify=_UpperCAmelCase							)
       for video in video_inputs:
          self.assertIsInstance(_UpperCAmelCase					,    _UpperCAmelCase							)
          self.assertIsInstance(video[0]					,    torch.Tensor							)

       # Test not batched input
       lowercase__:     Optional[Any]      =   image_processor(video_inputs[0]					,    return_tensors='''pt'''							).pixel_values
       self.assertEqual(
           encoded_videos.shape					,    (
               1,
               self.image_processor_tester.num_frames,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.crop_size['''height'''],
               self.image_processor_tester.crop_size['''width'''],
           )					,    )

       # Test batched
       lowercase__:     List[Any]      =   image_processor(_UpperCAmelCase					,    return_tensors='''pt'''							).pixel_values
       self.assertEqual(
           encoded_videos.shape					,    (
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_frames,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.crop_size['''height'''],
               self.image_processor_tester.crop_size['''width'''],
           )					,    )
