



'''simple docstring'''

from collections import OrderedDict
from typing import Any, List, Mapping, Optional

from ... import PreTrainedTokenizer, TensorType, is_torch_available
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfigWithPast, PatchingSpec
from ...utils import logging


__lowerCAmelCase									=   logging.get_logger(__name__)


__lowerCAmelCase									=   {
    '''Salesforce/codegen-350M-nl''': '''https://huggingface.co/Salesforce/codegen-350M-nl/resolve/main/config.json''',
    '''Salesforce/codegen-350M-multi''': '''https://huggingface.co/Salesforce/codegen-350M-multi/resolve/main/config.json''',
    '''Salesforce/codegen-350M-mono''': '''https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/config.json''',
    '''Salesforce/codegen-2B-nl''': '''https://huggingface.co/Salesforce/codegen-2B-nl/resolve/main/config.json''',
    '''Salesforce/codegen-2B-multi''': '''https://huggingface.co/Salesforce/codegen-2B-multi/resolve/main/config.json''',
    '''Salesforce/codegen-2B-mono''': '''https://huggingface.co/Salesforce/codegen-2B-mono/resolve/main/config.json''',
    '''Salesforce/codegen-6B-nl''': '''https://huggingface.co/Salesforce/codegen-6B-nl/resolve/main/config.json''',
    '''Salesforce/codegen-6B-multi''': '''https://huggingface.co/Salesforce/codegen-6B-multi/resolve/main/config.json''',
    '''Salesforce/codegen-6B-mono''': '''https://huggingface.co/Salesforce/codegen-6B-mono/resolve/main/config.json''',
    '''Salesforce/codegen-16B-nl''': '''https://huggingface.co/Salesforce/codegen-16B-nl/resolve/main/config.json''',
    '''Salesforce/codegen-16B-multi''': '''https://huggingface.co/Salesforce/codegen-16B-multi/resolve/main/config.json''',
    '''Salesforce/codegen-16B-mono''': '''https://huggingface.co/Salesforce/codegen-16B-mono/resolve/main/config.json''',
}

class 	__magic_name__						(    _UpperCamelCase   ):
							lowerCAmelCase       :    str   				=					'codegen'
							lowerCAmelCase       :    Optional[Any]   				=					{
							    'max_position_embeddings': 'n_positions',
							    'hidden_size': 'n_embd',
							    'num_attention_heads': 'n_head',
							    'num_hidden_layers': 'n_layer',
							}







							def __init__(							self					:       Union[str, Any]	,_UpperCAmelCase					:       List[Any]=50400	,_UpperCAmelCase					:       Dict=2048	,_UpperCAmelCase					:       List[str]=2048	,_UpperCAmelCase					:       Union[str, Any]=4096	,_UpperCAmelCase					:       List[Any]=28	,_UpperCAmelCase					:       int=16	,_UpperCAmelCase					:       Optional[Any]=64	,_UpperCAmelCase					:       Tuple=None	,_UpperCAmelCase					:       List[str]="gelu_new"	,_UpperCAmelCase					:       Union[str, Any]=0.0	,_UpperCAmelCase					:       Union[str, Any]=0.0	,_UpperCAmelCase					:       Optional[int]=0.0	,_UpperCAmelCase					:       Optional[int]=1E-5	,_UpperCAmelCase					:       str=0.02	,_UpperCAmelCase					:       List[Any]=True	,_UpperCAmelCase					:       Any=50256	,_UpperCAmelCase					:       int=50256	,_UpperCAmelCase					:       Any=False	,**_UpperCAmelCase					:       List[str]	,):
									_a     :      Optional[Any]        =							vocab_size
									_a     :      Optional[int]        =							n_ctx
									_a     :      Dict        =							n_positions
									_a     :      int        =							n_embd
									_a     :      List[Any]        =							n_layer
									_a     :      Dict        =							n_head
									_a     :      Optional[int]        =							n_inner
									_a     :      Optional[Any]        =							rotary_dim
									_a     :      List[str]        =							activation_function
									_a     :      List[str]        =							resid_pdrop
									_a     :      List[str]        =							embd_pdrop
									_a     :      Union[str, Any]        =							attn_pdrop
									_a     :      List[str]        =							layer_norm_epsilon
									_a     :      Any        =							initializer_range
									_a     :      Any        =							use_cache

									_a     :      Any        =							bos_token_id
									_a     :      Dict        =							eos_token_id

									super().__init__(
									    bos_token_id=_UpperCAmelCase	,eos_token_id=_UpperCAmelCase	,tie_word_embeddings=_UpperCAmelCase	,**_UpperCAmelCase		)

class 	__magic_name__						(    _UpperCamelCase   ):







							def __init__(							self					:       List[str]	,_UpperCAmelCase					:       PretrainedConfig	,_UpperCAmelCase					:       str = "default"	,_UpperCAmelCase					:       List[PatchingSpec] = None	,_UpperCAmelCase					:       bool = False	,):
									super().__init__(_UpperCAmelCase	,task=_UpperCAmelCase	,patching_specs=_UpperCAmelCase	,use_past=_UpperCAmelCase		)
									if not getattr(self._config	,'pad_token_id'	,_UpperCAmelCase		):
											# TODO: how to do that better?
											_a     :      Dict        =							0







							@property
							def 							__lowercase			(							self					:       Tuple		):
									_a     :      List[Any]        =							OrderedDict({'input_ids': {0: 'batch', 1: 'sequence'}}		)
									if self.use_past:
											self.fill_with_past_key_values_(_UpperCAmelCase	,direction='inputs'		)
											_a     :      Union[str, Any]        =							{0: 'batch', 1: 'past_sequence + sequence'}
									else:
											_a     :      Any        =							{0: 'batch', 1: 'sequence'}

									return common_inputs







							@property
							def 							__lowercase			(							self					:       Dict		):
									return self._config.n_layer







							@property
							def 							__lowercase			(							self					:       Optional[int]		):
									return self._config.n_head







							def 							__lowercase			(							self					:       int	,_UpperCAmelCase					:       PreTrainedTokenizer	,_UpperCAmelCase					:       int = -1	,_UpperCAmelCase					:       int = -1	,_UpperCAmelCase					:       bool = False	,_UpperCAmelCase					:       Optional[TensorType] = None	,):
									_a     :      List[str]        =							super(_UpperCAmelCase	,self		).generate_dummy_inputs(
									    _UpperCAmelCase	,batch_size=_UpperCAmelCase	,seq_length=_UpperCAmelCase	,is_pair=_UpperCAmelCase	,framework=_UpperCAmelCase		)

									# We need to order the input in the way they appears in the forward()
									_a     :      List[Any]        =							OrderedDict({'input_ids': common_inputs['input_ids']}		)

									# Need to add the past_keys
									if self.use_past:
											if not is_torch_available():
													raise ValueError('Cannot generate dummy past_keys inputs without PyTorch installed.'		)
											else:
													import torch

													_a       ,					_a     :      int        =							common_inputs['input_ids'].shape
													# Not using the same length for past_key_values
													_a     :      Optional[int]        =							seqlen + 2
													_a     :      int        =							(
													    batch,
													    self.num_attention_heads,
													    past_key_values_length,
													    self._config.hidden_size // self.num_attention_heads,
													)
													_a     :      Any        =							[
													    (torch.zeros(_UpperCAmelCase		), torch.zeros(_UpperCAmelCase		)) for _ in range(self.num_layers		)
													]

									_a     :      Optional[Any]        =							common_inputs['attention_mask']
									if self.use_past:
											_a     :      str        =							ordered_inputs['attention_mask'].dtype
											_a     :      Dict        =							torch.cat(
											    [ordered_inputs['attention_mask'], torch.ones(_UpperCAmelCase	,_UpperCAmelCase	,dtype=_UpperCAmelCase		)]	,dim=1		)

									return ordered_inputs







							@property
							def 							__lowercase			(							self					:       Dict		):
									return 13









'''simple docstring'''

import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch

class 	__magic_name__						(    _UpperCamelCase   ):







							@require_torch
							def 							__lowercase			(							self					:       Tuple		):
									# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
									# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
									# while running an external program

									# python one-liner segments

									# this must be loaded before socket.socket is monkey-patched
									_a     :      Optional[int]        =							'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

									_a     :      List[str]        =							'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

									_a     :      Tuple        =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")\nsocket.socket = offline_socket\n        '

									# Force fetching the files so that we can use the cache
									_a     :      List[Any]        =							'hf-internal-testing/tiny-random-bert'
									BertConfig.from_pretrained(_UpperCAmelCase		)
									BertModel.from_pretrained(_UpperCAmelCase		)
									BertTokenizer.from_pretrained(_UpperCAmelCase		)
									pipeline(task='fill-mask'	,model=_UpperCAmelCase		)

									# baseline - just load from_pretrained with normal network
									_a     :      Optional[int]        =							[sys.executable, '-c', '\n'.join([load, run, mock]		)]

									# should succeed
									_a     :      Tuple        =							self.get_env()
									# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
									_a     :      int        =							'1'
									_a     :      List[Any]        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,0	,result.stderr		)
									self.assertIn('success'	,result.stdout.decode()		)







							@require_torch
							def 							__lowercase			(							self					:       Any		):
									# python one-liner segments
									# this must be loaded before socket.socket is monkey-patched
									_a     :      Dict        =							'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

									_a     :      Optional[int]        =							'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

									_a     :      Optional[Any]        =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")\nsocket.socket = offline_socket\n        '

									# Force fetching the files so that we can use the cache
									_a     :      int        =							'hf-internal-testing/tiny-random-bert'
									BertConfig.from_pretrained(_UpperCAmelCase		)
									BertModel.from_pretrained(_UpperCAmelCase		)
									BertTokenizer.from_pretrained(_UpperCAmelCase		)
									pipeline(task='fill-mask'	,model=_UpperCAmelCase		)

									# baseline - just load from_pretrained with normal network
									_a     :      Optional[int]        =							[sys.executable, '-c', '\n'.join([load, run, mock]		)]

									# should succeed
									_a     :      str        =							self.get_env()
									_a     :      Optional[Any]        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,0	,result.stderr		)
									self.assertIn('success'	,result.stdout.decode()		)







							@require_torch
							def 							__lowercase			(							self					:       List[str]		):
									# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
									# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
									# while running an external program

									# python one-liner segments

									# this must be loaded before socket.socket is monkey-patched
									_a     :      Union[str, Any]        =							'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

									_a     :      Optional[Any]        =							'\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

									_a     :      str        =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")\nsocket.socket = offline_socket\n        '

									# baseline - just load from_pretrained with normal network
									_a     :      Optional[Any]        =							[sys.executable, '-c', '\n'.join([load, run]		)]

									# should succeed
									_a     :      Dict        =							self.get_env()
									_a     :      int        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,0	,result.stderr		)
									self.assertIn('success'	,result.stdout.decode()		)

									# next emulate no network
									_a     :      List[Any]        =							[sys.executable, '-c', '\n'.join([load, mock, run]		)]

									# Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
									# env["TRANSFORMERS_OFFLINE"] = "0"
									# result = subprocess.run(cmd, env=env, check=False, capture_output=True)
									# self.assertEqual(result.returncode, 1, result.stderr)

									# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
									_a     :      int        =							'1'
									_a     :      Any        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,0	,result.stderr		)
									self.assertIn('success'	,result.stdout.decode()		)







							@require_torch
							def 							__lowercase			(							self					:       int		):
									_a     :      Optional[Any]        =							'\nfrom transformers import pipeline\n        '
									_a     :      str        =							'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

									_a     :      List[str]        =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
									_a     :      List[Any]        =							self.get_env()
									_a     :      Dict        =							'1'
									_a     :      Dict        =							[sys.executable, '-c', '\n'.join([load, mock, run]		)]
									_a     :      str        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,1	,result.stderr		)
									self.assertIn(
									    'You cannot infer task automatically within `pipeline` when using offline mode'	,result.stderr.decode().replace('\n'	,''		)	,)







							@require_torch
							def 							__lowercase			(							self					:       int		):
									_a     :      Optional[int]        =							'\nfrom transformers import AutoModel\n        '
									_a     :      List[Any]        =							'\nmname = "hf-internal-testing/test_dynamic_model"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint("success")\n        '

									# baseline - just load from_pretrained with normal network
									_a     :      Union[str, Any]        =							[sys.executable, '-c', '\n'.join([load, run]		)]

									# should succeed
									_a     :      Tuple        =							self.get_env()
									_a     :      List[str]        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,0	,result.stderr		)
									self.assertIn('success'	,result.stdout.decode()		)

									# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
									_a     :      Optional[Any]        =							'1'
									_a     :      Any        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,0	,result.stderr		)
									self.assertIn('success'	,result.stdout.decode()		)




