


import copy
import os
from typing import Union

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


SCREAMING_SNAKE_CASE :   List[str]							      =				logging.get_logger(__name__)

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






class _lowerCamelCase(		_a     ):
       lowercase_				:			Any									=    """pix2struct_text_model"""
       lowercase_				:			Any									=    ["""past_key_values"""]
       lowercase_				:			Dict									=    {
           """hidden_size""": """hidden_size""",
           """num_attention_heads""": """num_heads""",
           """num_hidden_layers""": """num_layers""",
       }
       def __init__(      self,   lowerCamelCase=5_02_44,   lowerCamelCase=7_68,   lowerCamelCase=64,   lowerCamelCase=20_48,   lowerCamelCase=12,   lowerCamelCase=12,   lowerCamelCase=32,   lowerCamelCase=1_28,   lowerCamelCase=0.1,   lowerCamelCase=1E-6,   lowerCamelCase=1.0,   lowerCamelCase="gelu_new",   lowerCamelCase=0,   lowerCamelCase=False,   lowerCamelCase=0,   lowerCamelCase=1,   lowerCamelCase=False,   lowerCamelCase=True,   **lowerCamelCase,   ) ->     Union[str, Any]:




           """simple docstring"""

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

           _lowercase :							Dict   	=	eos_token_id
           _lowercase :							List[Any]   	=	decoder_start_token_id

           # for backwards compatibility
           _lowercase :							Union[str, Any]   	=	dense_act_fn

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




       @classmethod
       def     UpperCamelCase  (      cls,   lowerCamelCase,   **lowerCamelCase) ->     "PretrainedConfig":




           """simple docstring"""

           cls._set_token_in_kwargs(lowerCamelCase)

           _lowercase	,       _lowercase :							Any   	=	cls.get_config_dict(lowerCamelCase,   **lowerCamelCase)

           # get the text config dict if we are loading from Pix2StructConfig
           if config_dict.get('model_type') == "pix2struct":
               _lowercase :							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(lowerCamelCase,   **lowerCamelCase)






class _lowerCamelCase(		_a     ):
       lowercase_				:			Optional[Any]									=    """pix2struct_vision_model"""
       def __init__(      self,   lowerCamelCase=7_68,   lowerCamelCase=7_68,   lowerCamelCase=20_48,   lowerCamelCase=64,   lowerCamelCase=12,   lowerCamelCase=12,   lowerCamelCase="gelu_new",   lowerCamelCase=1E-6,   lowerCamelCase=0.0,   lowerCamelCase=0.0,   lowerCamelCase=1E-10,   lowerCamelCase=1.0,   lowerCamelCase=40_96,   lowerCamelCase=32,   lowerCamelCase=1_28,   **lowerCamelCase,   ) ->     List[str]:




           """simple docstring"""

           super().__init__(**lowerCamelCase)

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




       @classmethod
       def     UpperCamelCase  (      cls,   lowerCamelCase,   **lowerCamelCase) ->     "PretrainedConfig":




           """simple docstring"""

           cls._set_token_in_kwargs(lowerCamelCase)

           _lowercase	,       _lowercase :							List[Any]   	=	cls.get_config_dict(lowerCamelCase,   **lowerCamelCase)

           # get the vision config dict if we are loading from Pix2StructConfig
           if config_dict.get('model_type') == "pix2struct":
               _lowercase :							Tuple   	=	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(lowerCamelCase,   **lowerCamelCase)






class _lowerCamelCase(		_a     ):
       lowercase_				:			Optional[Any]									=    """pix2struct"""
       lowercase_				:			Union[str, Any]									=    True
       def __init__(      self,   lowerCamelCase=None,   lowerCamelCase=None,   lowerCamelCase=1.0,   lowerCamelCase=0.0_2,   lowerCamelCase=False,   lowerCamelCase=False,   lowerCamelCase=True,   **lowerCamelCase,   ) ->     Tuple:




           """simple docstring"""

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

           if text_config is None:
               _lowercase :							List[Any]   	=	{}
               logger.info('text_config is None. Initializing the Pix2StructTextConfig with default values.')

           if vision_config is None:
               _lowercase :							Dict   	=	{}
               logger.info('vision_config is None. Initializing the Pix2StructVisionConfig with default values.')

           _lowercase :							Optional[int]   	=	PixaStructTextConfig(**lowerCamelCase)
           _lowercase :							Any   	=	PixaStructVisionConfig(**lowerCamelCase)

           _lowercase :							List[str]   	=	self.text_config.decoder_start_token_id
           _lowercase :							Dict   	=	self.text_config.pad_token_id
           _lowercase :							List[Any]   	=	self.text_config.eos_token_id

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

           _lowercase :							Any   	=	self.initializer_range
           _lowercase :							Optional[int]   	=	self.initializer_range

           _lowercase :							Tuple   	=	is_vqa
       @classmethod
       def     UpperCamelCase  (      cls,   lowerCamelCase,   lowerCamelCase,   **lowerCamelCase) ->     Union[str, Any]:




           """simple docstring"""

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




       def     UpperCamelCase  (      self) ->     List[Any]:




           """simple docstring"""

           _lowercase :							List[Any]   	=	copy.deepcopy(self.__dict__)
           _lowercase :							str   	=	self.text_config.to_dict()
           _lowercase :							int   	=	self.vision_config.to_dict()
           _lowercase :							Tuple   	=	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(		_a     ):
       @require_torch
       def     UpperCamelCase  (      self) ->     int:




           """simple docstring"""

           _lowercase :							Optional[Any]   	=	'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

           _lowercase :							Union[str, 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        '

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

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




           """simple docstring"""

           _lowercase :							int   	=	'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

           _lowercase :							Union[str, 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        '

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

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




           """simple docstring"""

           _lowercase :							Tuple   	=	'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

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

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

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




           """simple docstring"""

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

           _lowercase :							Optional[Any]   	=	'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
           _lowercase :							Tuple   	=	self.get_env()
           _lowercase :							Tuple   	=	'1'
           _lowercase :							Union[str, Any]   	=	[sys.executable, '-c', '\n'.join([load, mock, run])]
           _lowercase :							Tuple   	=	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     UpperCamelCase  (      self) ->     Tuple:




           """simple docstring"""

           _lowercase :							Dict   	=	'\nfrom transformers import AutoModel\n        '
           _lowercase :							int   	=	'\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
           _lowercase :							Optional[int]   	=	[sys.executable, '-c', '\n'.join([load, run])]

           # should succeed
           _lowercase :							int   	=	self.get_env()
           _lowercase :							List[str]   	=	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
           _lowercase :							Tuple   	=	'1'
           _lowercase :							Dict   	=	subprocess.run(lowerCamelCase,   env=lowerCamelCase,   check=lowerCamelCase,   capture_output=lowerCamelCase)
           self.assertEqual(result.returncode,   0,   result.stderr)
           self.assertIn('success',   result.stdout.decode())


