





'''simple docstring'''






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()

A :			Tuple							   =      logging.get_logger(__name__)





def    lowercase_   (		lowercase__					,	lowercase__ = True    ) ->Any:
						_snake_case: Optional[int]	       =		{}
						for name in model.attn_processors.keys():
												_snake_case: Optional[Any]	       =		name.endswith('attn2.processor'    ) and not (
												    name.startswith('transformer_in'    ) or 'temp_attentions' in name.split('.'    )
												)
												_snake_case: Any	       =		model.config.cross_attention_dim if has_cross_attention else None
												if name.startswith('mid_block'    ):
																		_snake_case: int	       =		model.config.block_out_channels[-1]
												elif name.startswith('up_blocks'    ):
																		_snake_case: Union[str, Any]	       =		int(name[len('up_blocks.'    )]    )
																		_snake_case: Optional[Any]	       =		list(reversed(model.config.block_out_channels    )    )[block_id]
												elif name.startswith('down_blocks'    ):
																		_snake_case: int	       =		int(name[len('down_blocks.'    )]    )
																		_snake_case: List[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
																		_snake_case: List[str]	       =		8 * model.config.attention_head_dim

												_snake_case: Tuple	       =		LoRAAttnProcessor(hidden_size=lowercase__					,	cross_attention_dim=lowercase__    )
												_snake_case: str	       =		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       (    __UpperCAmelCase  ,				__UpperCAmelCase  ,				unittest.TestCase	):
					_SCREAMING_SNAKE_CASE				 =     UNetaDConditionModel
					_SCREAMING_SNAKE_CASE				 =     "sample"


					@property
					def  SCREAMING_SNAKE_CASE_		(       self						:	Tuple							):





											'''simple docstring'''





											_snake_case: Any	       =		4
											_snake_case: int	       =		4
											_snake_case: Tuple	       =		4
											_snake_case: Tuple	       =		(32, 32)

											_snake_case: Union[str, Any]	       =		floats_tensor((batch_size, num_channels, num_frames) + sizes							).to(__snake_case							)
											_snake_case: List[str]	       =		torch.tensor([10]							).to(__snake_case							)
											_snake_case: int	       =		floats_tensor((batch_size, 4, 32)							).to(__snake_case							)

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


					@property
					def  SCREAMING_SNAKE_CASE_		(       self						:	Tuple							):





											'''simple docstring'''





											return (4, 4, 32, 32)


					@property
					def  SCREAMING_SNAKE_CASE_		(       self						:	str							):





											'''simple docstring'''





											return (4, 4, 32, 32)


					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[Any]							):





											'''simple docstring'''





											_snake_case: List[str]	       =		{
											    '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,
											}
											_snake_case: str	       =		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  SCREAMING_SNAKE_CASE_		(       self						:	Optional[Any]							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: Tuple	       =		self.prepare_init_args_and_inputs_for_common()
											_snake_case: Any	       =		self.model_class(**__snake_case							)

											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  SCREAMING_SNAKE_CASE_		(       self						:	Any							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: Optional[int]	       =		self.prepare_init_args_and_inputs_for_common()

											_snake_case: Optional[Any]	       =		32

											_snake_case: int	       =		self.model_class(**__snake_case							)
											model.to(__snake_case							)
											model.eval()

											with torch.no_grad():
																	_snake_case: List[Any]	       =		model(**__snake_case							)

																	if isinstance(__snake_case							,  __snake_case							):
																							_snake_case: Optional[Any]	       =		output.sample

											self.assertIsNotNone(__snake_case							)
											_snake_case: Optional[int]	       =		inputs_dict['sample'].shape
											self.assertEqual(output.shape							,  __snake_case							,  'Input and output shapes do not match'							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	Tuple							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: str	       =		self.prepare_init_args_and_inputs_for_common()
											_snake_case: Dict	       =		self.model_class(**__snake_case							)
											model.to(__snake_case							)
											model.eval()

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

																	_snake_case: Dict	       =		model(**__snake_case							)
																	if isinstance(__snake_case							,  __snake_case							):
																							_snake_case: Dict	       =		first.sample

																	_snake_case: List[Any]	       =		model(**__snake_case							)
																	if isinstance(__snake_case							,  __snake_case							):
																							_snake_case: int	       =		second.sample

											_snake_case: List[Any]	       =		first.cpu().numpy()
											_snake_case: int	       =		second.cpu().numpy()
											_snake_case: Dict	       =		out_a[~np.isnan(__snake_case							)]
											_snake_case: Optional[int]	       =		out_a[~np.isnan(__snake_case							)]
											_snake_case: int	       =		np.amax(np.abs(out_a - out_a							)							)
											self.assertLessEqual(__snake_case							,  1e-5							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	Tuple							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: str	       =		self.prepare_init_args_and_inputs_for_common()

											_snake_case: str	       =		8

											_snake_case: Dict	       =		self.model_class(**__snake_case							)
											model.to(__snake_case							)
											model.eval()

											model.set_attention_slice('auto'							)
											with torch.no_grad():
																	_snake_case: int	       =		model(**__snake_case							)
											assert output is not None

											model.set_attention_slice('max'							)
											with torch.no_grad():
																	_snake_case: Union[str, Any]	       =		model(**__snake_case							)
											assert output is not None

											model.set_attention_slice(2							)
											with torch.no_grad():
																	_snake_case: Any	       =		model(**__snake_case							)
											assert output is not None


					def  SCREAMING_SNAKE_CASE_		(       self						:	Any							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: Optional[int]	       =		self.prepare_init_args_and_inputs_for_common()

											_snake_case: Tuple	       =		8

											_snake_case: List[str]	       =		self.model_class(**__snake_case							)
											model.to(__snake_case							)

											with torch.no_grad():
																	_snake_case: int	       =		model(**__snake_case							).sample

											_snake_case: Union[str, Any]	       =		create_lora_layers(__snake_case							)

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

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

											with torch.no_grad():
																	_snake_case: Dict	       =		model(**__snake_case							,  cross_attention_kwargs={'scale': 0.0}							).sample
																	_snake_case: Optional[int]	       =		model(**__snake_case							,  cross_attention_kwargs={'scale': 0.5}							).sample
																	_snake_case: str	       =		model(**__snake_case							,  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  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: List[Any]	       =		self.prepare_init_args_and_inputs_for_common()

											_snake_case: int	       =		8

											torch.manual_seed(0							)
											_snake_case: str	       =		self.model_class(**__snake_case							)
											model.to(__snake_case							)

											with torch.no_grad():
																	_snake_case: Optional[int]	       =		model(**__snake_case							).sample

											_snake_case: Union[str, Any]	       =		create_lora_layers(__snake_case							)
											model.set_attn_processor(__snake_case							)

											with torch.no_grad():
																	_snake_case: Optional[int]	       =		model(**__snake_case							,  cross_attention_kwargs={'scale': 0.5}							).sample

											with tempfile.TemporaryDirectory() as tmpdirname:
																	model.save_attn_procs(__snake_case							)
																	self.assertTrue(os.path.isfile(os.path.join(__snake_case							,  'pytorch_lora_weights.bin'							)							)							)
																	torch.manual_seed(0							)
																	_snake_case: str	       =		self.model_class(**__snake_case							)
																	new_model.to(__snake_case							)
																	new_model.load_attn_procs(__snake_case							)

											with torch.no_grad():
																	_snake_case: int	       =		new_model(**__snake_case							,  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  SCREAMING_SNAKE_CASE_		(       self						:	int							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: Union[str, Any]	       =		self.prepare_init_args_and_inputs_for_common()

											_snake_case: Optional[int]	       =		8

											torch.manual_seed(0							)
											_snake_case: str	       =		self.model_class(**__snake_case							)
											model.to(__snake_case							)

											with torch.no_grad():
																	_snake_case: Tuple	       =		model(**__snake_case							).sample

											_snake_case: Tuple	       =		create_lora_layers(__snake_case							)
											model.set_attn_processor(__snake_case							)

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

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

											with torch.no_grad():
																	_snake_case: Tuple	       =		new_model(**__snake_case							,  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  SCREAMING_SNAKE_CASE_		(       self						:	List[str]							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: Dict	       =		self.prepare_init_args_and_inputs_for_common()

											_snake_case: Union[str, Any]	       =		8

											torch.manual_seed(0							)
											_snake_case: Optional[Any]	       =		self.model_class(**__snake_case							)
											model.to(__snake_case							)

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


					def  SCREAMING_SNAKE_CASE_		(       self						:	str							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: Any	       =		self.prepare_init_args_and_inputs_for_common()

											_snake_case: Dict	       =		8

											torch.manual_seed(0							)
											_snake_case: Dict	       =		self.model_class(**__snake_case							)
											model.to(__snake_case							)

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


					def  SCREAMING_SNAKE_CASE_		(       self						:	Union[str, Any]							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: List[str]	       =		self.prepare_init_args_and_inputs_for_common()

											_snake_case: Union[str, Any]	       =		8

											torch.manual_seed(0							)
											_snake_case: int	       =		self.model_class(**__snake_case							)
											model.to(__snake_case							)

											with torch.no_grad():
																	_snake_case: str	       =		model(**__snake_case							).sample

											_snake_case: Dict	       =		create_lora_layers(__snake_case							)
											model.set_attn_processor(__snake_case							)

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

											model.set_attn_processor(AttnProcessor()							)

											with torch.no_grad():
																	_snake_case: str	       =		model(**__snake_case							).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  SCREAMING_SNAKE_CASE_		(       self						:	Dict							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: Any	       =		self.prepare_init_args_and_inputs_for_common()

											_snake_case: str	       =		4

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

											# default
											with torch.no_grad():
																	_snake_case: List[str]	       =		model(**__snake_case							).sample

																	model.enable_xformers_memory_efficient_attention()
																	_snake_case: Dict	       =		model(**__snake_case							).sample

																	model.disable_xformers_memory_efficient_attention()
																	_snake_case: Optional[int]	       =		model(**__snake_case							).sample

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




					def  SCREAMING_SNAKE_CASE_		(       self						:	List[str]							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: Any	       =		self.prepare_init_args_and_inputs_for_common()
											_snake_case: Optional[int]	       =		32

											_snake_case: Any	       =		self.model_class(**__snake_case							)
											model.to(__snake_case							)
											model.eval()

											with torch.no_grad():
																	_snake_case: Union[str, Any]	       =		model(**__snake_case							)[0]

											model.enable_forward_chunking()
											with torch.no_grad():
																	_snake_case: Optional[int]	       =		model(**__snake_case							)[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.












'''simple docstring'''






import math
from collections.abc import Generator





def    lowercase_   (		lowercase__    ) ->Generator[int, None, None]:
						_snake_case: 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					,	lowercase__    ):
																		if (i % j) == 0:
																								break
												else:
																		yield i





def    lowercase_   (		lowercase__    ) ->Generator[int, None, None]:
						_snake_case: 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)
												_snake_case: int	       =		int(math.sqrt(lowercase__    )    ) + 1
												for j in range(2					,	lowercase__    ):
																		if (i % j) == 0:
																								break
												else:
																		yield i





def    lowercase_   (		lowercase__    ) ->Generator[int, None, None]:
						_snake_case: 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):
												_snake_case: List[str]	       =		int(math.sqrt(lowercase__    )    ) + 1
												for j in range(3					,	lowercase__					,	2    ):
																		# As we removed the even numbers, we don't need them now
																		if (i % j) == 0:
																								break
												else:
																		yield i





def    lowercase_   (		) ->Optional[int]:
						from timeit import timeit

						_snake_case: Dict	       =		'from __main__ import slow_primes, primes, fast_primes'
						print(timeit('slow_primes(1_000_000_000_000)'					,	setup=lowercase__					,	number=100_0000    )    )
						print(timeit('primes(1_000_000_000_000)'					,	setup=lowercase__					,	number=100_0000    )    )
						print(timeit('fast_primes(1_000_000_000_000)'					,	setup=lowercase__					,	number=100_0000    )    )


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





