code: 
  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     )
  
  
  
  
  
style_context: 
  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						)						)						)
  
  
  
  
  