
import json
import os
import pickle
import subprocess
from hashlib import mda
from pathlib import Path
from tempfile import gettempdir
from textwrap import dedent
from types import CodeType, FunctionType
from unittest import TestCase
from unittest.mock import patch

import pytest
from multiprocess import Pool
from packaging import version

import datasets
from datasets.fingerprint import Hasher, fingerprint_transform
from datasets.table import InMemoryTable

from .utils import (
    require_regex,
    require_spacy,
    require_spacy_model,
    require_tiktoken,
    require_torch,
    require_transformers,
)




class      __lowercase	:






	def __init__(self							:   str			,  snake_case							:   Any						)   -> Optional[int]:
							_lowercase					:  Tuple    			=      foo






	def __call__(self							:   List[str]						)   -> List[str]:
							return self.foo




class      __lowercase	(	datasets.Dataset					):






	@fingerprint_transform(inplace=snake_case						)
	def 			_a(self							:   List[str]			,  snake_case							:   Optional[Any]			,  *snake_case							:   str			,  **snake_case							:   int						)   -> Optional[int]:
							return DatasetChild(self.data			,  fingerprint=snake_case						)






	@fingerprint_transform(inplace=snake_case						)
	def 			_a(self							:   int			,  snake_case							:   List[Any]			,  *snake_case							:   Optional[int]			,  **snake_case							:   List[str]						)   -> Optional[int]:
							return DatasetChild(self.data			,  fingerprint=snake_case						)




class      __lowercase	:






	def __init__(self							:   Optional[Any]			,  snake_case							:   Dict						)   -> int:
							_lowercase					:  str    			=      callable






	def __call__(self							:   Union[str, Any]			,  *snake_case							:   str			,  **snake_case							:   Any						)   -> str:
							if self.callable is not None:
													return self.callable(*snake_case			,  **snake_case						)






	def __getstate__(self							:   str						)   -> Tuple:
							raise pickle.PicklingError()




class      __lowercase	(	__snake_case					):






	@require_transformers
	@pytest.mark.integration
	def 			_a(self							:   Optional[Any]						)   -> Optional[int]:
							from transformers import AutoTokenizer

							def encode(snake_case							:   List[str]						):
													return tokenizer(snake_case						)

							# TODO: add hash consistency tests across sessions
							_lowercase					:  Optional[Any]    			=      AutoTokenizer.from_pretrained("bert-base-uncased"						)
							_lowercase					:  Any    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  List[str]    			=      mda(datasets.utils.py_utils.dumps(lambda snake_case						: tokenizer(snake_case						)						)						).hexdigest()
							_lowercase					:  Tuple    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  Any    			=      AutoTokenizer.from_pretrained("bert-base-cased"						)
							_lowercase					:  str    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  Union[str, Any]    			=      mda(datasets.utils.py_utils.dumps(lambda snake_case						: tokenizer(snake_case						)						)						).hexdigest()
							_lowercase					:  Tuple    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  List[Any]    			=      AutoTokenizer.from_pretrained("bert-base-uncased"						)
							_lowercase					:  List[Any]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  int    			=      mda(datasets.utils.py_utils.dumps(lambda snake_case						: tokenizer(snake_case						)						)						).hexdigest()
							_lowercase					:  Optional[int]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)






	@require_transformers
	@pytest.mark.integration
	def 			_a(self							:   List[Any]						)   -> int:
							from transformers import AutoTokenizer

							_lowercase					:  Tuple    			=      AutoTokenizer.from_pretrained("gpt2"						)
							_lowercase					:  str    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							tokenizer("Hello world !"						)  # call once to change the tokenizer's cache
							_lowercase					:  Tuple    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)






	@require_regex
	def 			_a(self							:   str						)   -> List[Any]:
							import regex

							_lowercase					:  Tuple    			=      regex.Regex("foo"						)
							_lowercase					:  List[str]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  Dict    			=      regex.Regex("bar"						)
							_lowercase					:  int    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  int    			=      regex.Regex("foo"						)
							_lowercase					:  int    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)




class      __lowercase	(	__snake_case					):






	def 			_a(self							:   Any						)   -> Dict:
							def func():
													return foo

							_lowercase					:  Any    			=      [0]
							_lowercase					:  Optional[Any]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  List[Any]    			=      [1]
							_lowercase					:  List[str]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  Any    			=      [0]
							_lowercase					:  Any    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)






	def 			_a(self							:   Optional[Any]						)   -> Union[str, Any]:
							def func():
													pass

							_lowercase					:  List[str]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()

							def func():
													pass

							_lowercase					:  int    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)






	def 			_a(self							:   Optional[int]						)   -> List[Any]:
							_lowercase					:  int    			=      mda(datasets.utils.py_utils.dumps(Foo([0]						)						)						).hexdigest()
							_lowercase					:  Optional[int]    			=      mda(datasets.utils.py_utils.dumps(Foo([1]						)						)						).hexdigest()
							_lowercase					:  Optional[Any]    			=      mda(datasets.utils.py_utils.dumps(Foo([0]						)						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)






	def 			_a(self							:   Any						)   -> Union[str, Any]:
							_lowercase					:  Optional[Any]    			=      mda(datasets.utils.py_utils.dumps(Foo([0]						).__call__						)						).hexdigest()
							_lowercase					:  int    			=      mda(datasets.utils.py_utils.dumps(Foo([1]						).__call__						)						).hexdigest()
							_lowercase					:  Any    			=      mda(datasets.utils.py_utils.dumps(Foo([0]						).__call__						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)






	def 			_a(self							:   Any						)   -> Any:
							_lowercase					:  Tuple    			=      (
							    "co_argcount",
							    "co_kwonlyargcount",
							    "co_nlocals",
							    "co_stacksize",
							    "co_flags",
							    "co_code",
							    "co_consts",
							    "co_names",
							    "co_varnames",
							    "co_filename",
							    "co_name",
							    "co_firstlineno",
							    "co_lnotab",
							    "co_freevars",
							    "co_cellvars",
							)

							def _create_code_pyaa(*snake_case							:   List[Any]						):
													if hasattr(args[-3]			,  "encode"						):
																			_lowercase					:  Any    			=      list(snake_case						)
																			_lowercase					:  str    			=      args[-3].encode()  # co_lnotab
																			_lowercase					:  List[str]    			=      args[-10].encode()  # co_code
													if hasattr(snake_case			,  "co_posonlyargcount"						):
																			if len(snake_case						) == 16:
																									return CodeType(*snake_case						)
																			elif len(snake_case						) == 15:
																									return CodeType(args[0]			,  0			,  *args[1:]						)
																			return CodeType(args[0]			,  0			,  0			,  *args[1:]						)
													elif hasattr(snake_case			,  "co_kwonlyargcount"						):
																			if len(snake_case						) == 16:
																									return CodeType(args[0]			,  *args[2:]						)
																			elif len(snake_case						) == 15:
																									return CodeType(*snake_case						)
																			return CodeType(args[0]			,  0			,  *args[1:]						)
													if len(snake_case						) == 16:
																			return CodeType(args[0]			,  *args[3:]						)
													elif len(snake_case						) == 15:
																			return CodeType(args[0]			,  *args[2:]						)
													return CodeType(*snake_case						)

							def create_ipython_func(snake_case							:   Dict			,  snake_case							:   str						):
													def func():
																			return returned_obj

													_lowercase					:  Tuple    			=      func.__code__
													# Use _create_code from dill in order to make it work for different python versions
													if datasets.config.PY_VERSION < version.parse("3.8"						):
																			_lowercase					:  Tuple    			=      _create_code_pyaa(
																			    *[getattr(snake_case			,  snake_case						) if k != "co_filename" else co_filename for k in code_args_pyaa]						)
													else:
																			_lowercase					:  Any    			=      code.replace(co_filename=snake_case						)
													return FunctionType(snake_case			,  func.__globals__			,  func.__name__			,  func.__defaults__			,  func.__closure__						)

							_lowercase			,						_lowercase					:  Union[str, Any]    			=      "<ipython-input-2-e0383a102aae>", [0]
							_lowercase					:  Optional[int]    			=      mda(datasets.utils.py_utils.dumps(create_ipython_func(snake_case			,  snake_case						)						)						).hexdigest()
							_lowercase			,						_lowercase					:  Tuple    			=      "<ipython-input-2-e0383a102aae>", [1]
							_lowercase					:  Dict    			=      mda(datasets.utils.py_utils.dumps(create_ipython_func(snake_case			,  snake_case						)						)						).hexdigest()
							_lowercase			,						_lowercase					:  List[str]    			=      "<ipython-input-5-713f6613acf3>", [0]
							_lowercase					:  Dict    			=      mda(datasets.utils.py_utils.dumps(create_ipython_func(snake_case			,  snake_case						)						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)

							_lowercase			,						_lowercase					:  Dict    			=      os.path.join(gettempdir()			,  "ipykernel_12345"			,  "321456789.py"						), [0]
							_lowercase					:  int    			=      mda(datasets.utils.py_utils.dumps(create_ipython_func(snake_case			,  snake_case						)						)						).hexdigest()
							_lowercase			,						_lowercase					:  Union[str, Any]    			=      os.path.join(gettempdir()			,  "ipykernel_12345"			,  "321456789.py"						), [1]
							_lowercase					:  int    			=      mda(datasets.utils.py_utils.dumps(create_ipython_func(snake_case			,  snake_case						)						)						).hexdigest()
							_lowercase			,						_lowercase					:  List[str]    			=      os.path.join(gettempdir()			,  "ipykernel_12345"			,  "654123987.py"						), [0]
							_lowercase					:  Union[str, Any]    			=      mda(datasets.utils.py_utils.dumps(create_ipython_func(snake_case			,  snake_case						)						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)






	def 			_a(self							:   str						)   -> str:
							_lowercase			,						_lowercase					:  Optional[int]    			=      [0], [1]

							def func():
													return foo, bar

							_lowercase					:  Dict    			=      "__main__"

							def globalvars_mocka_side_effect(snake_case							:   Optional[int]			,  *snake_case							:   str			,  **snake_case							:   Optional[int]						):
													return {"foo": foo, "bar": bar}

							def globalvars_mocka_side_effect(snake_case							:   int			,  *snake_case							:   List[Any]			,  **snake_case							:   str						):
													return {"bar": bar, "foo": foo}

							with patch("dill.detect.globalvars"			,  side_effect=snake_case						) as globalvars_mocka:
													_lowercase					:  List[str]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
													self.assertGreater(globalvars_mocka.call_count			,  0						)
							with patch("dill.detect.globalvars"			,  side_effect=snake_case						) as globalvars_mocka:
													_lowercase					:  str    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
													self.assertGreater(globalvars_mocka.call_count			,  0						)
							self.assertEqual(snake_case			,  snake_case						)




class      __lowercase	(	__snake_case					):






	def 			_a(self							:   Any						)   -> Tuple:
							_lowercase					:  int    			=      Hasher.hash("hello"						)
							_lowercase					:  Optional[int]    			=      Hasher.hash("hello"						)
							_lowercase					:  Dict    			=      Hasher.hash("there"						)
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)






	def 			_a(self							:   Dict						)   -> int:
							_lowercase					:  Union[str, Any]    			=      Hasher.hash(Foo("hello"						)						)
							_lowercase					:  List[Any]    			=      Hasher.hash(Foo("hello"						)						)
							_lowercase					:  Dict    			=      Hasher.hash(Foo("there"						)						)
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)






	def 			_a(self							:   int						)   -> Optional[int]:
							_lowercase					:  int    			=      Hasher()
							for x in ["hello", Foo("hello"						)]:
													hasher.update(snake_case						)
							_lowercase					:  Union[str, Any]    			=      hasher.hexdigest()
							_lowercase					:  Any    			=      Hasher()
							for x in ["hello", Foo("hello"						)]:
													hasher.update(snake_case						)
							_lowercase					:  Union[str, Any]    			=      hasher.hexdigest()
							_lowercase					:  Optional[Any]    			=      Hasher()
							for x in ["there", Foo("there"						)]:
													hasher.update(snake_case						)
							_lowercase					:  List[str]    			=      hasher.hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)






	def 			_a(self							:   str						)   -> Optional[int]:
							with self.assertRaises(pickle.PicklingError						):
													Hasher.hash(UnpicklableCallable(Foo("hello"						)						)						)






	def 			_a(self							:   Optional[Any]						)   -> Any:
							_lowercase					:  int    			=      "abc"
							_lowercase					:  Tuple    			=      [string, string]  # two strings have the same ids
							_lowercase					:  Any    			=      [string, string]
							_lowercase					:  Optional[Any]    			=      json.loads(F"""[\"{string}\", \"{string}\"]"""						)  # two strings have different ids
							self.assertIs(obja[0]			,  snake_case						)
							self.assertIs(obja[0]			,  obja[1]						)
							self.assertIs(obja[0]			,  snake_case						)
							self.assertIs(obja[0]			,  obja[1]						)
							self.assertIsNot(obja[0]			,  snake_case						)
							self.assertIsNot(obja[0]			,  obja[1]						)
							_lowercase					:  Optional[int]    			=      Hasher.hash(snake_case						)
							_lowercase					:  List[str]    			=      Hasher.hash(snake_case						)
							_lowercase					:  Optional[int]    			=      Hasher.hash(snake_case						)
							self.assertEqual(snake_case			,  snake_case						)
							self.assertEqual(snake_case			,  snake_case						)






	@require_tiktoken
	def 			_a(self							:   Any						)   -> Dict:
							import tiktoken

							_lowercase					:  Dict    			=      tiktoken.get_encoding("gpt2"						)
							_lowercase					:  List[str]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  Optional[int]    			=      tiktoken.get_encoding("r50k_base"						)
							_lowercase					:  Union[str, Any]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  Optional[int]    			=      tiktoken.get_encoding("gpt2"						)
							_lowercase					:  Union[str, Any]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)






	@require_torch
	def 			_a(self							:   Optional[Any]						)   -> List[Any]:
							import torch

							_lowercase					:  Optional[Any]    			=      torch.tensor([1.0]						)
							_lowercase					:  str    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  Optional[Any]    			=      torch.tensor([2.0]						)
							_lowercase					:  List[str]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  List[str]    			=      torch.tensor([1.0]						)
							_lowercase					:  Union[str, Any]    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)






	@require_spacy
	@require_spacy_model("en_core_web_sm"						)
	@require_spacy_model("fr_core_news_sm"						)
	@pytest.mark.integration
	def 			_a(self							:   int						)   -> Optional[Any]:
							import spacy

							_lowercase					:  str    			=      spacy.load("en_core_web_sm"						)
							_lowercase					:  Tuple    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  Dict    			=      spacy.load("fr_core_news_sm"						)
							_lowercase					:  int    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							_lowercase					:  List[Any]    			=      spacy.load("en_core_web_sm"						)
							_lowercase					:  int    			=      mda(datasets.utils.py_utils.dumps(snake_case						)						).hexdigest()
							self.assertEqual(snake_case			,  snake_case						)
							self.assertNotEqual(snake_case			,  snake_case						)



@pytest.mark.integration
def 		UpperCamelCase    (					_UpperCAmelCase     :      Path     )      -> str:






						'''simple docstring'''

						_lowercase					:  str    			=      tmp_path / "dir1"
						_lowercase					:  Union[str, Any]    			=      tmp_path / "dir2"
						dira.mkdir()
						dira.mkdir()
						_lowercase					:  Union[str, Any]    			=      "script.py"
						_lowercase					:  Optional[Any]    			=      dedent(
						    "\n    from datasets.fingerprint import Hasher\n    def foo():\n        pass\n    print(Hasher.hash(foo))\n    "     )
						_lowercase					:  Tuple    			=      dira / script_filename
						_lowercase					:  Dict    			=      dira / script_filename
						with script_patha.open("w"     ) as f:
												f.write(_UpperCAmelCase     )
						with script_patha.open("w"     ) as f:
												f.write(_UpperCAmelCase     )
						_lowercase					:  Optional[int]    			=      subprocess.check_output(["python", str(_UpperCAmelCase     )]     )
						_lowercase					:  Tuple    			=      subprocess.check_output(["python", str(_UpperCAmelCase     )]     )
						assert fingerprinta == fingerprinta



def 		UpperCamelCase    (					)      -> Optional[Any]:






						'''simple docstring'''

						_lowercase					:  str    			=      {"a": [0, 1, 2]}
						_lowercase					:  Any    			=      DatasetChild(InMemoryTable.from_pydict(_UpperCAmelCase     )     )
						_lowercase					:  List[str]    			=      dataset.funca()._fingerprint
						assert expected_fingerprint == dataset.funca()._fingerprint
						assert expected_fingerprint != dataset.funca()._fingerprint

						with Pool(2     ) as p:
												assert expected_fingerprint == p.apply_async(dataset.funca     ).get()._fingerprint
												assert expected_fingerprint != p.apply_async(dataset.funca     ).get()._fingerprint



def 		UpperCamelCase    (					)      -> int:






						'''simple docstring'''

						_lowercase					:  str    			=      {"a": [0, 1, 2]}




						class      __lowercase	(	datasets.Dataset					):






							@fingerprint_transform(inplace=snake_case						)
							def 			_a(self							:   List[str]			,  snake_case							:   Dict						)   -> Optional[Any]:
													return DummyDatasetChild(self.data			,  fingerprint=snake_case						)


						_lowercase					:  Dict    			=      DummyDatasetChild(InMemoryTable.from_pydict(_UpperCAmelCase     )     ).func()




						class      __lowercase	(	datasets.Dataset					):






							@fingerprint_transform(inplace=snake_case			,  version="1.0.0"						)
							def 			_a(self							:   Dict			,  snake_case							:   Optional[int]						)   -> List[Any]:
													return DummyDatasetChild(self.data			,  fingerprint=snake_case						)


						_lowercase					:  Optional[Any]    			=      DummyDatasetChild(InMemoryTable.from_pydict(_UpperCAmelCase     )     ).func()




						class      __lowercase	(	datasets.Dataset					):






							@fingerprint_transform(inplace=snake_case			,  version="2.0.0"						)
							def 			_a(self							:   str			,  snake_case							:   Dict						)   -> Optional[int]:
													return DummyDatasetChild(self.data			,  fingerprint=snake_case						)


						_lowercase					:  List[Any]    			=      DummyDatasetChild(InMemoryTable.from_pydict(_UpperCAmelCase     )     ).func()

						assert len({fingeprint_no_version, fingeprint_a, fingeprint_a}     ) == 3



def 		UpperCamelCase    (					)      -> List[Any]:






						'''simple docstring'''

						_lowercase					:  Dict    			=      Hasher()
						hasher.update(lambda _UpperCAmelCase     : x     )






import gc
import os
import tempfile
import unittest

import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, StableDiffusionPipeline, UNetaDConditionModel
from diffusers.loaders import AttnProcsLayers, LoraLoaderMixin
from diffusers.models.attention_processor import (
    Attention,
    AttnProcessor,
    AttnProcessor2_0,
    LoRAAttnProcessor,
    LoRAAttnProcessor2_0,
    LoRAXFormersAttnProcessor,
    XFormersAttnProcessor,
)
from diffusers.utils import TEXT_ENCODER_ATTN_MODULE, floats_tensor, torch_device



def 		UpperCamelCase    (					_UpperCAmelCase     :      nn.Module     )      -> Any:






						'''simple docstring'''

						_lowercase					:  Optional[int]    			=      {}
						for name in unet.attn_processors.keys():
												_lowercase					:  Any    			=      None if name.endswith("attn1.processor"     ) else unet.config.cross_attention_dim
												if name.startswith("mid_block"     ):
																		_lowercase					:  Any    			=      unet.config.block_out_channels[-1]
												elif name.startswith("up_blocks"     ):
																		_lowercase					:  List[Any]    			=      int(name[len("up_blocks."     )]     )
																		_lowercase					:  Any    			=      list(reversed(unet.config.block_out_channels     )     )[block_id]
												elif name.startswith("down_blocks"     ):
																		_lowercase					:  Union[str, Any]    			=      int(name[len("down_blocks."     )]     )
																		_lowercase					:  Dict    			=      unet.config.block_out_channels[block_id]
												_lowercase					:  List[str]    			=      (
												    LoRAAttnProcessor2_0 if hasattr(_UpperCAmelCase   , "scaled_dot_product_attention"     ) else LoRAAttnProcessor
												)
												_lowercase					:  Optional[Any]    			=      lora_attn_processor_class(
												    hidden_size=_UpperCAmelCase   , cross_attention_dim=_UpperCAmelCase     )
						_lowercase					:  Dict    			=      AttnProcsLayers(_UpperCAmelCase     )
						return lora_attn_procs, unet_lora_layers



def 		UpperCamelCase    (					_UpperCAmelCase     :      nn.Module     )      -> Any:






						'''simple docstring'''

						_lowercase					:  List[str]    			=      {}
						_lowercase					:  Dict    			=      (
						    LoRAAttnProcessor2_0 if hasattr(_UpperCAmelCase   , "scaled_dot_product_attention"     ) else LoRAAttnProcessor
						)
						for name, module in text_encoder.named_modules():
												if name.endswith(_UpperCAmelCase     ):
																		_lowercase					:  Tuple    			=      lora_attn_processor_class(
																		    hidden_size=module.out_proj.out_features   , cross_attention_dim=_UpperCAmelCase     )
						return text_lora_attn_procs



def 		UpperCamelCase    (					_UpperCAmelCase     :      nn.Module     )      -> Union[str, Any]:






						'''simple docstring'''

						_lowercase					:  Union[str, Any]    			=      create_text_encoder_lora_attn_procs(_UpperCAmelCase     )
						_lowercase					:  str    			=      AttnProcsLayers(_UpperCAmelCase     )
						return text_encoder_lora_layers



def 		UpperCamelCase    (					_UpperCAmelCase     :      str   , _UpperCAmelCase     :      int=False     )      -> List[Any]:






						'''simple docstring'''

						for _, attn_proc in text_lora_attn_procs.items():
												# set up.weights
												for layer_name, layer_module in attn_proc.named_modules():
																		if layer_name.endswith("_lora"     ):
																								_lowercase					:  Tuple    			=      (
																								    torch.randn_like(layer_module.up.weight     )
																								    if randn_weight
																								    else torch.zeros_like(layer_module.up.weight     )
																								)
																								_lowercase					:  Dict    			=      torch.nn.Parameter(_UpperCAmelCase     )




class      __lowercase	(	unittest.TestCase					):






	def 			_a(self							:   str						)   -> Tuple:
							torch.manual_seed(0						)
							_lowercase					:  int    			=      UNetaDConditionModel(
							    block_out_channels=(32, 64)			,  layers_per_block=2			,  sample_size=32			,  in_channels=4			,  out_channels=4			,  down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")			,  up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")			,  cross_attention_dim=32			,  )
							_lowercase					:  Union[str, Any]    			=      DDIMScheduler(
							    beta_start=0.0_00_85			,  beta_end=0.0_12			,  beta_schedule="scaled_linear"			,  clip_sample=snake_case			,  set_alpha_to_one=snake_case			,  steps_offset=1			,  )
							torch.manual_seed(0						)
							_lowercase					:  int    			=      AutoencoderKL(
							    block_out_channels=[32, 64]			,  in_channels=3			,  out_channels=3			,  down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]			,  up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]			,  latent_channels=4			,  )
							_lowercase					:  Optional[int]    			=      CLIPTextConfig(
							    bos_token_id=0			,  eos_token_id=2			,  hidden_size=32			,  intermediate_size=37			,  layer_norm_eps=1e-05			,  num_attention_heads=4			,  num_hidden_layers=5			,  pad_token_id=1			,  vocab_size=1000			,  )
							_lowercase					:  List[Any]    			=      CLIPTextModel(snake_case						)
							_lowercase					:  Dict    			=      CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"						)

							_lowercase			,						_lowercase					:  List[Any]    			=      create_unet_lora_layers(snake_case						)
							_lowercase					:  Union[str, Any]    			=      create_text_encoder_lora_layers(snake_case						)

							_lowercase					:  List[Any]    			=      {
							    "unet": unet,
							    "scheduler": scheduler,
							    "vae": vae,
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "safety_checker": None,
							    "feature_extractor": None,
							}
							_lowercase					:  Union[str, Any]    			=      {
							    "unet_lora_layers": unet_lora_layers,
							    "text_encoder_lora_layers": text_encoder_lora_layers,
							    "unet_lora_attn_procs": unet_lora_attn_procs,
							}
							return pipeline_components, lora_components






	def 			_a(self							:   List[Any]						)   -> Optional[Any]:
							_lowercase					:  str    			=      1
							_lowercase					:  str    			=      10
							_lowercase					:  Any    			=      4
							_lowercase					:  List[str]    			=      (32, 32)

							_lowercase					:  Union[str, Any]    			=      torch.manual_seed(0						)
							_lowercase					:  List[str]    			=      floats_tensor((batch_size, num_channels) + sizes						)
							_lowercase					:  Dict    			=      torch.randint(1			,  snake_case			,  size=(batch_size, sequence_length)			,  generator=snake_case						)

							_lowercase					:  Tuple    			=      {
							    "prompt": "A painting of a squirrel eating a burger",
							    "generator": generator,
							    "num_inference_steps": 2,
							    "guidance_scale": 6.0,
							    "output_type": "numpy",
							}

							return noise, input_ids, pipeline_inputs

							# copied from: https://colab.research.google.com/gist/sayakpaul/df2ef6e1ae6d8c10a49d859883b10860/scratchpad.ipynb






	def 			_a(self							:   Union[str, Any]						)   -> int:
							_lowercase					:  Any    			=      77

							_lowercase					:  Any    			=      torch.randint(2			,  56			,  size=(1, max_seq_length)			,  generator=torch.manual_seed(0						)						)

							_lowercase					:  int    			=      {}
							_lowercase					:  int    			=      inputs
							return prepared_inputs






	def 			_a(self							:   Union[str, Any]			,  snake_case							:   Dict						)   -> List[Any]:
							_lowercase			,						_lowercase					:  List[Any]    			=      self.get_dummy_components()
							LoraLoaderMixin.save_lora_weights(
							    save_directory=snake_case			,  unet_lora_layers=lora_components["unet_lora_layers"]			,  text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]			,  )
							self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.bin"						)						)						)






	def 			_a(self							:   Any						)   -> Union[str, Any]:
							_lowercase			,						_lowercase					:  int    			=      self.get_dummy_components()
							_lowercase					:  List[str]    			=      StableDiffusionPipeline(**snake_case						)
							_lowercase					:  Dict    			=      sd_pipe.to(snake_case						)
							sd_pipe.set_progress_bar_config(disable=snake_case						)

							_lowercase			,						_lowercase			,						_lowercase					:  Optional[int]    			=      self.get_dummy_inputs()

							_lowercase					:  Any    			=      sd_pipe(**snake_case						).images
							_lowercase					:  int    			=      original_images[0, -3:, -3:, -1]

							with tempfile.TemporaryDirectory() as tmpdirname:
													LoraLoaderMixin.save_lora_weights(
													    save_directory=snake_case			,  unet_lora_layers=lora_components["unet_lora_layers"]			,  text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]			,  )
													self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.bin"						)						)						)
													sd_pipe.load_lora_weights(snake_case						)

							_lowercase					:  Dict    			=      sd_pipe(**snake_case						).images
							_lowercase					:  List[Any]    			=      lora_images[0, -3:, -3:, -1]

							# Outputs shouldn't match.
							self.assertFalse(torch.allclose(torch.from_numpy(snake_case						)			,  torch.from_numpy(snake_case						)						)						)






	def 			_a(self							:   Optional[int]						)   -> List[Any]:
							_lowercase			,						_lowercase					:  str    			=      self.get_dummy_components()
							_lowercase					:  Any    			=      StableDiffusionPipeline(**snake_case						)
							_lowercase					:  Any    			=      sd_pipe.to(snake_case						)
							sd_pipe.set_progress_bar_config(disable=snake_case						)

							_lowercase			,						_lowercase			,						_lowercase					:  Optional[int]    			=      self.get_dummy_inputs()

							_lowercase					:  Optional[int]    			=      sd_pipe(**snake_case						).images
							_lowercase					:  Any    			=      original_images[0, -3:, -3:, -1]

							with tempfile.TemporaryDirectory() as tmpdirname:
													LoraLoaderMixin.save_lora_weights(
													    save_directory=snake_case			,  unet_lora_layers=lora_components["unet_lora_layers"]			,  text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]			,  safe_serialization=snake_case			,  )
													self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.safetensors"						)						)						)
													sd_pipe.load_lora_weights(snake_case						)

							_lowercase					:  Any    			=      sd_pipe(**snake_case						).images
							_lowercase					:  Optional[int]    			=      lora_images[0, -3:, -3:, -1]

							# Outputs shouldn't match.
							self.assertFalse(torch.allclose(torch.from_numpy(snake_case						)			,  torch.from_numpy(snake_case						)						)						)






	def 			_a(self							:   str						)   -> Union[str, Any]:
							_lowercase			,						_lowercase					:  Optional[Any]    			=      self.get_dummy_components()
							_lowercase					:  int    			=      lora_components["unet_lora_attn_procs"]
							_lowercase					:  str    			=      StableDiffusionPipeline(**snake_case						)
							_lowercase					:  List[str]    			=      sd_pipe.to(snake_case						)
							sd_pipe.set_progress_bar_config(disable=snake_case						)

							_lowercase			,						_lowercase			,						_lowercase					:  Any    			=      self.get_dummy_inputs()

							_lowercase					:  int    			=      sd_pipe(**snake_case						).images
							_lowercase					:  List[str]    			=      original_images[0, -3:, -3:, -1]

							with tempfile.TemporaryDirectory() as tmpdirname:
													_lowercase					:  Optional[Any]    			=      sd_pipe.unet
													unet.set_attn_processor(snake_case						)
													unet.save_attn_procs(snake_case						)
													self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.bin"						)						)						)
													sd_pipe.load_lora_weights(snake_case						)

							_lowercase					:  int    			=      sd_pipe(**snake_case						).images
							_lowercase					:  int    			=      lora_images[0, -3:, -3:, -1]

							# Outputs shouldn't match.
							self.assertFalse(torch.allclose(torch.from_numpy(snake_case						)			,  torch.from_numpy(snake_case						)						)						)






	def 			_a(self							:   Dict						)   -> List[str]:
							_lowercase			,						_lowercase					:  Optional[Any]    			=      self.get_dummy_components()
							_lowercase					:  List[Any]    			=      StableDiffusionPipeline(**snake_case						)

							_lowercase					:  Tuple    			=      self.get_dummy_tokens()

							# inference without lora
							_lowercase					:  int    			=      pipe.text_encoder(**snake_case						)[0]
							assert outputs_without_lora.shape == (1, 77, 32)

							# create lora_attn_procs with zeroed out up.weights
							_lowercase					:  str    			=      create_text_encoder_lora_attn_procs(pipe.text_encoder						)
							set_lora_up_weights(snake_case			,  randn_weight=snake_case						)

							# monkey patch
							pipe._modify_text_encoder(snake_case						)

							# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
							del text_attn_procs
							gc.collect()

							# inference with lora
							_lowercase					:  Dict    			=      pipe.text_encoder(**snake_case						)[0]
							assert outputs_with_lora.shape == (1, 77, 32)

							assert torch.allclose(
							    snake_case			,  snake_case						), "lora_up_weight are all zero, so the lora outputs should be the same to without lora outputs"

							# create lora_attn_procs with randn up.weights
							_lowercase					:  Any    			=      create_text_encoder_lora_attn_procs(pipe.text_encoder						)
							set_lora_up_weights(snake_case			,  randn_weight=snake_case						)

							# monkey patch
							pipe._modify_text_encoder(snake_case						)

							# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
							del text_attn_procs
							gc.collect()

							# inference with lora
							_lowercase					:  Any    			=      pipe.text_encoder(**snake_case						)[0]
							assert outputs_with_lora.shape == (1, 77, 32)

							assert not torch.allclose(
							    snake_case			,  snake_case						), "lora_up_weight are not zero, so the lora outputs should be different to without lora outputs"






	def 			_a(self							:   str						)   -> Tuple:
							_lowercase			,						_lowercase					:  List[Any]    			=      self.get_dummy_components()
							_lowercase					:  str    			=      StableDiffusionPipeline(**snake_case						)

							_lowercase					:  Dict    			=      self.get_dummy_tokens()

							# inference without lora
							_lowercase					:  Tuple    			=      pipe.text_encoder(**snake_case						)[0]
							assert outputs_without_lora.shape == (1, 77, 32)

							# create lora_attn_procs with randn up.weights
							_lowercase					:  Optional[int]    			=      create_text_encoder_lora_attn_procs(pipe.text_encoder						)
							set_lora_up_weights(snake_case			,  randn_weight=snake_case						)

							# monkey patch
							pipe._modify_text_encoder(snake_case						)

							# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
							del text_attn_procs
							gc.collect()

							# inference with lora
							_lowercase					:  Union[str, Any]    			=      pipe.text_encoder(**snake_case						)[0]
							assert outputs_with_lora.shape == (1, 77, 32)

							assert not torch.allclose(
							    snake_case			,  snake_case						), "lora outputs should be different to without lora outputs"

							# remove monkey patch
							pipe._remove_text_encoder_monkey_patch()

							# inference with removed lora
							_lowercase					:  Union[str, Any]    			=      pipe.text_encoder(**snake_case						)[0]
							assert outputs_without_lora_removed.shape == (1, 77, 32)

							assert torch.allclose(
							    snake_case			,  snake_case						), "remove lora monkey patch should restore the original outputs"






	def 			_a(self							:   Optional[int]						)   -> Dict:
							_lowercase			,						_lowercase					:  Dict    			=      self.get_dummy_components()
							_lowercase					:  Optional[Any]    			=      StableDiffusionPipeline(**snake_case						)
							_lowercase					:  Union[str, Any]    			=      sd_pipe.to(snake_case						)
							sd_pipe.set_progress_bar_config(disable=snake_case						)

							_lowercase			,						_lowercase			,						_lowercase					:  str    			=      self.get_dummy_inputs()

							with tempfile.TemporaryDirectory() as tmpdirname:
													LoraLoaderMixin.save_lora_weights(
													    save_directory=snake_case			,  unet_lora_layers=lora_components["unet_lora_layers"]			,  text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]			,  )
													self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.bin"						)						)						)
													sd_pipe.load_lora_weights(snake_case						)

							_lowercase					:  Dict    			=      sd_pipe(**snake_case						).images
							_lowercase					:  int    			=      lora_images[0, -3:, -3:, -1]

							_lowercase					:  List[str]    			=      sd_pipe(**snake_case			,  cross_attention_kwargs={"scale": 0.5}						).images
							_lowercase					:  int    			=      lora_images_with_scale[0, -3:, -3:, -1]

							# Outputs shouldn't match.
							self.assertFalse(
							    torch.allclose(torch.from_numpy(snake_case						)			,  torch.from_numpy(snake_case						)						)						)






	def 			_a(self							:   Tuple						)   -> str:
							with tempfile.TemporaryDirectory() as tmpdirname:
													self.create_lora_weight_file(snake_case						)

													_lowercase			,						_lowercase					:  str    			=      self.get_dummy_components()
													_lowercase					:  str    			=      StableDiffusionPipeline(**snake_case						)
													_lowercase					:  List[Any]    			=      sd_pipe.to(snake_case						)
													sd_pipe.set_progress_bar_config(disable=snake_case						)

													# check if vanilla attention processors are used
													for _, module in sd_pipe.unet.named_modules():
																			if isinstance(snake_case			,  snake_case						):
																									self.assertIsInstance(module.processor			,  (AttnProcessor, AttnProcessor2_0)						)

            # load LoRA weight file
													sd_pipe.load_lora_weights(snake_case						)

													# check if lora attention processors are used
													for _, module in sd_pipe.unet.named_modules():
																			if isinstance(snake_case			,  snake_case						):
																									_lowercase					:  Tuple    			=      (
																									    LoRAAttnProcessor2_0 if hasattr(snake_case			,  "scaled_dot_product_attention"						) else LoRAAttnProcessor
																									)
																									self.assertIsInstance(module.processor			,  snake_case						)






	@unittest.skipIf(torch_device != "cuda"			,  "This test is supposed to run on GPU"						)
	def 			_a(self							:   List[str]						)   -> Union[str, Any]:
							with tempfile.TemporaryDirectory() as tmpdirname:
													self.create_lora_weight_file(snake_case						)

													_lowercase			,						_lowercase					:  List[str]    			=      self.get_dummy_components()
													_lowercase					:  str    			=      StableDiffusionPipeline(**snake_case						)
													_lowercase					:  Tuple    			=      sd_pipe.to(snake_case						)
													sd_pipe.set_progress_bar_config(disable=snake_case						)

													# enable XFormers
													sd_pipe.enable_xformers_memory_efficient_attention()

													# check if xFormers attention processors are used
													for _, module in sd_pipe.unet.named_modules():
																			if isinstance(snake_case			,  snake_case						):
																									self.assertIsInstance(module.processor			,  snake_case						)

            # load LoRA weight file
													sd_pipe.load_lora_weights(snake_case						)

													# check if lora attention processors are used
													for _, module in sd_pipe.unet.named_modules():
																			if isinstance(snake_case			,  snake_case						):
																									self.assertIsInstance(module.processor			,  snake_case						)






	@unittest.skipIf(torch_device != "cuda"			,  "This test is supposed to run on GPU"						)
	def 			_a(self							:   List[str]						)   -> Any:
							_lowercase			,						_lowercase					:  Tuple    			=      self.get_dummy_components()
							_lowercase					:  List[Any]    			=      StableDiffusionPipeline(**snake_case						)
							_lowercase					:  Union[str, Any]    			=      sd_pipe.to(snake_case						)
							sd_pipe.set_progress_bar_config(disable=snake_case						)

							_lowercase			,						_lowercase			,						_lowercase					:  List[Any]    			=      self.get_dummy_inputs()

							# enable XFormers
							sd_pipe.enable_xformers_memory_efficient_attention()

							_lowercase					:  Tuple    			=      sd_pipe(**snake_case						).images
							_lowercase					:  Any    			=      original_images[0, -3:, -3:, -1]

							with tempfile.TemporaryDirectory() as tmpdirname:
													LoraLoaderMixin.save_lora_weights(
													    save_directory=snake_case			,  unet_lora_layers=lora_components["unet_lora_layers"]			,  text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]			,  )
													self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.bin"						)						)						)
													sd_pipe.load_lora_weights(snake_case						)

							_lowercase					:  Optional[int]    			=      sd_pipe(**snake_case						).images
							_lowercase					:  Tuple    			=      lora_images[0, -3:, -3:, -1]

							# Outputs shouldn't match.
							self.assertFalse(torch.allclose(torch.from_numpy(snake_case						)			,  torch.from_numpy(snake_case						)						)						)




