

import subprocess
import sys

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






class 	UpperCAmelCase__		(	A__	):






  """simple docstring"""


  @require_torch
  def 				lowercase_		(							self						:		Optional[Any] )		->							Any:
       # 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
       SCREAMING_SNAKE_CASE__   =   '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

       SCREAMING_SNAKE_CASE__   =   '''
mname = "hf-internal-testing/tiny-random-bert"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task="fill-mask", model=mname)
print("success")
        '''

       SCREAMING_SNAKE_CASE__   =   '''
import socket
def offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")
socket.socket = offline_socket
        '''

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

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

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






  @require_torch
  def 				lowercase_		(							self						:		List[str] )		->							Dict:
       # python one-liner segments
       # this must be loaded before socket.socket is monkey-patched
       SCREAMING_SNAKE_CASE__   =   '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

       SCREAMING_SNAKE_CASE__   =   '''
mname = "hf-internal-testing/tiny-random-bert"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task="fill-mask", model=mname)
print("success")
        '''

       SCREAMING_SNAKE_CASE__   =   '''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")
socket.socket = offline_socket
        '''

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

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

       # should succeed
       SCREAMING_SNAKE_CASE__   =   self.get_env()
       SCREAMING_SNAKE_CASE__   =   subprocess.run(__lowerCamelCase     ,       env=__lowerCamelCase     ,       check=__lowerCamelCase     ,       capture_output=__lowerCamelCase )
       self.assertEqual(result.returncode     ,       0     ,       result.stderr )
       self.assertIn('''success'''     ,       result.stdout.decode() )






  @require_torch
  def 				lowercase_		(							self						:		Optional[Any] )		->							Dict:
       # 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
       SCREAMING_SNAKE_CASE__   =   '''
from transformers import BertConfig, BertModel, BertTokenizer
        '''

       SCREAMING_SNAKE_CASE__   =   '''
mname = "hf-internal-testing/tiny-random-bert-sharded"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print("success")
        '''

       SCREAMING_SNAKE_CASE__   =   '''
import socket
def offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")
socket.socket = offline_socket
        '''

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

       # should succeed
       SCREAMING_SNAKE_CASE__   =   self.get_env()
       SCREAMING_SNAKE_CASE__   =   subprocess.run(__lowerCamelCase     ,       env=__lowerCamelCase     ,       check=__lowerCamelCase     ,       capture_output=__lowerCamelCase )
       self.assertEqual(result.returncode     ,       0     ,       result.stderr )
       self.assertIn('''success'''     ,       result.stdout.decode() )

       # next emulate no network
       SCREAMING_SNAKE_CASE__   =   [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
       SCREAMING_SNAKE_CASE__   =   '''1'''
       SCREAMING_SNAKE_CASE__   =   subprocess.run(__lowerCamelCase     ,       env=__lowerCamelCase     ,       check=__lowerCamelCase     ,       capture_output=__lowerCamelCase )
       self.assertEqual(result.returncode     ,       0     ,       result.stderr )
       self.assertIn('''success'''     ,       result.stdout.decode() )






  @require_torch
  def 				lowercase_		(							self						:		Tuple )		->							Union[str, Any]:
       SCREAMING_SNAKE_CASE__   =   '''
from transformers import pipeline
        '''
       SCREAMING_SNAKE_CASE__   =   '''
mname = "hf-internal-testing/tiny-random-bert"
pipe = pipeline(model=mname)
        '''

       SCREAMING_SNAKE_CASE__   =   '''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")
socket.socket = offline_socket
        '''
       SCREAMING_SNAKE_CASE__   =   self.get_env()
       SCREAMING_SNAKE_CASE__   =   '''1'''
       SCREAMING_SNAKE_CASE__   =   [sys.executable, '''-c''', '''\n'''.join([load, mock, run] )]
       SCREAMING_SNAKE_CASE__   =   subprocess.run(__lowerCamelCase     ,       env=__lowerCamelCase     ,       check=__lowerCamelCase     ,       capture_output=__lowerCamelCase )
       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						:		List[Any] )		->							Any:
       SCREAMING_SNAKE_CASE__   =   '''
from transformers import AutoModel
        '''
       SCREAMING_SNAKE_CASE__   =   '''
mname = "hf-internal-testing/test_dynamic_model"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print("success")
        '''

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

       # should succeed
       SCREAMING_SNAKE_CASE__   =   self.get_env()
       SCREAMING_SNAKE_CASE__   =   subprocess.run(__lowerCamelCase     ,       env=__lowerCamelCase     ,       check=__lowerCamelCase     ,       capture_output=__lowerCamelCase )
       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
       SCREAMING_SNAKE_CASE__   =   '''1'''
       SCREAMING_SNAKE_CASE__   =   subprocess.run(__lowerCamelCase     ,       env=__lowerCamelCase     ,       check=__lowerCamelCase     ,       capture_output=__lowerCamelCase )
       self.assertEqual(result.returncode     ,       0     ,       result.stderr )
       self.assertIn('''success'''     ,       result.stdout.decode() )





from ...configuration_utils import PretrainedConfig
from ...utils import logging


_SCREAMING_SNAKE_CASE					:  Any             =      logging.get_logger(__name__)

_SCREAMING_SNAKE_CASE					:  int             =      {
    '''edbeeching/decision-transformer-gym-hopper-medium''': (
        '''https://huggingface.co/edbeeching/decision-transformer-gym-hopper-medium/resolve/main/config.json'''
    ),
    # See all DecisionTransformer models at https://huggingface.co/models?filter=decision_transformer
}






class 	UpperCAmelCase__		(	A__	):






  """simple docstring"""


  a									= "decision_transformer"
  a									= ["past_key_values"]
  a									= {
      "max_position_embeddings": "n_positions",
      "num_attention_heads": "n_head",
      "num_hidden_layers": "n_layer",
  }






  def __init__(							self						:		Tuple     ,       __lowerCamelCase						:		Any=17     ,       __lowerCamelCase						:		Any=4     ,       __lowerCamelCase						:		List[Any]=128     ,       __lowerCamelCase						:		Union[str, Any]=4096     ,       __lowerCamelCase						:		Optional[Any]=True     ,       __lowerCamelCase						:		Any=1     ,       __lowerCamelCase						:		List[Any]=1024     ,       __lowerCamelCase						:		Any=3     ,       __lowerCamelCase						:		Dict=1     ,       __lowerCamelCase						:		List[Any]=None     ,       __lowerCamelCase						:		str="relu"     ,       __lowerCamelCase						:		Dict=0.1     ,       __lowerCamelCase						:		int=0.1     ,       __lowerCamelCase						:		List[str]=0.1     ,       __lowerCamelCase						:		str=1e-5     ,       __lowerCamelCase						:		Optional[Any]=0.02     ,       __lowerCamelCase						:		int=True     ,       __lowerCamelCase						:		Tuple=True     ,       __lowerCamelCase						:		Optional[int]=5_0256     ,       __lowerCamelCase						:		Tuple=5_0256     ,       __lowerCamelCase						:		int=False     ,       __lowerCamelCase						:		Optional[int]=False     ,       **__lowerCamelCase						:		Tuple     ,       )		->							Optional[Any]:
       SCREAMING_SNAKE_CASE__   =   state_dim
       SCREAMING_SNAKE_CASE__   =   act_dim
       SCREAMING_SNAKE_CASE__   =   hidden_size
       SCREAMING_SNAKE_CASE__   =   max_ep_len
       SCREAMING_SNAKE_CASE__   =   action_tanh
       SCREAMING_SNAKE_CASE__   =   vocab_size
       SCREAMING_SNAKE_CASE__   =   n_positions
       SCREAMING_SNAKE_CASE__   =   n_layer
       SCREAMING_SNAKE_CASE__   =   n_head
       SCREAMING_SNAKE_CASE__   =   n_inner
       SCREAMING_SNAKE_CASE__   =   activation_function
       SCREAMING_SNAKE_CASE__   =   resid_pdrop
       SCREAMING_SNAKE_CASE__   =   embd_pdrop
       SCREAMING_SNAKE_CASE__   =   attn_pdrop
       SCREAMING_SNAKE_CASE__   =   layer_norm_epsilon
       SCREAMING_SNAKE_CASE__   =   initializer_range
       SCREAMING_SNAKE_CASE__   =   scale_attn_weights
       SCREAMING_SNAKE_CASE__   =   use_cache
       SCREAMING_SNAKE_CASE__   =   scale_attn_by_inverse_layer_idx
       SCREAMING_SNAKE_CASE__   =   reorder_and_upcast_attn

       SCREAMING_SNAKE_CASE__   =   bos_token_id
       SCREAMING_SNAKE_CASE__   =   eos_token_id

       super().__init__(bos_token_id=__lowerCamelCase     ,       eos_token_id=__lowerCamelCase     ,       **__lowerCamelCase )


