

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   __lowercase				(_UpperCAmelCase			):
 _UpperCamelCase        =  42







class   __lowercase				(_UpperCAmelCase		,       _UpperCAmelCase			):
 @register_to_config
 def __init__(							self				,							A_ = None				,							A_ = 3				,							A_ = 3				,							A_ = False				,							A_ = "positional"				,							A_ = 0				,							A_ = True				,							A_ = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")				,							A_ = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")				,							A_ = (224, 448, 672, 896)				,							A_ = 2				,							A_ = 1				,							A_ = 1				,							A_ = "conv"				,							A_ = "conv"				,							A_ = "silu"				,							A_ = 8				,							A_ = 32				,							A_ = 1e-5				,							A_ = "default"				,							A_ = True				,							A_ = None				,							A_ = None				,							) ->Union[str, Any]:



       '''simple docstring'''





       super().__init__()

       __lowerCAmelCase :						str = sample_size
       __lowerCAmelCase :						List[Any] = block_out_channels[0] * 4

       # Check inputs
       if len(A_ ) != len(A_ ):
             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(A_ ) != len(A_ ):
             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
       __lowerCAmelCase :						Optional[Any] = nn.Convad(A_				,							block_out_channels[0]				,							kernel_size=3				,							padding=(1, 1) )

       # time
       if time_embedding_type == "fourier":
             __lowerCAmelCase :						Tuple = GaussianFourierProjection(embedding_size=block_out_channels[0]				,							scale=16 )
             __lowerCAmelCase :						int = 2 * block_out_channels[0]
       elif time_embedding_type == "positional":
             __lowerCAmelCase :						Any = Timesteps(block_out_channels[0]				,							A_				,							A_ )
             __lowerCAmelCase :						Optional[int] = block_out_channels[0]

       __lowerCAmelCase :						Optional[int] = TimestepEmbedding(A_				,							A_ )

       # class embedding
       if class_embed_type is None and num_class_embeds is not None:
             __lowerCAmelCase :						Optional[int] = nn.Embedding(A_				,							A_ )
       elif class_embed_type == "timestep":
             __lowerCAmelCase :						Optional[int] = TimestepEmbedding(A_				,							A_ )
       elif class_embed_type == "identity":
             __lowerCAmelCase :						int = nn.Identity(A_				,							A_ )
       else:
             __lowerCAmelCase :						int = None

       __lowerCAmelCase :						Tuple = nn.ModuleList([] )
       __lowerCAmelCase :						Any = None
       __lowerCAmelCase :						Any = nn.ModuleList([] )

       # down
       __lowerCAmelCase :						Dict = block_out_channels[0]
       for i, down_block_type in enumerate(A_ ):
             __lowerCAmelCase :						Optional[int] = output_channel
             __lowerCAmelCase :						Tuple = block_out_channels[i]
             __lowerCAmelCase :						List[Any] = i == len(A_ ) - 1

             __lowerCAmelCase :						Tuple = get_down_block(
                 A_				,							num_layers=A_				,							in_channels=A_				,							out_channels=A_				,							temb_channels=A_				,							add_downsample=not is_final_block				,							resnet_eps=A_				,							resnet_act_fn=A_				,							resnet_groups=A_				,							attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel				,							downsample_padding=A_				,							resnet_time_scale_shift=A_				,							downsample_type=A_				,							)
             self.down_blocks.append(A_ )

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

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

             __lowerCAmelCase :						int = i == len(A_ ) - 1

             __lowerCAmelCase :						Union[str, Any] = get_up_block(
                 A_				,							num_layers=layers_per_block + 1				,							in_channels=A_				,							out_channels=A_				,							prev_output_channel=A_				,							temb_channels=A_				,							add_upsample=not is_final_block				,							resnet_eps=A_				,							resnet_act_fn=A_				,							resnet_groups=A_				,							attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel				,							resnet_time_scale_shift=A_				,							upsample_type=A_				,							)
             self.up_blocks.append(A_ )
             __lowerCAmelCase :						List[Any] = output_channel

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





 def      UpperCamelCase__  (							self				,							A_				,							A_				,							A_ = None				,							A_ = True				,							) ->Union[UNetaDOutput, Tuple]:



       '''simple docstring'''





       if self.config.center_input_sample:
             __lowerCAmelCase :						Optional[Any] = 2 * sample - 1.0

       # 1. time
       __lowerCAmelCase :						Dict = timestep
       if not torch.is_tensor(A_ ):
             __lowerCAmelCase :						Dict = torch.tensor([timesteps]				,							dtype=torch.long				,							device=sample.device )
       elif torch.is_tensor(A_ ) and len(timesteps.shape ) == 0:
             __lowerCAmelCase :						Optional[int] = timesteps[None].to(sample.device )

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

       __lowerCAmelCase :						Optional[int] = self.time_proj(A_ )

       # 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.
       __lowerCAmelCase :						Any = t_emb.to(dtype=self.dtype )
       __lowerCAmelCase :						int = self.time_embedding(A_ )

       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":
                   __lowerCAmelCase :						Any = self.time_proj(A_ )

             __lowerCAmelCase :						Optional[int] = self.class_embedding(A_ ).to(dtype=self.dtype )
             __lowerCAmelCase :						Dict = emb + class_emb

       # 2. pre-process
       __lowerCAmelCase :						List[Any] = sample
       __lowerCAmelCase :						Union[str, Any] = self.conv_in(A_ )

       # 3. down
       __lowerCAmelCase :						int = (sample,)
       for downsample_block in self.down_blocks:
             if hasattr(A_				,							'''skip_conv''' ):
                   __lowerCAmelCase,							__lowerCAmelCase,							__lowerCAmelCase :						Optional[Any] = downsample_block(
                       hidden_states=A_				,							temb=A_				,							skip_sample=A_ )
             else:
                   __lowerCAmelCase,							__lowerCAmelCase :						Any = downsample_block(hidden_states=A_				,							temb=A_ )

             down_block_res_samples += res_samples

       # 4. mid
       __lowerCAmelCase :						Optional[Any] = self.mid_block(A_				,							A_ )

       # 5. up
       __lowerCAmelCase :						Union[str, Any] = None
       for upsample_block in self.up_blocks:
             __lowerCAmelCase :						List[str] = down_block_res_samples[-len(upsample_block.resnets ) :]
             __lowerCAmelCase :						int = down_block_res_samples[: -len(upsample_block.resnets )]

             if hasattr(A_				,							'''skip_conv''' ):
                   __lowerCAmelCase,							__lowerCAmelCase :						Dict = upsample_block(A_				,							A_				,							A_				,							A_ )
             else:
                   __lowerCAmelCase :						List[str] = upsample_block(A_				,							A_				,							A_ )

        # 6. post-process
       __lowerCAmelCase :						Any = self.conv_norm_out(A_ )
       __lowerCAmelCase :						Any = self.conv_act(A_ )
       __lowerCAmelCase :						List[Any] = self.conv_out(A_ )

       if skip_sample is not None:
             sample += skip_sample

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

       if not return_dict:
             return (sample,)

       return UNetaDOutput(sample=A_ )





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 BlipImageProcessor







class   __lowercase				(unittest.TestCase			):
 def __init__(							self				,							A_				,							A_=7				,							A_=3				,							A_=18				,							A_=30				,							A_=400				,							A_=True				,							A_=None				,							A_=True				,							A_=False				,							A_=[0.48_145_466, 0.4_578_275, 0.40_821_073]				,							A_=[0.26_862_954, 0.26_130_258, 0.27_577_711]				,							A_=True				,							) ->Dict:



       '''simple docstring'''





       __lowerCAmelCase :						Union[str, Any] = size if size is not None else {'''height''': 20, '''width''': 20}
       __lowerCAmelCase :						List[str] = parent
       __lowerCAmelCase :						List[str] = batch_size
       __lowerCAmelCase :						Optional[Any] = num_channels
       __lowerCAmelCase :						List[Any] = image_size
       __lowerCAmelCase :						Optional[int] = min_resolution
       __lowerCAmelCase :						Optional[int] = max_resolution
       __lowerCAmelCase :						List[Any] = do_resize
       __lowerCAmelCase :						Dict = size
       __lowerCAmelCase :						Dict = do_normalize
       __lowerCAmelCase :						Any = image_mean
       __lowerCAmelCase :						Optional[int] = image_std
       __lowerCAmelCase :						Any = do_pad
       __lowerCAmelCase :						Any = do_convert_rgb
 def      UpperCamelCase__  (							self ) ->List[str]:



       '''simple docstring'''





       return {
           "do_resize": self.do_resize,
           "size": self.size,
           "do_normalize": self.do_normalize,
           "image_mean": self.image_mean,
           "image_std": self.image_std,
           "do_convert_rgb": self.do_convert_rgb,
           "do_pad": self.do_pad,
       }





 def      UpperCamelCase__  (							self				,							A_=False				,							A_=False				,							A_=False ) ->Dict:



       '''simple docstring'''





       assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

       if equal_resolution:
             __lowerCAmelCase :						int = []
             for i in range(self.batch_size ):
                   image_inputs.append(
                       np.random.randint(
                           255				,							size=(self.num_channels, self.max_resolution, self.max_resolution)				,							dtype=np.uinta ) )
       else:
             __lowerCAmelCase :						List[Any] = []
             for i in range(self.batch_size ):
                   __lowerCAmelCase,							__lowerCAmelCase :						Any = np.random.choice(np.arange(self.min_resolution				,							self.max_resolution )				,							2 )
                   image_inputs.append(np.random.randint(255				,							size=(self.num_channels, width, height)				,							dtype=np.uinta ) )

       if not numpify and not torchify:
             # PIL expects the channel dimension as last dimension
             __lowerCAmelCase :						List[str] = [Image.fromarray(np.moveaxis(A_				,							0				,							-1 ) ) for x in image_inputs]

       if torchify:
             __lowerCAmelCase :						int = [torch.from_numpy(A_ ) for x in image_inputs]

       return image_inputs







@require_torch
@require_vision
class   __lowercase				(_UpperCAmelCase		,       unittest.TestCase			):
 _UpperCamelCase        =  BlipImageProcessor if is_vision_available() else None
 def      UpperCamelCase__  (							self ) ->Tuple:



       '''simple docstring'''





       __lowerCAmelCase :						Union[str, Any] = BlipImageProcessingTester(self )
 @property
 def      UpperCamelCase__  (							self ) ->int:



       '''simple docstring'''





       return self.image_processor_tester.prepare_image_processor_dict()
 def      UpperCamelCase__  (							self ) ->int:



       '''simple docstring'''





       __lowerCAmelCase :						str = self.image_processing_class(**self.image_processor_dict )
       self.assertTrue(hasattr(A_				,							'''do_resize''' ) )
       self.assertTrue(hasattr(A_				,							'''size''' ) )
       self.assertTrue(hasattr(A_				,							'''do_normalize''' ) )
       self.assertTrue(hasattr(A_				,							'''image_mean''' ) )
       self.assertTrue(hasattr(A_				,							'''image_std''' ) )
       self.assertTrue(hasattr(A_				,							'''do_convert_rgb''' ) )
 def      UpperCamelCase__  (							self ) ->str:



       '''simple docstring'''





       pass
 def      UpperCamelCase__  (							self ) ->List[Any]:



       '''simple docstring'''





       __lowerCAmelCase :						Dict = self.image_processing_class(**self.image_processor_dict )
       # create random PIL images
       __lowerCAmelCase :						List[str] = self.image_processor_tester.prepare_inputs(equal_resolution=A_ )
       for image in image_inputs:
             self.assertIsInstance(A_				,							Image.Image )

       # Test not batched input
       __lowerCAmelCase :						Optional[Any] = image_processor(image_inputs[0]				,							return_tensors='''pt''' ).pixel_values
       self.assertEqual(
           encoded_images.shape				,							(
               1,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )				,							)

       # Test batched
       __lowerCAmelCase :						str = image_processor(A_				,							return_tensors='''pt''' ).pixel_values
       self.assertEqual(
           encoded_images.shape				,							(
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )				,							)
 def      UpperCamelCase__  (							self ) ->Union[str, Any]:



       '''simple docstring'''





       __lowerCAmelCase :						Optional[Any] = self.image_processing_class(**self.image_processor_dict )
       # create random numpy tensors
       __lowerCAmelCase :						Union[str, Any] = self.image_processor_tester.prepare_inputs(equal_resolution=A_				,							numpify=A_ )
       for image in image_inputs:
             self.assertIsInstance(A_				,							np.ndarray )

       # Test not batched input
       __lowerCAmelCase :						Dict = image_processor(image_inputs[0]				,							return_tensors='''pt''' ).pixel_values
       self.assertEqual(
           encoded_images.shape				,							(
               1,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )				,							)

       # Test batched
       __lowerCAmelCase :						List[Any] = image_processor(A_				,							return_tensors='''pt''' ).pixel_values
       self.assertEqual(
           encoded_images.shape				,							(
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )				,							)





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



       '''simple docstring'''





       __lowerCAmelCase :						List[Any] = self.image_processing_class(**self.image_processor_dict )
       # create random PyTorch tensors
       __lowerCAmelCase :						str = self.image_processor_tester.prepare_inputs(equal_resolution=A_				,							torchify=A_ )
       for image in image_inputs:
             self.assertIsInstance(A_				,							torch.Tensor )

       # Test not batched input
       __lowerCAmelCase :						List[Any] = image_processor(image_inputs[0]				,							return_tensors='''pt''' ).pixel_values
       self.assertEqual(
           encoded_images.shape				,							(
               1,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )				,							)

       # Test batched
       __lowerCAmelCase :						Tuple = image_processor(A_				,							return_tensors='''pt''' ).pixel_values
       self.assertEqual(
           encoded_images.shape				,							(
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )				,							)







@require_torch
@require_vision
class   __lowercase				(_UpperCAmelCase		,       unittest.TestCase			):
 _UpperCamelCase        =  BlipImageProcessor if is_vision_available() else None
 def      UpperCamelCase__  (							self ) ->Optional[Any]:



       '''simple docstring'''





       __lowerCAmelCase :						Any = BlipImageProcessingTester(self				,							num_channels=4 )
       __lowerCAmelCase :						str = 3
 @property
 def      UpperCamelCase__  (							self ) ->List[str]:



       '''simple docstring'''





       return self.image_processor_tester.prepare_image_processor_dict()
 def      UpperCamelCase__  (							self ) ->Tuple:



       '''simple docstring'''





       __lowerCAmelCase :						int = self.image_processing_class(**self.image_processor_dict )
       self.assertTrue(hasattr(A_				,							'''do_resize''' ) )
       self.assertTrue(hasattr(A_				,							'''size''' ) )
       self.assertTrue(hasattr(A_				,							'''do_normalize''' ) )
       self.assertTrue(hasattr(A_				,							'''image_mean''' ) )
       self.assertTrue(hasattr(A_				,							'''image_std''' ) )
       self.assertTrue(hasattr(A_				,							'''do_convert_rgb''' ) )
 def      UpperCamelCase__  (							self ) ->Union[str, Any]:



       '''simple docstring'''





       pass





 def      UpperCamelCase__  (							self ) ->Any:



       '''simple docstring'''





       __lowerCAmelCase :						Tuple = self.image_processing_class(**self.image_processor_dict )
       # create random PIL images
       __lowerCAmelCase :						Optional[int] = self.image_processor_tester.prepare_inputs(equal_resolution=A_ )
       for image in image_inputs:
             self.assertIsInstance(A_				,							Image.Image )

       # Test not batched input
       __lowerCAmelCase :						Tuple = image_processor(image_inputs[0]				,							return_tensors='''pt''' ).pixel_values
       self.assertEqual(
           encoded_images.shape				,							(
               1,
               self.expected_encoded_image_num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )				,							)

       # Test batched
       __lowerCAmelCase :						List[str] = image_processor(A_				,							return_tensors='''pt''' ).pixel_values
       self.assertEqual(
           encoded_images.shape				,							(
               self.image_processor_tester.batch_size,
               self.expected_encoded_image_num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )				,							)


