




import os
import tempfile
import unittest

import numpy as np
import torch

from diffusers.models import ModelMixin, UNetaDConditionModel
from diffusers.models.attention_processor import AttnProcessor, LoRAAttnProcessor
from diffusers.utils import (
    floats_tensor,
    logging,
    skip_mps,
    torch_device,
)
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


enable_full_determinism()

__lowerCamelCase :    Any      						=      logging.get_logger(__name__)







def 		lowercase__ (		__A:						Dict				,__A:						bool = True					):





					'''simple docstring'''






					__magic_name__				:						Any     				=				{}
					for name in model.attn_processors.keys():
										__magic_name__				:						Any     				=				name.endswith('''attn2.processor'''					) and not (
										    name.startswith('''transformer_in'''					) or '''temp_attentions''' in name.split('''.'''					)
										)
										__magic_name__				:						Union[str, Any]     				=				model.config.cross_attention_dim if has_cross_attention else None
										if name.startswith('''mid_block'''					):
															__magic_name__				:						Tuple     				=				model.config.block_out_channels[-1]
										elif name.startswith('''up_blocks'''					):
															__magic_name__				:						Union[str, Any]     				=				int(name[len('''up_blocks.'''					)]					)
															__magic_name__				:						Tuple     				=				list(reversed(model.config.block_out_channels					)					)[block_id]
										elif name.startswith('''down_blocks'''					):
															__magic_name__				:						Any     				=				int(name[len('''down_blocks.'''					)]					)
															__magic_name__				:						str     				=				model.config.block_out_channels[block_id]
										elif name.startswith('''transformer_in'''					):
															# Note that the `8 * ...` comes from: https://github.com/huggingface/diffusers/blob/7139f0e874f10b2463caa8cbd585762a309d12d6/src/diffusers/models/unet_3d_condition.py#L148
															__magic_name__				:						Union[str, Any]     				=				8 * model.config.attention_head_dim

										__magic_name__				:						List[str]     				=				LoRAAttnProcessor(hidden_size=__A				,cross_attention_dim=__A					)
										__magic_name__				:						Optional[Any]     				=				lora_attn_procs[name].to(model.device					)

										if mock_weights:
															# add 1 to weights to mock trained weights
															with torch.no_grad():
																				lora_attn_procs[name].to_q_lora.up.weight += 1
																				lora_attn_procs[name].to_k_lora.up.weight += 1
																				lora_attn_procs[name].to_v_lora.up.weight += 1
																				lora_attn_procs[name].to_out_lora.up.weight += 1

					return lora_attn_procs


@skip_mps
class  lowerCamelCase     (		_lowerCamelCase					,_lowerCamelCase					,unittest.TestCase    ):






			'''simple docstring'''





			UpperCamelCase__										=UNetaDConditionModel
			UpperCamelCase__										='''sample'''

			@property
			def 				UpperCAmelCase__							(			self      : str	)    -> Optional[Any]:
								__magic_name__				:						Optional[int]     				=				4
								__magic_name__				:						Optional[Any]     				=				4
								__magic_name__				:						List[Any]     				=				4
								__magic_name__				:						str     				=				(32, 32)

								__magic_name__				:						Any     				=				floats_tensor((batch_size, num_channels, num_frames) + sizes	).to(lowerCamelCase_	)
								__magic_name__				:						Tuple     				=				torch.tensor([10]	).to(lowerCamelCase_	)
								__magic_name__				:						List[str]     				=				floats_tensor((batch_size, 4, 32)	).to(lowerCamelCase_	)

								return {"sample": noise, "timestep": time_step, "encoder_hidden_states": encoder_hidden_states}

			@property
			def 				UpperCAmelCase__							(			self      : Dict	)    -> Any:
								return (4, 4, 32, 32)

			@property
			def 				UpperCAmelCase__							(			self      : Union[str, Any]	)    -> List[str]:
								return (4, 4, 32, 32)

			def 				UpperCAmelCase__							(			self      : str	)    -> List[str]:
								__magic_name__				:						Optional[int]     				=				{
								    '''block_out_channels''': (32, 64),
								    '''down_block_types''': (
								        '''CrossAttnDownBlock3D''',
								        '''DownBlock3D''',
								    ),
								    '''up_block_types''': ('''UpBlock3D''', '''CrossAttnUpBlock3D'''),
								    '''cross_attention_dim''': 32,
								    '''attention_head_dim''': 8,
								    '''out_channels''': 4,
								    '''in_channels''': 4,
								    '''layers_per_block''': 1,
								    '''sample_size''': 32,
								}
								__magic_name__				:						Any     				=				self.dummy_input
								return init_dict, inputs_dict

			@unittest.skipIf(
			    torch_device != '''cuda''' or not is_xformers_available()							,    reason='''XFormers attention is only available with CUDA and `xformers` installed'''							,    )
			def 				UpperCAmelCase__							(			self      : int	)    -> Optional[Any]:
								__magic_name__				,     __magic_name__				:						Tuple     				=				self.prepare_init_args_and_inputs_for_common()
								__magic_name__				:						Any     				=				self.model_class(**lowerCamelCase_	)

								model.enable_xformers_memory_efficient_attention()

								assert (
								    model.mid_block.attentions[0].transformer_blocks[0].attna.processor.__class__.__name__
								    == "XFormersAttnProcessor"
								), "xformers is not enabled"

			def 				UpperCAmelCase__							(			self      : Any	)    -> Tuple:
								__magic_name__				,     __magic_name__				:						Union[str, Any]     				=				self.prepare_init_args_and_inputs_for_common()

								__magic_name__				:						Union[str, Any]     				=				32

								__magic_name__				:						List[Any]     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)
								model.eval()

								with torch.no_grad():
													__magic_name__				:						Union[str, Any]     				=				model(**lowerCamelCase_	)

													if isinstance(lowerCamelCase_							,    lowerCamelCase_	):
																		__magic_name__				:						Tuple     				=				output.sample

								self.assertIsNotNone(lowerCamelCase_	)
								__magic_name__				:						Tuple     				=				inputs_dict['''sample'''].shape
								self.assertEqual(output.shape							,    lowerCamelCase_							,    '''Input and output shapes do not match'''	)

			def 				UpperCAmelCase__							(			self      : Dict	)    -> List[Any]:
								__magic_name__				,     __magic_name__				:						Optional[int]     				=				self.prepare_init_args_and_inputs_for_common()
								__magic_name__				:						List[Any]     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)
								model.eval()

								with torch.no_grad():
													# Warmup pass when using mps (see #372)
													if torch_device == "mps" and isinstance(lowerCamelCase_							,    lowerCamelCase_	):
																		model(**self.dummy_input	)

													__magic_name__				:						Tuple     				=				model(**lowerCamelCase_	)
													if isinstance(lowerCamelCase_							,    lowerCamelCase_	):
																		__magic_name__				:						List[str]     				=				first.sample

													__magic_name__				:						Dict     				=				model(**lowerCamelCase_	)
													if isinstance(lowerCamelCase_							,    lowerCamelCase_	):
																		__magic_name__				:						Any     				=				second.sample

								__magic_name__				:						Optional[int]     				=				first.cpu().numpy()
								__magic_name__				:						Union[str, Any]     				=				second.cpu().numpy()
								__magic_name__				:						Optional[Any]     				=				out_a[~np.isnan(lowerCamelCase_	)]
								__magic_name__				:						Optional[int]     				=				out_a[~np.isnan(lowerCamelCase_	)]
								__magic_name__				:						int     				=				np.amax(np.abs(out_a - out_a	)	)
								self.assertLessEqual(lowerCamelCase_							,    1E-5	)

			def 				UpperCAmelCase__							(			self      : str	)    -> List[str]:
								__magic_name__				,     __magic_name__				:						Union[str, Any]     				=				self.prepare_init_args_and_inputs_for_common()

								__magic_name__				:						int     				=				8

								__magic_name__				:						Optional[int]     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)
								model.eval()

								model.set_attention_slice('''auto'''	)
								with torch.no_grad():
													__magic_name__				:						str     				=				model(**lowerCamelCase_	)
								assert output is not None

								model.set_attention_slice('''max'''	)
								with torch.no_grad():
													__magic_name__				:						Tuple     				=				model(**lowerCamelCase_	)
								assert output is not None

								model.set_attention_slice(2	)
								with torch.no_grad():
													__magic_name__				:						str     				=				model(**lowerCamelCase_	)
								assert output is not None

			def 				UpperCAmelCase__							(			self      : int	)    -> str:
								__magic_name__				,     __magic_name__				:						Optional[Any]     				=				self.prepare_init_args_and_inputs_for_common()

								__magic_name__				:						str     				=				8

								__magic_name__				:						Any     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)

								with torch.no_grad():
													__magic_name__				:						Any     				=				model(**lowerCamelCase_	).sample

								__magic_name__				:						List[str]     				=				create_lora_layers(lowerCamelCase_	)

								# make sure we can set a list of attention processors
								model.set_attn_processor(lowerCamelCase_	)
								model.to(lowerCamelCase_	)

								# test that attn processors can be set to itself
								model.set_attn_processor(model.attn_processors	)

								with torch.no_grad():
													__magic_name__				:						List[str]     				=				model(**lowerCamelCase_							,    cross_attention_kwargs={'''scale''': 0.0}	).sample
													__magic_name__				:						int     				=				model(**lowerCamelCase_							,    cross_attention_kwargs={'''scale''': 0.5}	).sample
													__magic_name__				:						Any     				=				model(**lowerCamelCase_							,    cross_attention_kwargs={'''scale''': 0.5}	).sample

								assert (samplea - samplea).abs().max() < 3E-3
								assert (samplea - samplea).abs().max() < 3E-3

								# sample 2 and sample 3 should be different
								assert (samplea - samplea).abs().max() > 3E-3

			def 				UpperCAmelCase__							(			self      : Optional[Any]	)    -> int:
								__magic_name__				,     __magic_name__				:						Tuple     				=				self.prepare_init_args_and_inputs_for_common()

								__magic_name__				:						Optional[int]     				=				8

								torch.manual_seed(0	)
								__magic_name__				:						List[Any]     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)

								with torch.no_grad():
													__magic_name__				:						str     				=				model(**lowerCamelCase_	).sample

								__magic_name__				:						Tuple     				=				create_lora_layers(lowerCamelCase_	)
								model.set_attn_processor(lowerCamelCase_	)

								with torch.no_grad():
													__magic_name__				:						int     				=				model(**lowerCamelCase_							,    cross_attention_kwargs={'''scale''': 0.5}	).sample

								with tempfile.TemporaryDirectory() as tmpdirname:
													model.save_attn_procs(lowerCamelCase_	)
													self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase_							,    '''pytorch_lora_weights.bin'''	)	)	)
													torch.manual_seed(0	)
													__magic_name__				:						Union[str, Any]     				=				self.model_class(**lowerCamelCase_	)
													new_model.to(lowerCamelCase_	)
													new_model.load_attn_procs(lowerCamelCase_	)

								with torch.no_grad():
													__magic_name__				:						List[str]     				=				new_model(**lowerCamelCase_							,    cross_attention_kwargs={'''scale''': 0.5}	).sample

								assert (sample - new_sample).abs().max() < 1E-3

								# LoRA and no LoRA should NOT be the same
								assert (sample - old_sample).abs().max() > 1E-4

			def 				UpperCAmelCase__							(			self      : Union[str, Any]	)    -> List[Any]:
								__magic_name__				,     __magic_name__				:						Tuple     				=				self.prepare_init_args_and_inputs_for_common()

								__magic_name__				:						Optional[int]     				=				8

								torch.manual_seed(0	)
								__magic_name__				:						List[Any]     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)

								with torch.no_grad():
													__magic_name__				:						str     				=				model(**lowerCamelCase_	).sample

								__magic_name__				:						Optional[int]     				=				create_lora_layers(lowerCamelCase_	)
								model.set_attn_processor(lowerCamelCase_	)

								with torch.no_grad():
													__magic_name__				:						Tuple     				=				model(**lowerCamelCase_							,    cross_attention_kwargs={'''scale''': 0.5}	).sample

								with tempfile.TemporaryDirectory() as tmpdirname:
													model.save_attn_procs(lowerCamelCase_							,    safe_serialization=lowerCamelCase_	)
													self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase_							,    '''pytorch_lora_weights.safetensors'''	)	)	)
													torch.manual_seed(0	)
													__magic_name__				:						Any     				=				self.model_class(**lowerCamelCase_	)
													new_model.to(lowerCamelCase_	)
													new_model.load_attn_procs(lowerCamelCase_	)

								with torch.no_grad():
													__magic_name__				:						int     				=				new_model(**lowerCamelCase_							,    cross_attention_kwargs={'''scale''': 0.5}	).sample

								assert (sample - new_sample).abs().max() < 3E-3

								# LoRA and no LoRA should NOT be the same
								assert (sample - old_sample).abs().max() > 1E-4

			def 				UpperCAmelCase__							(			self      : Dict	)    -> Dict:
								# enable deterministic behavior for gradient checkpointing
								__magic_name__				,     __magic_name__				:						Any     				=				self.prepare_init_args_and_inputs_for_common()

								__magic_name__				:						Optional[int]     				=				8

								torch.manual_seed(0	)
								__magic_name__				:						Any     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)

								__magic_name__				:						Optional[int]     				=				create_lora_layers(lowerCamelCase_							,    mock_weights=lowerCamelCase_	)
								model.set_attn_processor(lowerCamelCase_	)
								# Saving as torch, properly reloads with directly filename
								with tempfile.TemporaryDirectory() as tmpdirname:
													model.save_attn_procs(lowerCamelCase_	)
													self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase_							,    '''pytorch_lora_weights.bin'''	)	)	)
													torch.manual_seed(0	)
													__magic_name__				:						int     				=				self.model_class(**lowerCamelCase_	)
													new_model.to(lowerCamelCase_	)
													new_model.load_attn_procs(lowerCamelCase_							,    weight_name='''pytorch_lora_weights.bin'''	)

			def 				UpperCAmelCase__							(			self      : int	)    -> Optional[Any]:
								__magic_name__				,     __magic_name__				:						Optional[Any]     				=				self.prepare_init_args_and_inputs_for_common()

								__magic_name__				:						Union[str, Any]     				=				8

								torch.manual_seed(0	)
								__magic_name__				:						int     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)

								__magic_name__				:						Tuple     				=				create_lora_layers(lowerCamelCase_							,    mock_weights=lowerCamelCase_	)
								model.set_attn_processor(lowerCamelCase_	)
								# Saving as torch, properly reloads with directly filename
								with tempfile.TemporaryDirectory() as tmpdirname:
													model.save_attn_procs(lowerCamelCase_	)
													self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase_							,    '''pytorch_lora_weights.bin'''	)	)	)
													torch.manual_seed(0	)
													__magic_name__				:						Tuple     				=				self.model_class(**lowerCamelCase_	)
													new_model.to(lowerCamelCase_	)
													with self.assertRaises(lowerCamelCase_	) as e:
																		new_model.load_attn_procs(lowerCamelCase_							,    use_safetensors=lowerCamelCase_	)
													self.assertIn('''Error no file named pytorch_lora_weights.safetensors'''							,    str(e.exception	)	)

			def 				UpperCAmelCase__							(			self      : Any	)    -> int:
								__magic_name__				,     __magic_name__				:						List[Any]     				=				self.prepare_init_args_and_inputs_for_common()

								__magic_name__				:						str     				=				8

								torch.manual_seed(0	)
								__magic_name__				:						Any     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)

								with torch.no_grad():
													__magic_name__				:						List[str]     				=				model(**lowerCamelCase_	).sample

								__magic_name__				:						List[str]     				=				create_lora_layers(lowerCamelCase_	)
								model.set_attn_processor(lowerCamelCase_	)

								with torch.no_grad():
													__magic_name__				:						str     				=				model(**lowerCamelCase_							,    cross_attention_kwargs={'''scale''': 0.0}	).sample

								model.set_attn_processor(AttnProcessor()	)

								with torch.no_grad():
													__magic_name__				:						Tuple     				=				model(**lowerCamelCase_	).sample

								assert (sample - new_sample).abs().max() < 1E-4
								assert (sample - old_sample).abs().max() < 3E-3

			@unittest.skipIf(
			    torch_device != '''cuda''' or not is_xformers_available()							,    reason='''XFormers attention is only available with CUDA and `xformers` installed'''							,    )
			def 				UpperCAmelCase__							(			self      : List[Any]	)    -> List[Any]:
								# enable deterministic behavior for gradient checkpointing
								__magic_name__				,     __magic_name__				:						Dict     				=				self.prepare_init_args_and_inputs_for_common()

								__magic_name__				:						Tuple     				=				4

								torch.manual_seed(0	)
								__magic_name__				:						str     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)
								__magic_name__				:						Union[str, Any]     				=				create_lora_layers(lowerCamelCase_	)
								model.set_attn_processor(lowerCamelCase_	)

								# default
								with torch.no_grad():
													__magic_name__				:						Optional[int]     				=				model(**lowerCamelCase_	).sample

													model.enable_xformers_memory_efficient_attention()
													__magic_name__				:						Union[str, Any]     				=				model(**lowerCamelCase_	).sample

													model.disable_xformers_memory_efficient_attention()
													__magic_name__				:						Tuple     				=				model(**lowerCamelCase_	).sample

								assert (sample - on_sample).abs().max() < 1E-4
								assert (sample - off_sample).abs().max() < 1E-4


			def 				UpperCAmelCase__							(			self      : Optional[Any]	)    -> str:
								__magic_name__				,     __magic_name__				:						Tuple     				=				self.prepare_init_args_and_inputs_for_common()
								__magic_name__				:						Optional[int]     				=				32

								__magic_name__				:						List[str]     				=				self.model_class(**lowerCamelCase_	)
								model.to(lowerCamelCase_	)
								model.eval()

								with torch.no_grad():
													__magic_name__				:						Tuple     				=				model(**lowerCamelCase_	)[0]

								model.enable_forward_chunking()
								with torch.no_grad():
													__magic_name__				:						Union[str, Any]     				=				model(**lowerCamelCase_	)[0]

								self.assertEqual(output.shape							,    output_a.shape							,    '''Shape doesn\'t match'''	)
								assert np.abs(output.cpu() - output_a.cpu()	).max() < 1E-2


# (todo: sayakpaul) implement SLOW tests.









import math
from collections.abc import Generator







def 		lowercase__ (		__A:						int					):





					'''simple docstring'''






					__magic_name__				:						Generator     				=				(i for i in range(1				,(max_n + 1)					))
					for i in (n for n in numbers if n > 1):
										for j in range(2				,__A					):
															if (i % j) == 0:
																				break
										else:
															yield i







def 		lowercase__ (		__A:						int					):





					'''simple docstring'''






					__magic_name__				:						Generator     				=				(i for i in range(1				,(max_n + 1)					))
					for i in (n for n in numbers if n > 1):
										# only need to check for factors up to sqrt(i)
										__magic_name__				:						Tuple     				=				int(math.sqrt(__A					)					) + 1
										for j in range(2				,__A					):
															if (i % j) == 0:
																				break
										else:
															yield i







def 		lowercase__ (		__A:						int					):





					'''simple docstring'''






					__magic_name__				:						Generator     				=				(i for i in range(1				,(max_n + 1)				,2					))
					# It's useless to test even numbers as they will not be prime
					if max_n > 2:
										yield 2  # Because 2 will not be tested, it's necessary to yield it now
					for i in (n for n in numbers if n > 1):
										__magic_name__				:						str     				=				int(math.sqrt(__A					)					) + 1
										for j in range(3				,__A				,2					):
															# As we removed the even numbers, we don't need them now
															if (i % j) == 0:
																				break
										else:
															yield i







def 		lowercase__ (		):





					'''simple docstring'''






					from timeit import timeit

					__magic_name__				:						Dict     				=				'''from __main__ import slow_primes, primes, fast_primes'''
					print(timeit('''slow_primes(1_000_000_000_000)'''				,setup=__A				,number=1_0_0_0_0_0_0					)					)
					print(timeit('''primes(1_000_000_000_000)'''				,setup=__A				,number=1_0_0_0_0_0_0					)					)
					print(timeit('''fast_primes(1_000_000_000_000)'''				,setup=__A				,number=1_0_0_0_0_0_0					)					)


if __name__ == "__main__":
	__lowerCamelCase :    Union[str, Any]      						=      int(input('''Calculate primes up to:\n>> ''').strip())
	for ret in primes(number):
		print(ret)
	benchmark()



