





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 get_down_block, get_mid_block, get_out_block, get_up_block

@dataclass
class SCREAMING_SNAKE_CASE_ (					lowercase_   ):





					"""simple docstring"""


					A__				 =    42

class SCREAMING_SNAKE_CASE_ (					lowercase_							,       lowercase_   ):





					"""simple docstring"""


					@register_to_config
					def __init__( self	,  _lowerCAmelCase = 6_5536	,  _lowerCAmelCase = None	,  _lowerCAmelCase = 2	,  _lowerCAmelCase = 2	,  _lowerCAmelCase = 0	,  _lowerCAmelCase = "fourier"	,  _lowerCAmelCase = True	,  _lowerCAmelCase = False	,  _lowerCAmelCase = 0.0	,  _lowerCAmelCase = ("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")	,  _lowerCAmelCase = ("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")	,  _lowerCAmelCase = "UNetMidBlock1D"	,  _lowerCAmelCase = None	,  _lowerCAmelCase = (32, 32, 64)	,  _lowerCAmelCase = None	,  _lowerCAmelCase = 8	,  _lowerCAmelCase = 1	,  _lowerCAmelCase = False	,  ):
										super().__init__()
										lowerCamelCase__       =       sample_size

										# time
										if time_embedding_type == "fourier":
															lowerCamelCase__       =       GaussianFourierProjection(
															    embedding_size=8	,  set_W_to_weight=_lowerCAmelCase	,  log=_lowerCAmelCase	,  flip_sin_to_cos=_lowerCAmelCase	)
															lowerCamelCase__       =       2 * block_out_channels[0]
										elif time_embedding_type == "positional":
															lowerCamelCase__       =       Timesteps(
															    block_out_channels[0]	,  flip_sin_to_cos=_lowerCAmelCase	,  downscale_freq_shift=_lowerCAmelCase	)
															lowerCamelCase__       =       block_out_channels[0]

										if use_timestep_embedding:
															lowerCamelCase__       =       block_out_channels[0] * 4
															lowerCamelCase__       =       TimestepEmbedding(
															    in_channels=_lowerCAmelCase	,  time_embed_dim=_lowerCAmelCase	,  act_fn=_lowerCAmelCase	,  out_dim=block_out_channels[0]	,  )

										lowerCamelCase__       =       nn.ModuleList([]	)
										lowerCamelCase__       =       None
										lowerCamelCase__       =       nn.ModuleList([]	)
										lowerCamelCase__       =       None

										# down
										lowerCamelCase__       =       in_channels
										for i, down_block_type in enumerate(_lowerCAmelCase	):
															lowerCamelCase__       =       output_channel
															lowerCamelCase__       =       block_out_channels[i]

															if i == 0:
																				input_channel += extra_in_channels

															lowerCamelCase__       =       i == len(_lowerCAmelCase	) - 1

															lowerCamelCase__       =       get_down_block(
															    _lowerCAmelCase	,  num_layers=_lowerCAmelCase	,  in_channels=_lowerCAmelCase	,  out_channels=_lowerCAmelCase	,  temb_channels=block_out_channels[0]	,  add_downsample=not is_final_block or downsample_each_block	,  )
															self.down_blocks.append(_lowerCAmelCase	)

										# mid
										lowerCamelCase__       =       get_mid_block(
										    _lowerCAmelCase	,  in_channels=block_out_channels[-1]	,  mid_channels=block_out_channels[-1]	,  out_channels=block_out_channels[-1]	,  embed_dim=block_out_channels[0]	,  num_layers=_lowerCAmelCase	,  add_downsample=_lowerCAmelCase	,  )

										# up
										lowerCamelCase__       =       list(reversed(_lowerCAmelCase	)	)
										lowerCamelCase__       =       reversed_block_out_channels[0]
										if out_block_type is None:
															lowerCamelCase__       =       out_channels
										else:
															lowerCamelCase__       =       block_out_channels[0]

										for i, up_block_type in enumerate(_lowerCAmelCase	):
															lowerCamelCase__       =       output_channel
															lowerCamelCase__       =       (
															    reversed_block_out_channels[i + 1] if i < len(_lowerCAmelCase	) - 1 else final_upsample_channels
															)

															lowerCamelCase__       =       i == len(_lowerCAmelCase	) - 1

															lowerCamelCase__       =       get_up_block(
															    _lowerCAmelCase	,  num_layers=_lowerCAmelCase	,  in_channels=_lowerCAmelCase	,  out_channels=_lowerCAmelCase	,  temb_channels=block_out_channels[0]	,  add_upsample=not is_final_block	,  )
															self.up_blocks.append(_lowerCAmelCase	)
															lowerCamelCase__       =       output_channel

										# out
										lowerCamelCase__       =       norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4	,  32	)
										lowerCamelCase__       =       get_out_block(
										    out_block_type=_lowerCAmelCase	,  num_groups_out=_lowerCAmelCase	,  embed_dim=block_out_channels[0]	,  out_channels=_lowerCAmelCase	,  act_fn=_lowerCAmelCase	,  fc_dim=block_out_channels[-1] // 4	,  )






					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	,  _lowerCAmelCase = True	,  ):
										lowerCamelCase__       =       timestep
										if not torch.is_tensor(_lowerCAmelCase	):
															lowerCamelCase__       =       torch.tensor([timesteps]	,  dtype=torch.long	,  device=sample.device	)
										elif torch.is_tensor(_lowerCAmelCase	) and len(timesteps.shape	) == 0:
															lowerCamelCase__       =       timesteps[None].to(sample.device	)

										lowerCamelCase__       =       self.time_proj(_lowerCAmelCase	)
										if self.config.use_timestep_embedding:
															lowerCamelCase__       =       self.time_mlp(_lowerCAmelCase	)
										else:
															lowerCamelCase__       =       timestep_embed[..., None]
															lowerCamelCase__       =       timestep_embed.repeat([1, 1, sample.shape[2]]	).to(sample.dtype	)
															lowerCamelCase__       =       timestep_embed.broadcast_to((sample.shape[:1] + timestep_embed.shape[1:])	)

										# 2. down
										lowerCamelCase__       =       ()
										for downsample_block in self.down_blocks:
															lowerCamelCase__    ,		lowerCamelCase__       =       downsample_block(hidden_states=_lowerCAmelCase	,  temb=_lowerCAmelCase	)
															down_block_res_samples += res_samples

										# 3. mid
										if self.mid_block:
															lowerCamelCase__       =       self.mid_block(_lowerCAmelCase	,  _lowerCAmelCase	)

										# 4. up
										for i, upsample_block in enumerate(self.up_blocks	):
															lowerCamelCase__       =       down_block_res_samples[-1:]
															lowerCamelCase__       =       down_block_res_samples[:-1]
															lowerCamelCase__       =       upsample_block(_lowerCAmelCase	,  res_hidden_states_tuple=_lowerCAmelCase	,  temb=_lowerCAmelCase	)

										# 5. post-process
										if self.out_block:
															lowerCamelCase__       =       self.out_block(_lowerCAmelCase	,  _lowerCAmelCase	)

										if not return_dict:
															return (sample,)

										return UNetaDOutput(sample=_lowerCAmelCase	)








import json
import os
import shutil
import tempfile
import unittest

import numpy as np

from transformers import BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES, BertTokenizer
from transformers.testing_utils import require_tokenizers, require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
						from PIL import Image

						from transformers import VisionTextDualEncoderProcessor, ViTImageProcessor

@require_tokenizers
@require_vision
class SCREAMING_SNAKE_CASE_ (					unittest.TestCase   ):





					"""simple docstring"""


					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       tempfile.mkdtemp()

										# fmt: off
										lowerCamelCase__       =       ["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "want", "##want", "##ed", "wa", "un", "runn", "##ing", ",", "low", "lowest"]
										# fmt: on
										lowerCamelCase__       =       os.path.join(self.tmpdirname	,  VOCAB_FILES_NAMES["vocab_file"]	)
										with open(self.vocab_file	,  "w"	,  encoding="utf-8"	) as vocab_writer:
															vocab_writer.write("".join([x + "\n" for x in vocab_tokens]	)	)

										lowerCamelCase__       =       {
										    "do_resize": True,
										    "size": {"height": 18, "width": 18},
										    "do_normalize": True,
										    "image_mean": [0.5, 0.5, 0.5],
										    "image_std": [0.5, 0.5, 0.5],
										}
										lowerCamelCase__       =       os.path.join(self.tmpdirname	,  _lowerCAmelCase	)
										with open(self.image_processor_file	,  "w"	,  encoding="utf-8"	) as fp:
															json.dump(_lowerCAmelCase	,  _lowerCAmelCase	)





					def 					__magic_name__ ( self	,  **_lowerCAmelCase	):
										return BertTokenizer.from_pretrained(self.tmpdirname	,  **_lowerCAmelCase	)





					def 					__magic_name__ ( self	,  **_lowerCAmelCase	):
										return ViTImageProcessor.from_pretrained(self.tmpdirname	,  **_lowerCAmelCase	)





					def 					__magic_name__ ( self	):
										shutil.rmtree(self.tmpdirname	)





					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       [np.random.randint(255	,  size=(3, 30, 400)	,  dtype=np.uinta	)]

										lowerCamelCase__       =       [Image.fromarray(np.moveaxis(_lowerCAmelCase	,  0	,  -1	)	) for x in image_inputs]

										return image_inputs





					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       self.get_tokenizer()
										lowerCamelCase__       =       self.get_image_processor()

										lowerCamelCase__       =       VisionTextDualEncoderProcessor(tokenizer=_lowerCAmelCase	,  image_processor=_lowerCAmelCase	)

										processor.save_pretrained(self.tmpdirname	)
										lowerCamelCase__       =       VisionTextDualEncoderProcessor.from_pretrained(self.tmpdirname	)

										self.assertEqual(processor.tokenizer.get_vocab()	,  tokenizer.get_vocab()	)
										self.assertIsInstance(processor.tokenizer	,  (BertTokenizer, BertTokenizerFast)	)

										self.assertEqual(processor.image_processor.to_json_string()	,  image_processor.to_json_string()	)
										self.assertIsInstance(processor.image_processor	,  _lowerCAmelCase	)





					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       VisionTextDualEncoderProcessor(
										    tokenizer=self.get_tokenizer()	,  image_processor=self.get_image_processor()	)
										processor.save_pretrained(self.tmpdirname	)

										lowerCamelCase__       =       self.get_tokenizer(bos_token="(BOS)"	,  eos_token="(EOS)"	)
										lowerCamelCase__       =       self.get_image_processor(do_normalize=_lowerCAmelCase	,  padding_value=1.0	)

										lowerCamelCase__       =       VisionTextDualEncoderProcessor.from_pretrained(
										    self.tmpdirname	,  bos_token="(BOS)"	,  eos_token="(EOS)"	,  do_normalize=_lowerCAmelCase	,  padding_value=1.0	)

										self.assertEqual(processor.tokenizer.get_vocab()	,  tokenizer_add_kwargs.get_vocab()	)
										self.assertIsInstance(processor.tokenizer	,  (BertTokenizer, BertTokenizerFast)	)

										self.assertEqual(processor.image_processor.to_json_string()	,  image_processor_add_kwargs.to_json_string()	)
										self.assertIsInstance(processor.image_processor	,  _lowerCAmelCase	)





					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       self.get_image_processor()
										lowerCamelCase__       =       self.get_tokenizer()

										lowerCamelCase__       =       VisionTextDualEncoderProcessor(tokenizer=_lowerCAmelCase	,  image_processor=_lowerCAmelCase	)

										lowerCamelCase__       =       self.prepare_image_inputs()

										lowerCamelCase__       =       image_processor(_lowerCAmelCase	,  return_tensors="np"	)
										lowerCamelCase__       =       processor(images=_lowerCAmelCase	,  return_tensors="np"	)

										for key in input_feat_extract.keys():
															self.assertAlmostEqual(input_feat_extract[key].sum()	,  input_processor[key].sum()	,  delta=1E-2	)





					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       self.get_image_processor()
										lowerCamelCase__       =       self.get_tokenizer()

										lowerCamelCase__       =       VisionTextDualEncoderProcessor(tokenizer=_lowerCAmelCase	,  image_processor=_lowerCAmelCase	)

										lowerCamelCase__       =       "lower newer"

										lowerCamelCase__       =       processor(text=_lowerCAmelCase	)

										lowerCamelCase__       =       tokenizer(_lowerCAmelCase	)

										for key in encoded_tok.keys():
															self.assertListEqual(encoded_tok[key]	,  encoded_processor[key]	)





					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       self.get_image_processor()
										lowerCamelCase__       =       self.get_tokenizer()

										lowerCamelCase__       =       VisionTextDualEncoderProcessor(tokenizer=_lowerCAmelCase	,  image_processor=_lowerCAmelCase	)

										lowerCamelCase__       =       "lower newer"
										lowerCamelCase__       =       self.prepare_image_inputs()

										lowerCamelCase__       =       processor(text=_lowerCAmelCase	,  images=_lowerCAmelCase	)

										self.assertListEqual(list(inputs.keys()	)	,  ["input_ids", "token_type_ids", "attention_mask", "pixel_values"]	)

										# test if it raises when no input is passed
										with self.assertRaises(_lowerCAmelCase	):
															processor()





					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       self.get_image_processor()
										lowerCamelCase__       =       self.get_tokenizer()

										lowerCamelCase__       =       VisionTextDualEncoderProcessor(tokenizer=_lowerCAmelCase	,  image_processor=_lowerCAmelCase	)

										lowerCamelCase__       =       [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

										lowerCamelCase__       =       processor.batch_decode(_lowerCAmelCase	)
										lowerCamelCase__       =       tokenizer.batch_decode(_lowerCAmelCase	)

										self.assertListEqual(_lowerCAmelCase	,  _lowerCAmelCase	)






					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       self.get_image_processor()
										lowerCamelCase__       =       self.get_tokenizer()

										lowerCamelCase__       =       VisionTextDualEncoderProcessor(tokenizer=_lowerCAmelCase	,  image_processor=_lowerCAmelCase	)

										lowerCamelCase__       =       "lower newer"
										lowerCamelCase__       =       self.prepare_image_inputs()

										lowerCamelCase__       =       processor(text=_lowerCAmelCase	,  images=_lowerCAmelCase	)

										self.assertListEqual(list(inputs.keys()	)	,  processor.model_input_names	)

