





import copy
import os
from typing import Union

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


SCREAMING_SNAKE_CASE__       :			Dict	      =						logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__       :			Tuple	      =						{
    """google/pix2struct-textcaps-base""": (
        """https://huggingface.co/google/pix2struct-textcaps-base/resolve/main/config.json"""
    ),
}







class   __lowerCAmelCase			(					__lowercase  ):
   _UpperCamelCase :						int     		=				"""pix2struct_text_model"""
   _UpperCamelCase :						List[Any]     		=				["""past_key_values"""]
   _UpperCamelCase :						Union[str, Any]     		=				{
       """hidden_size""": """hidden_size""",
       """num_attention_heads""": """num_heads""",
       """num_hidden_layers""": """num_layers""",
   }





   def __init__( self					,			snake_case=50_244					,			snake_case=768					,			snake_case=64					,			snake_case=2_048					,			snake_case=12					,			snake_case=12					,			snake_case=32					,			snake_case=128					,			snake_case=0.1					,			snake_case=1E-6					,			snake_case=1.0					,			snake_case="gelu_new"					,			snake_case=0					,			snake_case=False					,			snake_case=0					,			snake_case=1					,			snake_case=False					,			snake_case=True					,			**snake_case					,			)				->  str:
      """simple docstring"""





      a__	:						Any      		=  vocab_size
      a__	:						List[str]      		=  hidden_size
      a__	:						Tuple      		=  d_kv
      a__	:						Dict      		=  d_ff
      a__	:						str      		=  num_layers
      a__	:						Any      		=  num_heads
      a__	:						Optional[Any]      		=  relative_attention_num_buckets
      a__	:						List[Any]      		=  relative_attention_max_distance
      a__	:						str      		=  dropout_rate
      a__	:						Tuple      		=  layer_norm_epsilon
      a__	:						List[Any]      		=  initializer_factor
      a__	:						Optional[int]      		=  use_cache

      a__	:						str      		=  eos_token_id
      a__	:						int      		=  decoder_start_token_id

      # for backwards compatibility
      a__	:						int      		=  dense_act_fn

      super().__init__(
          pad_token_id=_A					,			eos_token_id=_A					,			decoder_start_token_id=_A					,			tie_word_embeddings=_A					,			is_decoder=_A					,			**_A					,			)





   @classmethod
   def 	_snake_case ( cls					,			snake_case					,			**snake_case      )				->  List[str]:
      """simple docstring"""





      cls._set_token_in_kwargs(_A      )

      a__		,     a__	:						List[Any]      		=  cls.get_config_dict(_A					,			**_A      )

      # get the text config dict if we are loading from Pix2StructConfig
      if config_dict.get("model_type"      ) == "pix2struct":
         a__	:						Union[str, Any]      		=  config_dict["text_config"]

      if "model_type" in config_dict and hasattr(cls					,			"model_type"      ) and config_dict["model_type"] != cls.model_type:
         logger.warning(
             F"""You are using a model of type {config_dict["model_type"]} to instantiate a model of type """
             F"""{cls.model_type}. This is not supported for all configurations of models and can yield errors."""      )

      return cls.from_dict(_A					,			**_A      )







class   __lowerCAmelCase			(					__lowercase  ):
   _UpperCamelCase :						int     		=				"""pix2struct_vision_model"""





   def __init__( self					,			snake_case=768					,			snake_case=768					,			snake_case=2_048					,			snake_case=64					,			snake_case=12					,			snake_case=12					,			snake_case="gelu_new"					,			snake_case=1E-6					,			snake_case=0.0					,			snake_case=0.0					,			snake_case=1E-10					,			snake_case=1.0					,			snake_case=4_096					,			snake_case=32					,			snake_case=128					,			**snake_case					,			)				->  int:
      """simple docstring"""





      super().__init__(**_A      )

      a__	:						str      		=  hidden_size
      a__	:						List[str]      		=  patch_embed_hidden_size
      a__	:						List[str]      		=  d_ff
      a__	:						int      		=  dropout_rate
      a__	:						Union[str, Any]      		=  num_hidden_layers
      a__	:						Tuple      		=  num_attention_heads
      a__	:						List[Any]      		=  initializer_range
      a__	:						Optional[int]      		=  initializer_factor
      a__	:						Optional[Any]      		=  attention_dropout
      a__	:						List[Any]      		=  layer_norm_eps
      a__	:						List[Any]      		=  dense_act_fn
      a__	:						Tuple      		=  seq_len
      a__	:						Dict      		=  relative_attention_num_buckets
      a__	:						Optional[Any]      		=  relative_attention_max_distance
      a__	:						Optional[Any]      		=  d_kv





   @classmethod
   def 	_snake_case ( cls					,			snake_case					,			**snake_case      )				->  Union[str, Any]:
      """simple docstring"""





      cls._set_token_in_kwargs(_A      )

      a__		,     a__	:						Optional[int]      		=  cls.get_config_dict(_A					,			**_A      )

      # get the vision config dict if we are loading from Pix2StructConfig
      if config_dict.get("model_type"      ) == "pix2struct":
         a__	:						Optional[int]      		=  config_dict["vision_config"]

      if "model_type" in config_dict and hasattr(cls					,			"model_type"      ) and config_dict["model_type"] != cls.model_type:
         logger.warning(
             F"""You are using a model of type {config_dict["model_type"]} to instantiate a model of type """
             F"""{cls.model_type}. This is not supported for all configurations of models and can yield errors."""      )

      return cls.from_dict(_A					,			**_A      )







class   __lowerCAmelCase			(					__lowercase  ):
   _UpperCamelCase :						Tuple     		=				"""pix2struct"""
   _UpperCamelCase :						List[str]     		=				True





   def __init__( self					,			snake_case=None					,			snake_case=None					,			snake_case=1.0					,			snake_case=0.02					,			snake_case=False					,			snake_case=False					,			snake_case=True					,			**snake_case					,			)				->  Any:
      """simple docstring"""





      super().__init__(tie_word_embeddings=_A					,			is_encoder_decoder=_A					,			**_A      )

      if text_config is None:
         a__	:						str      		=  {}
         logger.info("text_config is None. Initializing the Pix2StructTextConfig with default values."      )

      if vision_config is None:
         a__	:						Any      		=  {}
         logger.info("vision_config is None. Initializing the Pix2StructVisionConfig with default values."      )

      a__	:						Dict      		=  PixaStructTextConfig(**_A      )
      a__	:						Optional[Any]      		=  PixaStructVisionConfig(**_A      )

      a__	:						Optional[Any]      		=  self.text_config.decoder_start_token_id
      a__	:						Tuple      		=  self.text_config.pad_token_id
      a__	:						List[str]      		=  self.text_config.eos_token_id

      a__	:						List[Any]      		=  initializer_factor
      a__	:						Any      		=  initializer_range

      a__	:						Dict      		=  self.initializer_range
      a__	:						Dict      		=  self.initializer_range

      a__	:						str      		=  is_vqa





   @classmethod
   def 	_snake_case ( cls					,			snake_case					,			snake_case					,			**snake_case      )				->  str:
      """simple docstring"""





      return cls(text_config=text_config.to_dict()					,			vision_config=vision_config.to_dict()					,			**_A      )





   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						Tuple      		=  copy.deepcopy(self.__dict__      )
      a__	:						int      		=  self.text_config.to_dict()
      a__	:						str      		=  self.vision_config.to_dict()
      a__	:						Any      		=  self.__class__.model_type
      return output








import subprocess
import sys

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







class   __lowerCAmelCase			(					_UpperCamelCase  ):





   @require_torch
   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      a__	:						Tuple      		=  "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

      a__	:						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__	:						Optional[Any]      		=  "\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__	:						Tuple      		=  "hf-internal-testing/tiny-random-bert"
      BertConfig.from_pretrained(snake_case      )
      BertModel.from_pretrained(snake_case      )
      BertTokenizer.from_pretrained(snake_case      )
      pipeline(task="fill-mask"					,			model=snake_case      )

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

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





   @require_torch
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      a__	:						Tuple      		=  "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

      a__	:						Optional[Any]      		=  "\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 socket.error(\"Faking flaky 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(snake_case      )
      BertModel.from_pretrained(snake_case      )
      BertTokenizer.from_pretrained(snake_case      )
      pipeline(task="fill-mask"					,			model=snake_case      )

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

      # should succeed
      a__	:						Any      		=  self.get_env()
      a__	:						int      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )





   @require_torch
   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						List[Any]      		=  "\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        "

      a__	:						Dict      		=  "\nmname = \"hf-internal-testing/tiny-random-bert-sharded\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint(\"success\")\n        "

      a__	:						int      		=  "\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__	:						List[Any]      		=  [sys.executable, "-c", "\n".join([load, run]      )]

      # should succeed
      a__	:						str      		=  self.get_env()
      a__	:						List[str]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      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__	:						Union[str, Any]      		=  "1"
      a__	:						Tuple      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )





   @require_torch
   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						Optional[Any]      		=  "\nfrom transformers import pipeline\n        "
      a__	:						int      		=  "\nmname = \"hf-internal-testing/tiny-random-bert\"\npipe = pipeline(model=mname)\n        "

      a__	:						Dict      		=  "\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")\nsocket.socket = offline_socket\n        "
      a__	:						List[str]      		=  self.get_env()
      a__	:						Union[str, Any]      		=  "1"
      a__	:						List[str]      		=  [sys.executable, "-c", "\n".join([load, mock, run]      )]
      a__	:						Any      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      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 	_snake_case ( self      )				->  Union[str, Any]:
      """simple docstring"""





      a__	:						Any      		=  "\nfrom transformers import AutoModel\n        "
      a__	:						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__	:						List[str]      		=  [sys.executable, "-c", "\n".join([load, run]      )]

      # should succeed
      a__	:						Optional[Any]      		=  self.get_env()
      a__	:						Union[str, Any]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      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__	:						Dict      		=  "1"
      a__	:						Union[str, Any]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )

