




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

snake_case__          =						logging.get_logger(__name__)




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

										A_ :     int =							LoRAAttnProcessor(hidden_size=lowerCamelCase__			,		cross_attention_dim=lowerCamelCase__			)
										A_ :     List[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     UpperCamelCase_     (a__,      a__,      unittest.TestCase     ):







						"""simple docstring"""


						_lowerCAmelCase    						=					UNetaDConditionModel
						_lowerCAmelCase    						=					'sample'







						@property
						def       _a    (     self      :       int    ):





											"""simple docstring"""







											A_ :     int =							4
											A_ :     Optional[Any] =							4
											A_ :     Any =							4
											A_ :     Dict =							(32, 32)

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

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







						@property
						def       _a    (     self      :       Tuple    ):





											"""simple docstring"""







											return (4, 4, 32, 32)







						@property
						def       _a    (     self      :       Dict    ):





											"""simple docstring"""







											return (4, 4, 32, 32)







						def       _a    (     self      :       Optional[Any]    ):





											"""simple docstring"""







											A_ :     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,
											}
											A_ :     int =							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       _a    (     self      :       Dict    ):





											"""simple docstring"""







											A_ :     Dict =							self.prepare_init_args_and_inputs_for_common()
											A_ :     int =							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       _a    (     self      :       int    ):





											"""simple docstring"""







											A_ :     Any =							self.prepare_init_args_and_inputs_for_common()

											A_ :     List[str] =							32

											A_ :     Dict =							self.model_class(**_lowerCamelCase    )
											model.to(_lowerCamelCase    )
											model.eval()

											with torch.no_grad():
																A_ :     List[Any] =							model(**_lowerCamelCase    )

																if isinstance(_lowerCamelCase							, _lowerCamelCase    ):
																					A_ :     int =							output.sample

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







						def       _a    (     self      :       str    ):





											"""simple docstring"""







											A_ :     List[Any] =							self.prepare_init_args_and_inputs_for_common()
											A_ :     Union[str, 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    )

																A_ :     List[Any] =							model(**_lowerCamelCase    )
																if isinstance(_lowerCamelCase							, _lowerCamelCase    ):
																					A_ :     Optional[int] =							first.sample

																A_ :     Optional[int] =							model(**_lowerCamelCase    )
																if isinstance(_lowerCamelCase							, _lowerCamelCase    ):
																					A_ :     Any =							second.sample

											A_ :     Any =							first.cpu().numpy()
											A_ :     Tuple =							second.cpu().numpy()
											A_ :     Tuple =							out_a[~np.isnan(_lowerCamelCase    )]
											A_ :     List[str] =							out_a[~np.isnan(_lowerCamelCase    )]
											A_ :     Dict =							np.amax(np.abs(out_a - out_a    )    )
											self.assertLessEqual(_lowerCamelCase							, 1E-5    )







						def       _a    (     self      :       Dict    ):





											"""simple docstring"""







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

											A_ :     List[Any] =							8

											A_ :     str =							self.model_class(**_lowerCamelCase    )
											model.to(_lowerCamelCase    )
											model.eval()

											model.set_attention_slice('''auto'''    )
											with torch.no_grad():
																A_ :     Optional[Any] =							model(**_lowerCamelCase    )
											assert output is not None

											model.set_attention_slice('''max'''    )
											with torch.no_grad():
																A_ :     List[Any] =							model(**_lowerCamelCase    )
											assert output is not None

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







						def       _a    (     self      :       List[str]    ):





											"""simple docstring"""







											A_ :     Optional[Any] =							self.prepare_init_args_and_inputs_for_common()

											A_ :     List[Any] =							8

											A_ :     Any =							self.model_class(**_lowerCamelCase    )
											model.to(_lowerCamelCase    )

											with torch.no_grad():
																A_ :     List[str] =							model(**_lowerCamelCase    ).sample

											A_ :     Tuple =							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():
																A_ :     str =							model(**_lowerCamelCase							, cross_attention_kwargs={'''scale''': 0.0}    ).sample
																A_ :     Union[str, Any] =							model(**_lowerCamelCase							, cross_attention_kwargs={'''scale''': 0.5}    ).sample
																A_ :     List[str] =							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       _a    (     self      :       Optional[int]    ):





											"""simple docstring"""







											A_ :     List[Any] =							self.prepare_init_args_and_inputs_for_common()

											A_ :     Tuple =							8

											torch.manual_seed(0    )
											A_ :     List[str] =							self.model_class(**_lowerCamelCase    )
											model.to(_lowerCamelCase    )

											with torch.no_grad():
																A_ :     Tuple =							model(**_lowerCamelCase    ).sample

											A_ :     List[Any] =							create_lora_layers(_lowerCamelCase    )
											model.set_attn_processor(_lowerCamelCase    )

											with torch.no_grad():
																A_ :     str =							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    )
																A_ :     List[str] =							self.model_class(**_lowerCamelCase    )
																new_model.to(_lowerCamelCase    )
																new_model.load_attn_procs(_lowerCamelCase    )

											with torch.no_grad():
																A_ :     Union[str, Any] =							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       _a    (     self      :       int    ):





											"""simple docstring"""







											A_ :     Any =							self.prepare_init_args_and_inputs_for_common()

											A_ :     List[str] =							8

											torch.manual_seed(0    )
											A_ :     Tuple =							self.model_class(**_lowerCamelCase    )
											model.to(_lowerCamelCase    )

											with torch.no_grad():
																A_ :     int =							model(**_lowerCamelCase    ).sample

											A_ :     str =							create_lora_layers(_lowerCamelCase    )
											model.set_attn_processor(_lowerCamelCase    )

											with torch.no_grad():
																A_ :     str =							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    )
																A_ :     str =							self.model_class(**_lowerCamelCase    )
																new_model.to(_lowerCamelCase    )
																new_model.load_attn_procs(_lowerCamelCase    )

											with torch.no_grad():
																A_ :     List[Any] =							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       _a    (     self      :       List[Any]    ):





											"""simple docstring"""







											A_ :     Tuple =							self.prepare_init_args_and_inputs_for_common()

											A_ :     Any =							8

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

											A_ :     Union[str, Any] =							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    )
																A_ :     Optional[int] =							self.model_class(**_lowerCamelCase    )
																new_model.to(_lowerCamelCase    )
																new_model.load_attn_procs(_lowerCamelCase							, weight_name='''pytorch_lora_weights.bin'''    )







						def       _a    (     self      :       List[Any]    ):





											"""simple docstring"""







											A_ :     Any =							self.prepare_init_args_and_inputs_for_common()

											A_ :     Union[str, Any] =							8

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

											A_ :     str =							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    )
																A_ :     List[Any] =							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       _a    (     self      :       Tuple    ):





											"""simple docstring"""







											A_ :     List[Any] =							self.prepare_init_args_and_inputs_for_common()

											A_ :     Optional[Any] =							8

											torch.manual_seed(0    )
											A_ :     List[str] =							self.model_class(**_lowerCamelCase    )
											model.to(_lowerCamelCase    )

											with torch.no_grad():
																A_ :     str =							model(**_lowerCamelCase    ).sample

											A_ :     Any =							create_lora_layers(_lowerCamelCase    )
											model.set_attn_processor(_lowerCamelCase    )

											with torch.no_grad():
																A_ :     Union[str, Any] =							model(**_lowerCamelCase							, cross_attention_kwargs={'''scale''': 0.0}    ).sample

											model.set_attn_processor(AttnProcessor()    )

											with torch.no_grad():
																A_ :     Any =							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       _a    (     self      :       List[str]    ):





											"""simple docstring"""







											A_ :     Tuple =							self.prepare_init_args_and_inputs_for_common()

											A_ :     Optional[int] =							4

											torch.manual_seed(0    )
											A_ :     Dict =							self.model_class(**_lowerCamelCase    )
											model.to(_lowerCamelCase    )
											A_ :     str =							create_lora_layers(_lowerCamelCase    )
											model.set_attn_processor(_lowerCamelCase    )

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

																model.enable_xformers_memory_efficient_attention()
																A_ :     Optional[int] =							model(**_lowerCamelCase    ).sample

																model.disable_xformers_memory_efficient_attention()
																A_ :     Dict =							model(**_lowerCamelCase    ).sample

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







						def       _a    (     self      :       List[str]    ):





											"""simple docstring"""







											A_ :     Optional[Any] =							self.prepare_init_args_and_inputs_for_common()
											A_ :     List[Any] =							32

											A_ :     str =							self.model_class(**_lowerCamelCase    )
											model.to(_lowerCamelCase    )
											model.eval()

											with torch.no_grad():
																A_ :     str =							model(**_lowerCamelCase    )[0]

											model.enable_forward_chunking()
											with torch.no_grad():
																A_ :     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.









'''simple docstring'''




import argparse
import logging
import sys
import time

import tensorflow as tf
from datasets import load_dataset

from transformers import AutoTokenizer, TFAutoModelForSequenceClassification


if __name__ == "__main__":
							snake_case__          =						argparse.ArgumentParser()

							# Hyperparameters sent by the client are passed as command-line arguments to the script.
							parser.add_argument("""--epochs""", type=int, default=1)
							parser.add_argument("""--per_device_train_batch_size""", type=int, default=16)
							parser.add_argument("""--per_device_eval_batch_size""", type=int, default=8)
							parser.add_argument("""--model_name_or_path""", type=str)
							parser.add_argument("""--learning_rate""", type=str, default=5e-5)
							parser.add_argument("""--do_train""", type=bool, default=True)
							parser.add_argument("""--do_eval""", type=bool, default=True)
							parser.add_argument("""--output_dir""", type=str)

							snake_case__       ,						snake_case__          =						parser.parse_known_args()

							# overwrite batch size until we have tf_glue.py
							snake_case__          =						16
							snake_case__          =						16

							# Set up logging
							snake_case__          =						logging.getLogger(__name__)

							logging.basicConfig(
							    level=logging.getLevelName("""INFO"""),
							    handlers=[logging.StreamHandler(sys.stdout)],
							    format="""%(asctime)s - %(name)s - %(levelname)s - %(message)s""",
							)

							# Load model and tokenizer
							snake_case__          =						TFAutoModelForSequenceClassification.from_pretrained(args.model_name_or_path)
							snake_case__          =						AutoTokenizer.from_pretrained(args.model_name_or_path)

							# Load dataset
							snake_case__       ,						snake_case__          =						load_dataset("""imdb""", split=["""train""", """test"""])
							snake_case__          =						train_dataset.shuffle().select(range(50_00))  # smaller the size for train dataset to 5k
							snake_case__          =						test_dataset.shuffle().select(range(5_00))  # smaller the size for test dataset to 500

							# Preprocess train dataset
							snake_case__          =						train_dataset.map(
							    lambda e: tokenizer(e["""text"""], truncation=True, padding="""max_length"""), batched=True
							)
							train_dataset.set_format(type="""tensorflow""", columns=["""input_ids""", """attention_mask""", """label"""])

							snake_case__          =						{
							    x: train_dataset[x].to_tensor(default_value=0, shape=[None, tokenizer.model_max_length])
							    for x in ["""input_ids""", """attention_mask"""]
							}
							snake_case__          =						tf.data.Dataset.from_tensor_slices((train_features, train_dataset["""label"""])).batch(
							    args.per_device_train_batch_size
							)

							# Preprocess test dataset
							snake_case__          =						test_dataset.map(
							    lambda e: tokenizer(e["""text"""], truncation=True, padding="""max_length"""), batched=True
							)
							test_dataset.set_format(type="""tensorflow""", columns=["""input_ids""", """attention_mask""", """label"""])

							snake_case__          =						{
							    x: test_dataset[x].to_tensor(default_value=0, shape=[None, tokenizer.model_max_length])
							    for x in ["""input_ids""", """attention_mask"""]
							}
							snake_case__          =						tf.data.Dataset.from_tensor_slices((test_features, test_dataset["""label"""])).batch(
							    args.per_device_eval_batch_size
							)

							# fine optimizer and loss
							snake_case__          =						tf.keras.optimizers.Adam(learning_rate=args.learning_rate)
							snake_case__          =						tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)
							snake_case__          =						[tf.keras.metrics.SparseCategoricalAccuracy()]
							model.compile(optimizer=optimizer, loss=loss, metrics=metrics)

							snake_case__          =						time.time()
							snake_case__          =						model.fit(tf_train_dataset, epochs=args.epochs, batch_size=args.per_device_train_batch_size)
							snake_case__          =						time.time() - start_train_time

							logger.info("""*** Train ***""")
							logger.info(F'train_runtime = {end_train_time}')
							for key, value in train_results.history.items():
														logger.info(F'  {key} = {value}')



