
import tempfile
import unittest

import torch

from diffusers import UNetaDConditionModel
from diffusers.training_utils import EMAModel
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps, torch_device


enable_full_determinism()


class 		__magic_name__   (						unittest.TestCase    ):
		_lowerCAmelCase		=	"hf-internal-testing/tiny-stable-diffusion-pipe"
		_lowerCAmelCase		=	1
		_lowerCAmelCase		=	77
		_lowerCAmelCase		=	32
		_lowerCAmelCase		=	4
		_lowerCAmelCase		=	_lowerCAmelCase		=	64
		_lowerCAmelCase		=	torch.manual_seed(0    )





		def       _A							(							self    :	Optional[Any]						,  lowerCamelCase__    :	List[Any]=0.9_9_9_9				):
							lowerCAmelCase		:					str  	=     UNetaDConditionModel.from_pretrained(self.model_id						,  subfolder='''unet'''				)
							lowerCAmelCase		:					Union[str, Any]  	=     unet.to(lowerCamelCase__				)
							lowerCAmelCase		:					List[Any]  	=     EMAModel(unet.parameters()						,  decay=lowerCamelCase__						,  model_cls=lowerCamelCase__						,  model_config=unet.config				)
							return unet, ema_unet





		def       _A							(							self    :	Union[str, Any]				):
							lowerCAmelCase		:					Union[str, Any]  	=     torch.randn(
							    self.batch_size						,  self.num_in_channels						,  self.latent_height						,  self.latent_width						,  generator=self.generator				).to(lowerCamelCase__				)
							lowerCAmelCase		:					Optional[int]  	=     torch.randint(0						,  1_0_0_0						,  size=(self.batch_size,)						,  generator=self.generator				).to(lowerCamelCase__				)
							lowerCAmelCase		:					List[Any]  	=     torch.randn(
							    self.batch_size						,  self.prompt_length						,  self.text_encoder_hidden_dim						,  generator=self.generator				).to(lowerCamelCase__				)
							return noisy_latents, timesteps, encoder_hidden_states





		def       _A							(							self    :	Tuple						,  lowerCamelCase__    :	Dict				):
							lowerCAmelCase		:					str  	=     {}
							for k, param in unet.state_dict().items():
												lowerCAmelCase		:					Tuple  	=     torch.randn_like(lowerCamelCase__				) + (param * torch.randn_like(lowerCamelCase__				))
												updated_state_dict.update({k: updated_param}				)
							unet.load_state_dict(lowerCamelCase__				)
							return unet





		def       _A							(							self    :	str				):
							lowerCAmelCase		:					Tuple  	=     self.get_models()
							# Take the first (hypothetical) EMA step.
							ema_unet.step(unet.parameters()				)
							assert ema_unet.optimization_step == 1

							# Take two more.
							for _ in range(2				):
												ema_unet.step(unet.parameters()				)
							assert ema_unet.optimization_step == 3





		def       _A							(							self    :	Optional[int]				):
							lowerCAmelCase		:					Any  	=     self.get_models()
							# Since the `unet` is not being updated (i.e., backprop'd)
							# there won't be any difference between the `params` of `unet`
							# and `ema_unet` even if we call `ema_unet.step(unet.parameters())`.
							ema_unet.step(unet.parameters()				)
							lowerCAmelCase		:					Tuple  	=     list(unet.parameters()				)
							for s_param, param in zip(ema_unet.shadow_params						,  lowerCamelCase__				):
												assert torch.allclose(lowerCamelCase__						,  lowerCamelCase__				)

							# The above holds true even if we call `ema.step()` multiple times since
							# `unet` params are still not being updated.
							for _ in range(4				):
												ema_unet.step(unet.parameters()				)
							for s_param, param in zip(ema_unet.shadow_params						,  lowerCamelCase__				):
												assert torch.allclose(lowerCamelCase__						,  lowerCamelCase__				)





		def       _A							(							self    :	str				):
							lowerCAmelCase		:					int  	=     self.get_models()
							# Here we simulate the parameter updates for `unet`. Since there might
							# be some parameters which are initialized to zero we take extra care to
							# initialize their values to something non-zero before the multiplication.
							lowerCAmelCase		:					Tuple  	=     self.simulate_backprop(lowerCamelCase__				)

							# Take the EMA step.
							ema_unet.step(unet_pseudo_updated_step_one.parameters()				)

							# Now the EMA'd parameters won't be equal to the original model parameters.
							lowerCAmelCase		:					Any  	=     list(unet_pseudo_updated_step_one.parameters()				)
							for s_param, param in zip(ema_unet.shadow_params						,  lowerCamelCase__				):
												assert ~torch.allclose(lowerCamelCase__						,  lowerCamelCase__				)

							# Ensure this is the case when we take multiple EMA steps.
							for _ in range(4				):
												ema_unet.step(unet.parameters()				)
							for s_param, param in zip(ema_unet.shadow_params						,  lowerCamelCase__				):
												assert ~torch.allclose(lowerCamelCase__						,  lowerCamelCase__				)





		def       _A							(							self    :	List[Any]				):
							# If we call EMA step after a backpropagation consecutively for two times,
							# the shadow params from those two steps should be different.
							lowerCAmelCase		:					int  	=     self.get_models()

							# First backprop + EMA
							lowerCAmelCase		:					Any  	=     self.simulate_backprop(lowerCamelCase__				)
							ema_unet.step(unet_step_one.parameters()				)
							lowerCAmelCase		:					int  	=     ema_unet.shadow_params

							# Second backprop + EMA
							lowerCAmelCase		:					str  	=     self.simulate_backprop(lowerCamelCase__				)
							ema_unet.step(unet_step_two.parameters()				)
							lowerCAmelCase		:					Any  	=     ema_unet.shadow_params

							for step_one, step_two in zip(lowerCamelCase__						,  lowerCamelCase__				):
												assert ~torch.allclose(lowerCamelCase__						,  lowerCamelCase__				)





		def       _A							(							self    :	List[str]				):
							# If there's no decay even if there are backprops, EMA steps
							# won't take any effect i.e., the shadow params would remain the
							# same.
							lowerCAmelCase		:					Optional[int]  	=     self.get_models(decay=0.0				)
							lowerCAmelCase		:					Tuple  	=     self.simulate_backprop(lowerCamelCase__				)
							ema_unet.step(unet_step_one.parameters()				)
							lowerCAmelCase		:					List[str]  	=     ema_unet.shadow_params

							lowerCAmelCase		:					int  	=     self.simulate_backprop(lowerCamelCase__				)
							ema_unet.step(unet_step_two.parameters()				)
							lowerCAmelCase		:					List[str]  	=     ema_unet.shadow_params

							for step_one, step_two in zip(lowerCamelCase__						,  lowerCamelCase__				):
												assert torch.allclose(lowerCamelCase__						,  lowerCamelCase__				)





		@skip_mps
		def       _A							(							self    :	Union[str, Any]				):
							lowerCAmelCase		:					Tuple  	=     self.get_models()
							lowerCAmelCase		:					Dict  	=     self.get_dummy_inputs()

							with tempfile.TemporaryDirectory() as tmpdir:
												ema_unet.save_pretrained(lowerCamelCase__				)
												lowerCAmelCase		:					Dict  	=     UNetaDConditionModel.from_pretrained(lowerCamelCase__						,  model_cls=lowerCamelCase__				)
												lowerCAmelCase		:					Tuple  	=     loaded_unet.to(unet.device				)

							# Since no EMA step has been performed the outputs should match.
							lowerCAmelCase		:					int  	=     unet(lowerCamelCase__						,  lowerCamelCase__						,  lowerCamelCase__				).sample
							lowerCAmelCase		:					List[Any]  	=     loaded_unet(lowerCamelCase__						,  lowerCamelCase__						,  lowerCamelCase__				).sample

							assert torch.allclose(lowerCamelCase__						,  lowerCamelCase__						,  atol=1E-4				)


import json
import os
from dataclasses import dataclass

import numpy as np
import pyarrow as pa

import datasets
from utils import get_duration


__SCREAMING_SNAKE_CASE			:   List[Any]         =     10_00_00_00_00_00
__SCREAMING_SNAKE_CASE			:   Optional[Any]         =     1_00_00

__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE			:   Any         =     os.path.split(__file__)
__SCREAMING_SNAKE_CASE			:   List[str]         =     os.path.join(RESULTS_BASEPATH, 'results', RESULTS_FILENAME.replace('.py', '.json'))
def     UpperCAmelCase__							(       __magic_name__						:     int							,  __magic_name__						:     int  ):


					'''simple docstring'''

					lowerCAmelCase		:					Optional[int]  	=     pa.Table.from_pydict({'''col''': [0] * chunk_size}  )
					lowerCAmelCase		:					int  	=     pa.concat_tables([table] * (num_examples // chunk_size)  )
					return datasets.Dataset(__magic_name__							,  fingerprint='''table_100B'''  )


@dataclass
class 		__magic_name__   :
		_lowerCAmelCase		=	42
		_lowerCAmelCase		=	42
		_lowerCAmelCase		=	42
		_lowerCAmelCase		=	42





		def       _A							(							self    :	Optional[int]				):
							lowerCAmelCase		:					Union[str, Any]  	=     np.random.default_rng(self.seed				)
							lowerCAmelCase		:					Tuple  	=     rng.integers(low=self.low						,  high=self.high						,  size=self.size				).tolist()





		def __iter__(							self    :	Dict				):
							return iter(self._sampled_values				)





		def __len__(							self    :	Tuple				):
							return self.size



@get_duration
def     UpperCAmelCase__							(       __magic_name__						:     datasets.Dataset  ):


					'''simple docstring'''

					lowerCAmelCase		:					Union[str, Any]  	=     dataset[0]
@get_duration
def     UpperCAmelCase__							(       __magic_name__						:     datasets.Dataset  ):


					'''simple docstring'''

					lowerCAmelCase		:					Any  	=     dataset[-1]
@get_duration
def     UpperCAmelCase__							(       __magic_name__						:     datasets.Dataset  ):


					'''simple docstring'''

					lowerCAmelCase		:					int  	=     dataset[range(len(__magic_name__  ) // 2							,  len(__magic_name__  ) // 2 + 10_24  )]
@get_duration
def     UpperCAmelCase__							(       __magic_name__						:     datasets.Dataset  ):


					'''simple docstring'''

					lowerCAmelCase		:					List[Any]  	=     dataset[RandIter(0							,  len(__magic_name__  )							,  10_24							,  seed=42  )]
def     UpperCAmelCase__							(       ):


					'''simple docstring'''

					lowerCAmelCase		:					Dict  	=     {'''num examples''': SPEED_TEST_N_EXAMPLES}
					lowerCAmelCase		:					Tuple  	=     (get_first_row, get_last_row, get_batch_of_aaaa_rows, get_batch_of_aaaa_random_rows)
					print('''generating dataset'''  )
					lowerCAmelCase		:					Dict  	=     generate_100B_dataset(num_examples=__magic_name__							,  chunk_size=__magic_name__  )
					print('''Functions'''  )
					for func in functions:
										print(func.__name__  )
										lowerCAmelCase		:					List[Any]  	=     func(__magic_name__  )

					with open(__magic_name__							,  '''wb'''  ) as f:
										f.write(json.dumps(__magic_name__  ).encode('''utf-8'''  )  )


if __name__ == "__main__":  # useful to run the profiler
		benchmark_table_100B()
