





import gc
import importlib.metadata
import tempfile
import unittest

from packaging import version

from transformers import (
    AutoModel,
    AutoModelForCausalLM,
    AutoModelForSeqaSeqLM,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    BitsAndBytesConfig,
    pipeline,
)
from transformers.testing_utils import (
    is_torch_available,
    require_accelerate,
    require_bitsandbytes,
    require_torch,
    require_torch_gpu,
    require_torch_multi_gpu,
    slow,
)



def A_	(		A__				)  ->		Any:
       if model.config.model_type == "gpt2":
              return model.transformer.h[0].mlp.c_fc
       return model.transformer.h[0].mlp.dense_ah_to_h


if is_torch_available():
    import torch
    import torch.nn as nn





    class  A__    (							nn.Module	):




       """simple docstring"""



       def __init__(					self					,  lowercase					,  lowercase)			->						Optional[int]:







              '''simple docstring'''

              super().__init__()
              a__						:	int =				module
              a__						:	Dict =				nn.Sequential(
                  nn.Linear(module.in_features					,  lowercase					,  bias=lowercase)					,  nn.Linear(lowercase					,  module.out_features					,  bias=lowercase)					,  )
              a__						:	Dict =				(2.0 / (5 * min(module.in_features					,  module.out_features))) ** 0.5
              nn.init.normal_(self.adapter[0].weight					,  std=lowercase)
              nn.init.zeros_(self.adapter[1].weight)
              self.adapter.to(module.weight.device)

       def 							__lowercase		(					self					,  lowercase					,  *lowercase					,  **lowercase)			->						Any:







              '''simple docstring'''

              return self.module(lowercase					,  *lowercase					,  **lowercase) + self.adapter(lowercase)





@require_bitsandbytes
@require_accelerate
@require_torch
@require_torch_gpu
@slow
class  A__    (							unittest.TestCase	):




   """simple docstring"""



   __A   :				Tuple         =  '''bigscience/bloom-1b7'''

   # Constant values
   __A   :				str         =  2.109_6595_5269_2574

   __A   :				List[Any]         =  '''Hello my name is'''
   __A   :				List[str]         =  set()
   EXPECTED_OUTPUTS.add('''Hello my name is John and I am a professional photographer. I'''	)
   EXPECTED_OUTPUTS.add('''Hello my name is John.\nI am a friend of your father.\n'''	)
   EXPECTED_OUTPUTS.add('''Hello my name is John Doe, I am a student at the University'''	)
   __A   :				List[Any]         =  1_0

   def 							__lowercase		(					self)			->						str:







          '''simple docstring'''

          a__						:	Union[str, Any] =				AutoTokenizer.from_pretrained(self.model_name)





class  A__    (							__UpperCAmelCase	):




   """simple docstring"""



   def 							__lowercase		(					self)			->						int:







          '''simple docstring'''

          super().setUp()

          # Models and tokenizer
          a__						:	str =				AutoModelForCausalLM.from_pretrained(
              self.model_name					,  torch_dtype=torch.floataa					,  device_map='auto')
          a__						:	Union[str, Any] =				AutoModelForCausalLM.from_pretrained(self.model_name					,  load_in_abit=lowercase					,  device_map='auto')

   def 							__lowercase		(					self)			->						Any:







          '''simple docstring'''

          del self.model_fpaa
          del self.model_abit

          gc.collect()
          torch.cuda.empty_cache()

   def 							__lowercase		(					self)			->						Optional[int]:







          '''simple docstring'''

          a__						:	Dict =				self.model_abit.config

          self.assertTrue(hasattr(lowercase					,  'quantization_config'))

          a__						:	str =				config.to_dict()
          a__						:	Dict =				config.to_diff_dict()

          a__						:	Dict =				config.to_json_string()

   def 							__lowercase		(					self)			->						Optional[Any]:







          '''simple docstring'''

          from bitsandbytes.nn import Paramsabit

          a__						:	Optional[Any] =				self.model_fpaa.get_memory_footprint()
          a__						:	Any =				self.model_abit.get_memory_footprint()

          self.assertAlmostEqual(mem_fpaa / mem_abit					,  self.EXPECTED_RELATIVE_DIFFERENCE)
          a__						:	Optional[int] =				get_some_linear_layer(self.model_abit)
          self.assertTrue(linear.weight.__class__ == Paramsabit)

   def 							__lowercase		(					self)			->						str:







          '''simple docstring'''

          from transformers import TaPreTrainedModel

          self.model_fpaa.get_memory_footprint()
          self.model_abit.get_memory_footprint()

          for name, module in self.model_abit.named_modules():
                 if isinstance(lowercase					,  torch.nn.Linear):
                        if name not in ["lm_head"] + TaPreTrainedModel._keep_in_fpaa_modules:
                               # 4-bit parameters are packed in uint8 variables
                               self.assertTrue(module.weight.dtype == torch.uinta)

   def 							__lowercase		(					self)			->						Optional[int]:







          '''simple docstring'''

          a__						:	List[str] =				self.tokenizer(self.input_text					,  return_tensors='pt')
          a__						:	List[str] =				self.model_abit.generate(input_ids=encoded_input['input_ids'].to(0)					,  max_new_tokens=10)

          self.assertIn(self.tokenizer.decode(output_sequences[0]					,  skip_special_tokens=lowercase)					,  self.EXPECTED_OUTPUTS)

   def 							__lowercase		(					self)			->						int:







          '''simple docstring'''

          a__						:	Dict =				BitsAndBytesConfig()
          a__						:	Union[str, Any] =				True

          a__						:	Optional[Any] =				AutoModelForCausalLM.from_pretrained(
              self.model_name					,  quantization_config=lowercase					,  device_map='auto')

          a__						:	Any =				self.tokenizer(self.input_text					,  return_tensors='pt')
          a__						:	List[str] =				model_abit_from_config.generate(
              input_ids=encoded_input['input_ids'].to(0)					,  max_new_tokens=10)

          self.assertIn(self.tokenizer.decode(output_sequences[0]					,  skip_special_tokens=lowercase)					,  self.EXPECTED_OUTPUTS)

   def 							__lowercase		(					self)			->						Any:







          '''simple docstring'''

          with self.assertRaises(lowercase), tempfile.TemporaryDirectory() as tmpdirname:
                 self.model_abit.save_pretrained(lowercase)

   def 							__lowercase		(					self)			->						str:







          '''simple docstring'''

          a__						:	Any =				BitsAndBytesConfig()

          with self.assertRaises(lowercase):
                 a__						:	Any =				AutoModelForCausalLM.from_pretrained(
                     self.model_name					,  quantization_config=lowercase					,  load_in_abit=lowercase					,  device_map='auto'					,  bnb_abit_quant_type='nf4'					,  )

   def 							__lowercase		(					self)			->						Optional[Any]:







          '''simple docstring'''

          with self.assertRaises(lowercase):
                 # Tries with `str`
                 self.model_abit.to('cpu')

          with self.assertRaises(lowercase):
                 # Tries with a `dtype``
                 self.model_abit.to(torch.floataa)

          with self.assertRaises(lowercase):
                 # Tries with a `device`
                 self.model_abit.to(torch.device('cuda:0'))

          with self.assertRaises(lowercase):
                 # Tries with a `device`
                 self.model_abit.float()

          with self.assertRaises(lowercase):
                 # Tries with a `device`
                 self.model_abit.half()

          # Test if we did not break anything
          a__						:	Optional[int] =				self.tokenizer(self.input_text					,  return_tensors='pt')

          a__						:	Optional[Any] =				self.model_fpaa.to(torch.floataa)
          a__						:	int =				self.model_fpaa.generate(input_ids=encoded_input['input_ids'].to(0)					,  max_new_tokens=10)

          # Check this does not throw an error
          a__						:	Tuple =				self.model_fpaa.to('cpu')

          # Check this does not throw an error
          a__						:	List[str] =				self.model_fpaa.half()

          # Check this does not throw an error
          a__						:	Any =				self.model_fpaa.float()

   def 							__lowercase		(					self)			->						Optional[Any]:







          '''simple docstring'''

          a__						:	List[Any] =				AutoModelForSeqaSeqLM.from_pretrained('t5-small'					,  load_in_abit=lowercase					,  device_map='auto')
          self.assertTrue(model.decoder.block[0].layer[2].DenseReluDense.wo.weight.dtype == torch.floataa)





@require_bitsandbytes
@require_accelerate
@require_torch
@require_torch_gpu
@slow
class  A__    (							unittest.TestCase	):




   """simple docstring"""



   @classmethod
   def 							__lowercase		(					cls)			->						List[str]:







          '''simple docstring'''

          a__						:	Union[str, Any] =				't5-small'
          a__						:	str =				'google/flan-t5-small'  # flan-t5 uses dense-act instead of dense-relu-dense
          a__						:	Dict =				AutoTokenizer.from_pretrained(cls.model_name)
          a__						:	List[str] =				'Translate in German: Hello, my dog is cute'

   def 							__lowercase		(					self)			->						List[Any]:







          '''simple docstring'''

          gc.collect()
          torch.cuda.empty_cache()

   def 							__lowercase		(					self)			->						List[Any]:







          '''simple docstring'''

          from transformers import TaForConditionalGeneration

          a__						:	Tuple =				TaForConditionalGeneration._keep_in_fpaa_modules
          a__						:	Any =				None

          # test with `t5-small`
          a__						:	Tuple =				TaForConditionalGeneration.from_pretrained(self.model_name					,  load_in_abit=lowercase					,  device_map='auto')
          a__						:	Any =				self.tokenizer(self.input_text					,  return_tensors='pt').to(0)
          a__						:	Optional[Any] =				model.generate(**lowercase)

          # test with `flan-t5-small`
          a__						:	Dict =				TaForConditionalGeneration.from_pretrained(
              self.dense_act_model_name					,  load_in_abit=lowercase					,  device_map='auto')
          a__						:	Optional[Any] =				self.tokenizer(self.input_text					,  return_tensors='pt').to(0)
          a__						:	Union[str, Any] =				model.generate(**lowercase)
          a__						:	Dict =				modules

   def 							__lowercase		(					self)			->						List[str]:







          '''simple docstring'''

          import bitsandbytes as bnb

          from transformers import TaForConditionalGeneration

          # test with `t5-small`
          a__						:	Tuple =				TaForConditionalGeneration.from_pretrained(self.model_name					,  load_in_abit=lowercase					,  device_map='auto')

          # there was a bug with decoders - this test checks that it is fixed
          self.assertTrue(isinstance(model.decoder.block[0].layer[0].SelfAttention.q					,  bnb.nn.Linearabit))

          a__						:	Optional[int] =				self.tokenizer(self.input_text					,  return_tensors='pt').to(0)
          a__						:	Union[str, Any] =				model.generate(**lowercase)

          # test with `flan-t5-small`
          a__						:	int =				TaForConditionalGeneration.from_pretrained(
              self.dense_act_model_name					,  load_in_abit=lowercase					,  device_map='auto')
          a__						:	str =				self.tokenizer(self.input_text					,  return_tensors='pt').to(0)
          a__						:	Any =				model.generate(**lowercase)





class  A__    (							__UpperCAmelCase	):




   """simple docstring"""



   def 							__lowercase		(					self)			->						Tuple:







          '''simple docstring'''

          super().setUp()
          # model_name
          a__						:	List[Any] =				'bigscience/bloom-560m'
          a__						:	str =				't5-small'

          # Different types of model

          a__						:	Optional[int] =				AutoModel.from_pretrained(self.model_name					,  load_in_abit=lowercase					,  device_map='auto')
          # Sequence classification model
          a__						:	Optional[int] =				AutoModelForSequenceClassification.from_pretrained(
              self.model_name					,  load_in_abit=lowercase					,  device_map='auto')
          # CausalLM model
          a__						:	List[Any] =				AutoModelForCausalLM.from_pretrained(self.model_name					,  load_in_abit=lowercase					,  device_map='auto')
          # Seq2seq model
          a__						:	List[Any] =				AutoModelForSeqaSeqLM.from_pretrained(
              self.seq_to_seq_name					,  load_in_abit=lowercase					,  device_map='auto')

   def 							__lowercase		(					self)			->						List[Any]:







          '''simple docstring'''

          del self.base_model
          del self.sequence_model
          del self.model_abit
          del self.seq_to_seq_model

          gc.collect()
          torch.cuda.empty_cache()

   def 							__lowercase		(					self)			->						List[str]:







          '''simple docstring'''

          from bitsandbytes.nn import Paramsabit

          self.assertTrue(self.base_model.h[-1].mlp.dense_ah_to_h.weight.__class__ == Paramsabit)

          # Other heads should be nn.Parameter
          self.assertTrue(self.model_abit.lm_head.weight.__class__ == torch.nn.Parameter)
          self.assertTrue(self.sequence_model.score.weight.__class__ == torch.nn.Parameter)
          self.assertTrue(self.seq_to_seq_model.lm_head.weight.__class__ == torch.nn.Parameter)





class  A__    (							__UpperCAmelCase	):




   """simple docstring"""



   def 							__lowercase		(					self)			->						Union[str, Any]:







          '''simple docstring'''

          super().setUp()

   def 							__lowercase		(					self)			->						Optional[int]:







          '''simple docstring'''

          del self.pipe

          gc.collect()
          torch.cuda.empty_cache()

   def 							__lowercase		(					self)			->						str:







          '''simple docstring'''

          a__						:	Optional[Any] =				pipeline(
              'text-generation'					,  model=self.model_name					,  model_kwargs={'device_map': 'auto', 'load_in_4bit': True, 'torch_dtype': torch.floataa}					,  max_new_tokens=self.MAX_NEW_TOKENS					,  )

          # Real second forward pass
          a__						:	int =				self.pipe(self.input_text)
          self.assertIn(pipeline_output[0]['generated_text']					,  self.EXPECTED_OUTPUTS)





@require_torch_multi_gpu
class  A__    (							__UpperCAmelCase	):




   """simple docstring"""



   def 							__lowercase		(					self)			->						Union[str, Any]:







          '''simple docstring'''

          super().setUp()

   def 							__lowercase		(					self)			->						int:







          '''simple docstring'''

          a__						:	Optional[Any] =				AutoModelForCausalLM.from_pretrained(
              self.model_name					,  load_in_abit=lowercase					,  device_map='balanced')

          # Check correct device map
          self.assertEqual(set(model_parallel.hf_device_map.values())					,  {0, 1})

          # Check that inference pass works on the model
          a__						:	Tuple =				self.tokenizer(self.input_text					,  return_tensors='pt')

          # Second real batch
          a__						:	Optional[Any] =				model_parallel.generate(input_ids=encoded_input['input_ids'].to(0)					,  max_new_tokens=10)
          self.assertIn(self.tokenizer.decode(output_parallel[0]					,  skip_special_tokens=lowercase)					,  self.EXPECTED_OUTPUTS)





class  A__    (							__UpperCAmelCase	):




   """simple docstring"""



   def 							__lowercase		(					self)			->						Optional[Any]:







          '''simple docstring'''

          a__						:	Optional[int] =				'facebook/opt-350m'
          super().setUp()

   def 							__lowercase		(					self)			->						List[Any]:







          '''simple docstring'''

          if version.parse(importlib.metadata.version('bitsandbytes')) < version.parse('0.37.0'):
                 return

          # Step 1: freeze all parameters
          a__						:	Optional[int] =				AutoModelForCausalLM.from_pretrained(self.model_name					,  load_in_abit=lowercase)

          self.assertEqual(set(model.hf_device_map.values())					,  {torch.cuda.current_device()})

          for param in model.parameters():
                 a__						:	Any =				False  # freeze the model - train adapters later
                 if param.ndim == 1:
                        # cast the small parameters (e.g. layernorm) to fp32 for stability
                        a__						:	List[Any] =				param.data.to(torch.floataa)

        # Step 2: add adapters
          for _, module in model.named_modules():
                 if "OPTAttention" in repr(type(lowercase)):
                        a__						:	Optional[int] =				LoRALayer(module.q_proj					,  rank=16)
                        a__						:	Optional[Any] =				LoRALayer(module.k_proj					,  rank=16)
                        a__						:	Optional[Any] =				LoRALayer(module.v_proj					,  rank=16)

        # Step 3: dummy batch
          a__						:	Optional[int] =				self.tokenizer('Test batch '					,  return_tensors='pt').to(0)

          # Step 4: Check if the gradient is not None
          with torch.cuda.amp.autocast():
                 a__						:	int =				model.forward(**lowercase)
                 out.logits.norm().backward()

          for module in model.modules():
                 if isinstance(lowercase					,  lowercase):
                        self.assertTrue(module.adapter[1].weight.grad is not None)
                        self.assertTrue(module.adapter[1].weight.grad.norm().item() > 0)
                 elif isinstance(lowercase					,  nn.Embedding):
                        self.assertTrue(module.weight.grad is None)





class  A__    (							__UpperCAmelCase	):




   """simple docstring"""



   __A   :				Tuple         =  '''gpt2-xl'''
   __A   :				Dict         =  3.3191_8548_5415_2187







'''simple docstring'''






import argparse
import os
import re

import packaging.version


UpperCamelCase__:   Union[str, Any]								=			"examples/"
UpperCamelCase__:   Optional[Any]								=			{
    "examples": (re.compile(r"^check_min_version\(\"[^\"]+\"\)\s*$", re.MULTILINE), "check_min_version(\"VERSION\")\n"),
    "init": (re.compile(r"^__version__\s+=\s+\"([^\"]+)\"\s*$", re.MULTILINE), "__version__ = \"VERSION\"\n"),
    "setup": (re.compile(r"^(\s*)version\s*=\s*\"[^\"]+\",", re.MULTILINE), r"\1version=\"VERSION\","),
    "doc": (re.compile(r"^(\s*)release\s*=\s*\"[^\"]+\"$", re.MULTILINE), "release = \"VERSION\"\n"),
}
UpperCamelCase__:   Optional[int]								=			{
    "init": "src/diffusers/__init__.py",
    "setup": "setup.py",
}
UpperCamelCase__:   List[Any]								=			"README.md"


def  snake_case_     (					_lowerCAmelCase  :						Optional[Any]   ,      _lowerCAmelCase  :						List[Any]   ,      _lowerCAmelCase  :						int  )      ->   Optional[int]:
							with open(_lowerCAmelCase   ,      '''r'''   ,      encoding='''utf-8'''   ,      newline='''\n'''  ) as f:
														UpperCAmelCase :       Optional[int] 							=		f.read()
							UpperCAmelCase							,		UpperCAmelCase :       List[Any] 							=		REPLACE_PATTERNS[pattern]
							UpperCAmelCase :       List[Any] 							=		replace.replace('''VERSION'''   ,      _lowerCAmelCase  )
							UpperCAmelCase :       Optional[Any] 							=		re_pattern.sub(_lowerCAmelCase   ,      _lowerCAmelCase  )
							with open(_lowerCAmelCase   ,      '''w'''   ,      encoding='''utf-8'''   ,      newline='''\n'''  ) as f:
														f.write(_lowerCAmelCase  )


def  snake_case_     (					_lowerCAmelCase  :						Any  )      ->   Optional[int]:
							for folder, directories, fnames in os.walk(_lowerCAmelCase  ):
														# Removing some of the folders with non-actively maintained examples from the walk
														if "research_projects" in directories:
																					directories.remove('''research_projects'''  )
														if "legacy" in directories:
																					directories.remove('''legacy'''  )
														for fname in fnames:
																					if fname.endswith('''.py'''  ):
																												update_version_in_file(os.path.join(_lowerCAmelCase   ,      _lowerCAmelCase  )   ,      _lowerCAmelCase   ,      pattern='''examples'''  )


def  snake_case_     (					_lowerCAmelCase  :						Any   ,      _lowerCAmelCase  :						str=False  )      ->   List[str]:
							for pattern, fname in REPLACE_FILES.items():
														update_version_in_file(_lowerCAmelCase   ,      _lowerCAmelCase   ,      _lowerCAmelCase  )
							if not patch:
														update_version_in_examples(_lowerCAmelCase  )


def  snake_case_     (					)      ->   Optional[Any]:
							UpperCAmelCase :       Optional[int] 							=		'''🤗 Transformers currently provides the following architectures'''
							UpperCAmelCase :       Optional[int] 							=		'''1. Want to contribute a new model?'''
							with open(_lowerCAmelCase   ,      '''r'''   ,      encoding='''utf-8'''   ,      newline='''\n'''  ) as f:
														UpperCAmelCase :       Optional[Any] 							=		f.readlines()

							# Find the start of the list.
							UpperCAmelCase :       List[Any] 							=		0
							while not lines[start_index].startswith(_start_prompt  ):
														start_index += 1
							start_index += 1

							UpperCAmelCase :       Optional[Any] 							=		start_index
							# Update the lines in the model list.
							while not lines[index].startswith(_end_prompt  ):
														if lines[index].startswith('''1.'''  ):
																					UpperCAmelCase :       Optional[int] 							=		lines[index].replace(
																					    '''https://huggingface.co/docs/diffusers/main/model_doc'''   ,      '''https://huggingface.co/docs/diffusers/model_doc'''   ,      )
														index += 1

							with open(_lowerCAmelCase   ,      '''w'''   ,      encoding='''utf-8'''   ,      newline='''\n'''  ) as f:
														f.writelines(_lowerCAmelCase  )


def  snake_case_     (					)      ->   Optional[Any]:
							with open(REPLACE_FILES['''init''']   ,      '''r'''  ) as f:
														UpperCAmelCase :       Union[str, Any] 							=		f.read()
							UpperCAmelCase :       int 							=		REPLACE_PATTERNS['''init'''][0].search(_lowerCAmelCase  ).groups()[0]
							return packaging.version.parse(_lowerCAmelCase  )


def  snake_case_     (					_lowerCAmelCase  :						List[str]=False  )      ->   Any:
							UpperCAmelCase :       Optional[Any] 							=		get_version()
							if patch and default_version.is_devrelease:
														raise ValueError('''Can\'t create a patch version from the dev branch, checkout a released version!'''  )
							if default_version.is_devrelease:
														UpperCAmelCase :       Optional[int] 							=		default_version.base_version
							elif patch:
														UpperCAmelCase :       Union[str, Any] 							=		f"""{default_version.major}.{default_version.minor}.{default_version.micro + 1}"""
							else:
														UpperCAmelCase :       Union[str, Any] 							=		f"""{default_version.major}.{default_version.minor + 1}.0"""

							# Now let's ask nicely if that's the right one.
							UpperCAmelCase :       Dict 							=		input(f"""Which version are you releasing? [{default_version}]"""  )
							if len(_lowerCAmelCase  ) == 0:
														UpperCAmelCase :       Tuple 							=		default_version

							print(f"""Updating version to {version}."""  )
							global_version_update(_lowerCAmelCase   ,      patch=_lowerCAmelCase  )


def  snake_case_     (					)      ->   Any:
							UpperCAmelCase :       List[Any] 							=		get_version()
							UpperCAmelCase :       List[str] 							=		f"""{current_version.major}.{current_version.minor + 1}.0.dev0"""
							UpperCAmelCase :       List[Any] 							=		current_version.base_version

							# Check with the user we got that right.
							UpperCAmelCase :       Optional[int] 							=		input(f"""Which version are we developing now? [{dev_version}]"""  )
							if len(_lowerCAmelCase  ) == 0:
														UpperCAmelCase :       Dict 							=		dev_version

							print(f"""Updating version to {version}."""  )
							global_version_update(_lowerCAmelCase  )


#    print("Cleaning main README, don't forget to run `make fix-copies`.")
#    clean_main_ref_in_model_list()


if __name__ == "__main__":
						UpperCamelCase__:   Union[str, Any]								=			argparse.ArgumentParser()
						parser.add_argument("--post_release", action="store_true", help="Whether this is pre or post release.")
						parser.add_argument("--patch", action="store_true", help="Whether or not this is a patch release.")
						UpperCamelCase__:   Optional[Any]								=			parser.parse_args()
						if not args.post_release:
												pre_release_work(patch=args.patch)
						elif args.patch:
												print("Nothing to do after a patch :-)")
						else:
												post_release_work()





