


'''simple docstring'''


import subprocess
import sys

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






class 	lowerCAmelCase__			(			SCREAMING_SNAKE_CASE_	):




   @require_torch
   def 							lowerCAmelCase__    (		self  :     Dict )   ->List[str]:



     '''simple docstring'''







     _UpperCAmelCase		:       Optional[Any]								=      '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

     _UpperCAmelCase		:       int								=      '''
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")
        '''

     _UpperCAmelCase		:       Dict								=      '''
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
     _UpperCAmelCase		:       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
     _UpperCAmelCase		:       List[Any]								=      [sys.executable, '''-c''', '''\n'''.join([load, run, mock] )]

     # should succeed
     _UpperCAmelCase		:       List[Any]								=      self.get_env()
     # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
     _UpperCAmelCase		:       Any								=      '''1'''
     _UpperCAmelCase		:       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() )




   @require_torch
   def 							lowerCAmelCase__    (		self  :     Union[str, Any] )   ->int:



     '''simple docstring'''







     _UpperCAmelCase		:       str								=      '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

     _UpperCAmelCase		:       List[str]								=      '''
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")
        '''

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

     # should succeed
     _UpperCAmelCase		:       Optional[int]								=      self.get_env()
     _UpperCAmelCase		:       int								=      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 							lowerCAmelCase__    (		self  :     Optional[int] )   ->Union[str, Any]:



     '''simple docstring'''







     _UpperCAmelCase		:       List[str]								=      '''
from transformers import BertConfig, BertModel, BertTokenizer
        '''

     _UpperCAmelCase		:       Union[str, Any]								=      '''
mname = "hf-internal-testing/tiny-random-bert-sharded"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print("success")
        '''

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

     # should succeed
     _UpperCAmelCase		:       Dict								=      self.get_env()
     _UpperCAmelCase		:       Optional[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
     _UpperCAmelCase		:       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
     _UpperCAmelCase		:       List[Any]								=      '''1'''
     _UpperCAmelCase		:       Dict								=      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 							lowerCAmelCase__    (		self  :     Dict )   ->List[str]:



     '''simple docstring'''







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

     _UpperCAmelCase		:       Optional[int]								=      '''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")
socket.socket = offline_socket
        '''
     _UpperCAmelCase		:       int								=      self.get_env()
     _UpperCAmelCase		:       Optional[int]								=      '''1'''
     _UpperCAmelCase		:       int								=      [sys.executable, '''-c''', '''\n'''.join([load, mock, run] )]
     _UpperCAmelCase		:       Dict								=      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 							lowerCAmelCase__    (		self  :     Dict )   ->List[str]:



     '''simple docstring'''







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

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







'''simple docstring'''


from typing import List, Optional, Union

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






class 	lowerCAmelCase__			(			UpperCAmelCase__	):
			lowerCAmelCase :	Any          =     ["image_processor", "tokenizer"]
			lowerCAmelCase :	List[Any]          =     "BlipImageProcessor"
			lowerCAmelCase :	Union[str, Any]          =     "AutoTokenizer"




			def __init__(		self  :     Any	,  lowerCamelCase__  :     Tuple	,  lowerCamelCase__  :     int )   ->Optional[int]:



					'''simple docstring'''







					_UpperCAmelCase		:       Optional[int]								=      False
					super().__init__(lowerCamelCase__	,  lowerCamelCase__ )
					_UpperCAmelCase		:       Tuple								=      self.image_processor




			def __call__(		self  :     Dict	,  lowerCamelCase__  :     ImageInput = None	,  lowerCamelCase__  :     Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None	,  lowerCamelCase__  :     bool = True	,  lowerCamelCase__  :     Union[bool, str, PaddingStrategy] = False	,  lowerCamelCase__  :     Union[bool, str, TruncationStrategy] = None	,  lowerCamelCase__  :     Optional[int] = None	,  lowerCamelCase__  :     int = 0	,  lowerCamelCase__  :     Optional[int] = None	,  lowerCamelCase__  :     Optional[bool] = None	,  lowerCamelCase__  :     bool = False	,  lowerCamelCase__  :     bool = False	,  lowerCamelCase__  :     bool = False	,  lowerCamelCase__  :     bool = False	,  lowerCamelCase__  :     bool = False	,  lowerCamelCase__  :     bool = True	,  lowerCamelCase__  :     Optional[Union[str, TensorType]] = None	,  **lowerCamelCase__  :     Tuple	,  )   ->BatchEncoding:



					'''simple docstring'''







					if images is None and text is None:
							raise ValueError("You have to specify either images or text." )

					# Get only text
					if images is None:
							_UpperCAmelCase		:       Optional[int]								=      self.tokenizer
							_UpperCAmelCase		:       List[Any]								=      self.tokenizer(
							    text=lowerCamelCase__	,  add_special_tokens=lowerCamelCase__	,  padding=lowerCamelCase__	,  truncation=lowerCamelCase__	,  max_length=lowerCamelCase__	,  stride=lowerCamelCase__	,  pad_to_multiple_of=lowerCamelCase__	,  return_attention_mask=lowerCamelCase__	,  return_overflowing_tokens=lowerCamelCase__	,  return_special_tokens_mask=lowerCamelCase__	,  return_offsets_mapping=lowerCamelCase__	,  return_token_type_ids=lowerCamelCase__	,  return_length=lowerCamelCase__	,  verbose=lowerCamelCase__	,  return_tensors=lowerCamelCase__	,  **lowerCamelCase__	,  )
							return text_encoding

					# add pixel_values
					_UpperCAmelCase		:       Optional[int]								=      self.image_processor(lowerCamelCase__	,  return_tensors=lowerCamelCase__ )

					if text is not None:
							_UpperCAmelCase		:       Dict								=      self.tokenizer(
							    text=lowerCamelCase__	,  add_special_tokens=lowerCamelCase__	,  padding=lowerCamelCase__	,  truncation=lowerCamelCase__	,  max_length=lowerCamelCase__	,  stride=lowerCamelCase__	,  pad_to_multiple_of=lowerCamelCase__	,  return_attention_mask=lowerCamelCase__	,  return_overflowing_tokens=lowerCamelCase__	,  return_special_tokens_mask=lowerCamelCase__	,  return_offsets_mapping=lowerCamelCase__	,  return_token_type_ids=lowerCamelCase__	,  return_length=lowerCamelCase__	,  verbose=lowerCamelCase__	,  return_tensors=lowerCamelCase__	,  **lowerCamelCase__	,  )
					else:
							_UpperCAmelCase		:       int								=      None

					if text_encoding is not None:
							encoding_image_processor.update(lowerCamelCase__ )

					return encoding_image_processor




			def 							lowerCAmelCase__    (		self  :     List[Any]	,  *lowerCamelCase__  :     Optional[int]	,  **lowerCamelCase__  :     Dict )   ->Optional[Any]:



					'''simple docstring'''







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




			def 							lowerCAmelCase__    (		self  :     int	,  *lowerCamelCase__  :     Dict	,  **lowerCamelCase__  :     str )   ->Optional[int]:



					'''simple docstring'''







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






			@property
			# Copied from transformers.models.blip.processing_blip.BlipProcessor.model_input_names
			def 							lowerCAmelCase__    (		self  :     Any )   ->Optional[int]:



					'''simple docstring'''







					_UpperCAmelCase		:       Optional[Any]								=      self.tokenizer.model_input_names
					_UpperCAmelCase		:       Dict								=      self.image_processor.model_input_names
					return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names ) )



