




'''simple docstring'''







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        lowerCamelCase__				(      __lowercase				):
     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							,__a							,__a							)      ->				Any:
                super().__init__()
                snake_case			:      Tuple        =			module
                snake_case			:      Optional[int]        =			nn.Sequential(
                    nn.Linear(module.in_features							,__a							,bias=__a							)							,nn.Linear(__a							,module.out_features							,bias=__a							)							,)
                snake_case			:      List[str]        =			(2.0 / (5 * min(module.in_features							,module.out_features							))) ** 0.5
                nn.init.normal_(self.adapter[0].weight							,std=__a							)
                nn.init.zeros_(self.adapter[1].weight							)
                self.adapter.to(module.weight.device							)







           def 			snake_case_      (				self							,__a							,*__a							,**__a							)      ->				Tuple:
                return self.module(__a							,*__a							,**__a							) + self.adapter(__a							)







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





     '''simple docstring'''



     lowerCAmelCase_  :     Optional[int]          =					"""bigscience/bloom-1b7"""

     # Constant values
     lowerCAmelCase_  :     Optional[Any]          =					2.109_6595_5269_2574

     lowerCAmelCase_  :     str          =					"""Hello my name is"""
     lowerCAmelCase_  :     int          =					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"""	)
     lowerCAmelCase_  :     Dict          =					10







     def 			snake_case_      (				self							)      ->				List[str]:
          # Models and tokenizer
          snake_case			:      str        =			AutoTokenizer.from_pretrained(self.model_name							)







class 					_a (a__	):





     '''simple docstring'''



     def 			snake_case_      (				self							)      ->				int:
          super().setUp()

          # Models and tokenizer
          snake_case			:      str        =			AutoModelForCausalLM.from_pretrained(
              self.model_name							,torch_dtype=torch.floataa							,device_map="""auto"""							)
          snake_case			:      Tuple        =			AutoModelForCausalLM.from_pretrained(self.model_name							,load_in_abit=__a							,device_map="""auto"""							)







     def 			snake_case_      (				self							)      ->				List[str]:
          del self.model_fpaa
          del self.model_abit

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







     def 			snake_case_      (				self							)      ->				int:
          snake_case			:      Tuple        =			self.model_abit.config

          self.assertTrue(hasattr(__a							,"""quantization_config"""							)							)

          snake_case			:      Optional[Any]        =			config.to_dict()
          snake_case			:      Dict        =			config.to_diff_dict()

          snake_case			:      Dict        =			config.to_json_string()







     def 			snake_case_      (				self							)      ->				int:
          from bitsandbytes.nn import Paramsabit

          snake_case			:      List[str]        =			self.model_fpaa.get_memory_footprint()
          snake_case			:      List[Any]        =			self.model_abit.get_memory_footprint()

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







     def 			snake_case_      (				self							)      ->				Dict:
          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(__a							,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 			snake_case_      (				self							)      ->				List[Any]:
          snake_case			:      List[Any]        =			self.tokenizer(self.input_text							,return_tensors="""pt"""							)
          snake_case			:      Dict        =			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=__a							)							,self.EXPECTED_OUTPUTS							)







     def 			snake_case_      (				self							)      ->				List[Any]:
          snake_case			:      Optional[Any]        =			BitsAndBytesConfig()
          snake_case			:      int        =			True

          snake_case			:      Dict        =			AutoModelForCausalLM.from_pretrained(
              self.model_name							,quantization_config=__a							,device_map="""auto"""							)

          snake_case			:      Dict        =			self.tokenizer(self.input_text							,return_tensors="""pt"""							)
          snake_case			:      Optional[Any]        =			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=__a							)							,self.EXPECTED_OUTPUTS							)







     def 			snake_case_      (				self							)      ->				Tuple:
          with self.assertRaises(__a							), tempfile.TemporaryDirectory() as tmpdirname:
               self.model_abit.save_pretrained(__a							)







     def 			snake_case_      (				self							)      ->				Tuple:
          snake_case			:      Optional[int]        =			BitsAndBytesConfig()

          with self.assertRaises(__a							):
               snake_case			:      str        =			AutoModelForCausalLM.from_pretrained(
                   self.model_name							,quantization_config=__a							,load_in_abit=__a							,device_map="""auto"""							,bnb_abit_quant_type="""nf4"""							,)







     def 			snake_case_      (				self							)      ->				Union[str, Any]:
          with self.assertRaises(__a							):
               # Tries with `str`
               self.model_abit.to("""cpu"""							)

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

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

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

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

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

          snake_case			:      Dict        =			self.model_fpaa.to(torch.floataa							)
          snake_case			:      List[str]        =			self.model_fpaa.generate(input_ids=encoded_input["""input_ids"""].to(0							)							,max_new_tokens=10							)

          # Check this does not throw an error
          snake_case			:      List[str]        =			self.model_fpaa.to("""cpu"""							)

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

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







     def 			snake_case_      (				self							)      ->				Optional[int]:
          snake_case			:      Optional[Any]        =			AutoModelForSeqaSeqLM.from_pretrained("""t5-small"""							,load_in_abit=__a							,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 			snake_case_      (				cls							)      ->				List[str]:
          snake_case			:      int        =			"""t5-small"""
          snake_case			:      int        =			"""google/flan-t5-small"""  # flan-t5 uses dense-act instead of dense-relu-dense
          snake_case			:      List[Any]        =			AutoTokenizer.from_pretrained(cls.model_name							)
          snake_case			:      Dict        =			"""Translate in German: Hello, my dog is cute"""







     def 			snake_case_      (				self							)      ->				Any:
          gc.collect()
          torch.cuda.empty_cache()







     def 			snake_case_      (				self							)      ->				List[Any]:
          from transformers import TaForConditionalGeneration

          snake_case			:      int        =			TaForConditionalGeneration._keep_in_fpaa_modules
          snake_case			:      Dict        =			None

          # test with `t5-small`
          snake_case			:      Dict        =			TaForConditionalGeneration.from_pretrained(self.model_name							,load_in_abit=__a							,device_map="""auto"""							)
          snake_case			:      Tuple        =			self.tokenizer(self.input_text							,return_tensors="""pt"""							).to(0							)
          snake_case			:      Dict        =			model.generate(**__a							)

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







     def 			snake_case_      (				self							)      ->				Union[str, Any]:
          import bitsandbytes as bnb

          from transformers import TaForConditionalGeneration

          # test with `t5-small`
          snake_case			:      int        =			TaForConditionalGeneration.from_pretrained(self.model_name							,load_in_abit=__a							,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							)							)

          snake_case			:      int        =			self.tokenizer(self.input_text							,return_tensors="""pt"""							).to(0							)
          snake_case			:      int        =			model.generate(**__a							)

          # test with `flan-t5-small`
          snake_case			:      Dict        =			TaForConditionalGeneration.from_pretrained(
              self.dense_act_model_name							,load_in_abit=__a							,device_map="""auto"""							)
          snake_case			:      int        =			self.tokenizer(self.input_text							,return_tensors="""pt"""							).to(0							)
          snake_case			:      Dict        =			model.generate(**__a							)







class 					_a (a__	):





     '''simple docstring'''



     def 			snake_case_      (				self							)      ->				Optional[int]:
          super().setUp()
          # model_name
          snake_case			:      Union[str, Any]        =			"""bigscience/bloom-560m"""
          snake_case			:      int        =			"""t5-small"""

          # Different types of model

          snake_case			:      Union[str, Any]        =			AutoModel.from_pretrained(self.model_name							,load_in_abit=__a							,device_map="""auto"""							)
          # Sequence classification model
          snake_case			:      Any        =			AutoModelForSequenceClassification.from_pretrained(
              self.model_name							,load_in_abit=__a							,device_map="""auto"""							)
          # CausalLM model
          snake_case			:      int        =			AutoModelForCausalLM.from_pretrained(self.model_name							,load_in_abit=__a							,device_map="""auto"""							)
          # Seq2seq model
          snake_case			:      List[Any]        =			AutoModelForSeqaSeqLM.from_pretrained(
              self.seq_to_seq_name							,load_in_abit=__a							,device_map="""auto"""							)







     def 			snake_case_      (				self							)      ->				Optional[int]:
          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 			snake_case_      (				self							)      ->				str:
          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 (a__	):





     '''simple docstring'''



     def 			snake_case_      (				self							)      ->				Tuple:
          super().setUp()







     def 			snake_case_      (				self							)      ->				int:
          del self.pipe

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







     def 			snake_case_      (				self							)      ->				str:
          snake_case			:      Optional[int]        =			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
          snake_case			:      Optional[Any]        =			self.pipe(self.input_text							)
          self.assertIn(pipeline_output[0]["""generated_text"""]							,self.EXPECTED_OUTPUTS							)







@require_torch_multi_gpu
class 					_a (a__	):





     '''simple docstring'''



     def 			snake_case_      (				self							)      ->				Optional[Any]:
          super().setUp()







     def 			snake_case_      (				self							)      ->				str:
          snake_case			:      Union[str, Any]        =			AutoModelForCausalLM.from_pretrained(
              self.model_name							,load_in_abit=__a							,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
          snake_case			:      Dict        =			self.tokenizer(self.input_text							,return_tensors="""pt"""							)

          # Second real batch
          snake_case			:      List[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=__a							)							,self.EXPECTED_OUTPUTS							)







class 					_a (a__	):





     '''simple docstring'''



     def 			snake_case_      (				self							)      ->				Optional[Any]:
          snake_case			:      Optional[Any]        =			"""facebook/opt-350m"""
          super().setUp()







     def 			snake_case_      (				self							)      ->				str:
          if version.parse(importlib.metadata.version("""bitsandbytes"""							)							) < version.parse("""0.37.0"""							):
               return

          # Step 1: freeze all parameters
          snake_case			:      int        =			AutoModelForCausalLM.from_pretrained(self.model_name							,load_in_abit=__a							)

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

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

        # Step 2: add adapters
          for _, module in model.named_modules():
               if "OPTAttention" in repr(type(__a							)							):
                    snake_case			:      str        =			LoRALayer(module.q_proj							,rank=16							)
                    snake_case			:      str        =			LoRALayer(module.k_proj							,rank=16							)
                    snake_case			:      Union[str, Any]        =			LoRALayer(module.v_proj							,rank=16							)

        # Step 3: dummy batch
          snake_case			:      Optional[Any]        =			self.tokenizer("""Test batch """							,return_tensors="""pt"""							).to(0							)

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

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







class 					_a (a__	):





     '''simple docstring'''



     lowerCAmelCase_  :     int          =					"""gpt2-xl"""
     lowerCAmelCase_  :     Dict          =					3.3191_8548_5415_2187









'''simple docstring'''







from __future__ import annotations







class 					_a :





     '''simple docstring'''



     def __init__(				self							,__a = 0							)      ->				str:
          snake_case			:      List[Any]        =			key







     def 			snake_case_      (				self							,__a							,__a							)      ->				list[str]:
          assert isinstance(__a							,__a							) and isinstance(__a							,__a							)

          snake_case			:      Tuple        =			key or self.__key or 1

          # make sure key is an appropriate size
          key %= 255

          return [chr(ord(__a							) ^ key							) for ch in content]







     def 			snake_case_      (				self							,__a							,__a							)      ->				list[str]:
          assert isinstance(__a							,__a							) and isinstance(__a							,__a							)

          snake_case			:      List[Any]        =			key or self.__key or 1

          # make sure key is an appropriate size
          key %= 255

          return [chr(ord(__a							) ^ key							) for ch in content]







     def 			snake_case_      (				self							,__a							,__a = 0							)      ->				str:
          assert isinstance(__a							,__a							) and isinstance(__a							,__a							)

          snake_case			:      List[Any]        =			key or self.__key or 1

          # make sure key can be any size
          while key > 255:
               key -= 255

          # This will be returned
          snake_case			:      List[str]        =			""""""

          for ch in content:
               ans += chr(ord(__a							) ^ key							)

          return ans







     def 			snake_case_      (				self							,__a							,__a = 0							)      ->				str:
          assert isinstance(__a							,__a							) and isinstance(__a							,__a							)

          snake_case			:      Any        =			key or self.__key or 1

          # make sure key can be any size
          while key > 255:
               key -= 255

          # This will be returned
          snake_case			:      List[str]        =			""""""

          for ch in content:
               ans += chr(ord(__a							) ^ key							)

          return ans







     def 			snake_case_      (				self							,__a							,__a = 0							)      ->				bool:
          assert isinstance(__a							,__a							) and isinstance(__a							,__a							)

          try:
               with open(__a							) as fin, open("""encrypt.out"""							,"""w+"""							) as fout:
                    # actual encrypt-process
                    for line in fin:
                         fout.write(self.encrypt_string(__a							,__a							)							)

          except OSError:
               return False

          return True







     def 			snake_case_      (				self							,__a							,__a							)      ->				bool:
          assert isinstance(__a							,__a							) and isinstance(__a							,__a							)

          try:
               with open(__a							) as fin, open("""decrypt.out"""							,"""w+"""							) as fout:
                    # actual encrypt-process
                    for line in fin:
                         fout.write(self.decrypt_string(__a							,__a							)							)

          except OSError:
               return False

          return True


# Tests
# crypt = XORCipher()
# key = 67

# # test encrypt
# print(crypt.encrypt("hallo welt",key))
# # test decrypt
# print(crypt.decrypt(crypt.encrypt("hallo welt",key), key))

# # test encrypt_string
# print(crypt.encrypt_string("hallo welt",key))

# # test decrypt_string
# print(crypt.decrypt_string(crypt.encrypt_string("hallo welt",key),key))

# if (crypt.encrypt_file("test.txt",key)):
#       print("encrypt successful")
# else:
#       print("encrypt unsuccessful")

# if (crypt.decrypt_file("encrypt.out",key)):
#       print("decrypt successful")
# else:
#       print("decrypt unsuccessful")



