



'''simple docstring'''





import subprocess
import sys

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






class __A     (    A		):



 '''simple docstring'''







 @require_torch
 def    a__    (self			)       ->	Optional[Any]:


      """simple docstring"""



      _a           =    '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

      _a           =    '''
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")
        '''

      _a           =    '''
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
      _a           =    '''hf-internal-testing/tiny-random-bert'''
      BertConfig.from_pretrained(A			)
      BertModel.from_pretrained(A			)
      BertTokenizer.from_pretrained(A			)
      pipeline(task='''fill-mask'''					,  model=A			)

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

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



 @require_torch
 def    a__    (self			)       ->	Dict:


      """simple docstring"""



      _a           =    '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

      _a           =    '''
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")
        '''

      _a           =    '''
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
      _a           =    '''hf-internal-testing/tiny-random-bert'''
      BertConfig.from_pretrained(A			)
      BertModel.from_pretrained(A			)
      BertTokenizer.from_pretrained(A			)
      pipeline(task='''fill-mask'''					,  model=A			)

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

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



 @require_torch
 def    a__    (self			)       ->	Optional[Any]:


      """simple docstring"""



      _a           =    '''
from transformers import BertConfig, BertModel, BertTokenizer
        '''

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

      _a           =    '''
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
      _a           =    [sys.executable, '''-c''', '''\n'''.join([load, run]			)]

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

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



 @require_torch
 def    a__    (self			)       ->	Optional[Any]:


      """simple docstring"""



      _a           =    '''
from transformers import pipeline
        '''
      _a           =    '''
mname = "hf-internal-testing/tiny-random-bert"
pipe = pipeline(model=mname)
        '''

      _a           =    '''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")
socket.socket = offline_socket
        '''
      _a           =    self.get_env()
      _a           =    '''1'''
      _a           =    [sys.executable, '''-c''', '''\n'''.join([load, mock, run]			)]
      _a           =    subprocess.run(A					,  env=A					,  check=A					,  capture_output=A			)
      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    a__    (self			)       ->	Optional[int]:


      """simple docstring"""



      _a           =    '''
from transformers import AutoModel
        '''
      _a           =    '''
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
      _a           =    [sys.executable, '''-c''', '''\n'''.join([load, run]			)]

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




'''simple docstring'''




import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType






class 							a__(   lowerCAmelCase__      ):


			'''simple docstring'''






			UpperCAmelCase_     :	str						=			['''image_processor''', '''tokenizer''']
			UpperCAmelCase_     :	str						=			'''ViltImageProcessor'''
			UpperCAmelCase_     :	Any						=			('''BertTokenizer''', '''BertTokenizerFast''')



			def __init__(  self			,       __lowerCAmelCase=None			,       __lowerCAmelCase=None			,       **__lowerCAmelCase):




					"""simple docstring"""


					lowerCAmelCase     =    None
					if "feature_extractor" in kwargs:
							warnings.warn(
							    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
							    """ instead."""			,       __lowerCAmelCase			,       )
							lowerCAmelCase     =    kwargs.pop("""feature_extractor""")

					lowerCAmelCase     =    image_processor if image_processor is not None else feature_extractor
					if image_processor is None:
							raise ValueError("""You need to specify an `image_processor`.""")
					if tokenizer is None:
							raise ValueError("""You need to specify a `tokenizer`.""")

					super().__init__(__lowerCAmelCase			,       __lowerCAmelCase)
					lowerCAmelCase     =    self.image_processor



			def __call__(  self			,       __lowerCAmelCase			,       __lowerCAmelCase = None			,       __lowerCAmelCase = True			,       __lowerCAmelCase = False			,       __lowerCAmelCase = None			,       __lowerCAmelCase = None			,       __lowerCAmelCase = 0			,       __lowerCAmelCase = None			,       __lowerCAmelCase = None			,       __lowerCAmelCase = None			,       __lowerCAmelCase = False			,       __lowerCAmelCase = False			,       __lowerCAmelCase = False			,       __lowerCAmelCase = False			,       __lowerCAmelCase = True			,       __lowerCAmelCase = None			,       **__lowerCAmelCase			,       ):




					"""simple docstring"""


					lowerCAmelCase     =    self.tokenizer(
					    text=__lowerCAmelCase			,       add_special_tokens=__lowerCAmelCase			,       padding=__lowerCAmelCase			,       truncation=__lowerCAmelCase			,       max_length=__lowerCAmelCase			,       stride=__lowerCAmelCase			,       pad_to_multiple_of=__lowerCAmelCase			,       return_token_type_ids=__lowerCAmelCase			,       return_attention_mask=__lowerCAmelCase			,       return_overflowing_tokens=__lowerCAmelCase			,       return_special_tokens_mask=__lowerCAmelCase			,       return_offsets_mapping=__lowerCAmelCase			,       return_length=__lowerCAmelCase			,       verbose=__lowerCAmelCase			,       return_tensors=__lowerCAmelCase			,       **__lowerCAmelCase			,       )
					# add pixel_values + pixel_mask
					lowerCAmelCase     =    self.image_processor(__lowerCAmelCase			,       return_tensors=__lowerCAmelCase)
					encoding.update(__lowerCAmelCase)

					return encoding



			def 				a_			(  self			,       *__lowerCAmelCase			,       **__lowerCAmelCase):




					"""simple docstring"""


					return self.tokenizer.batch_decode(*__lowerCAmelCase			,       **__lowerCAmelCase)



			def 				a_			(  self			,       *__lowerCAmelCase			,       **__lowerCAmelCase):




					"""simple docstring"""


					return self.tokenizer.decode(*__lowerCAmelCase			,       **__lowerCAmelCase)



			@property
			def 				a_			(  self):




					"""simple docstring"""


					lowerCAmelCase     =    self.tokenizer.model_input_names
					lowerCAmelCase     =    self.image_processor.model_input_names
					return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names))



			@property
			def 				a_			(  self):




					"""simple docstring"""


					warnings.warn(
					    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""			,       __lowerCAmelCase			,       )
					return self.image_processor_class







			@property
			def 				a_			(  self):




					"""simple docstring"""


					warnings.warn(
					    """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""			,       __lowerCAmelCase			,       )
					return self.image_processor




