





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()      )








from typing import List, Optional, Union

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


SCREAMING_SNAKE_CASE__       :			Any			      =						logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__       :			str			      =						{
    """huggingface/informer-tourism-monthly""": (
        """https://huggingface.co/huggingface/informer-tourism-monthly/resolve/main/config.json"""
    ),
    # See all Informer models at https://huggingface.co/models?filter=informer
}







class   __lowerCAmelCase			(					_UpperCamelCase  ):
   _UpperCamelCase :						Optional[Any]     		=				"""informer"""
   _UpperCamelCase :						Any     		=				{
       """hidden_size""": """d_model""",
       """num_attention_heads""": """encoder_attention_heads""",
       """num_hidden_layers""": """encoder_layers""",
   }





   def __init__( self					,			snake_case = None					,			snake_case = None					,			snake_case = "student_t"					,			snake_case = "nll"					,			snake_case = 1					,			snake_case = None					,			snake_case = "mean"					,			snake_case = 0					,			snake_case = 0					,			snake_case = 0					,			snake_case = 0					,			snake_case = None					,			snake_case = None					,			snake_case = 64					,			snake_case = 32					,			snake_case = 32					,			snake_case = 2					,			snake_case = 2					,			snake_case = 2					,			snake_case = 2					,			snake_case = True					,			snake_case = "gelu"					,			snake_case = 0.05					,			snake_case = 0.1					,			snake_case = 0.1					,			snake_case = 0.1					,			snake_case = 0.1					,			snake_case = 100					,			snake_case = 0.02					,			snake_case=True					,			snake_case = "prob"					,			snake_case = 5					,			snake_case = True					,			**snake_case					,			)				->  Union[str, Any]:
      """simple docstring"""





      a__	:						Optional[Any]      		=  prediction_length
      a__	:						Optional[int]      		=  context_length or prediction_length
      a__	:						Optional[int]      		=  distribution_output
      a__	:						str      		=  loss
      a__	:						Optional[Any]      		=  input_size
      a__	:						int      		=  num_time_features
      a__	:						Optional[int]      		=  lags_sequence if lags_sequence is not None else [1, 2, 3, 4, 5, 6, 7]
      a__	:						Optional[int]      		=  scaling
      a__	:						List[str]      		=  num_dynamic_real_features
      a__	:						Optional[int]      		=  num_static_real_features
      a__	:						Optional[int]      		=  num_static_categorical_features

      # set cardinality
      if cardinality and num_static_categorical_features > 0:
         if len(snake_case      ) != num_static_categorical_features:
            raise ValueError(
                "The cardinality should be a list of the same length as `num_static_categorical_features`"      )
         a__	:						List[Any]      		=  cardinality
      else:
         a__	:						Tuple      		=  [0]

      # set embedding_dimension
      if embedding_dimension and num_static_categorical_features > 0:
         if len(snake_case      ) != num_static_categorical_features:
            raise ValueError(
                "The embedding dimension should be a list of the same length as `num_static_categorical_features`"      )
         a__	:						Tuple      		=  embedding_dimension
      else:
         a__	:						int      		=  [min(50					,			(cat + 1) // 2      ) for cat in self.cardinality]

      a__	:						Optional[Any]      		=  num_parallel_samples

      # Transformer architecture configuration
      a__	:						int      		=  input_size * len(self.lags_sequence      ) + self._number_of_features
      a__	:						Union[str, Any]      		=  d_model
      a__	:						Any      		=  encoder_attention_heads
      a__	:						Optional[Any]      		=  decoder_attention_heads
      a__	:						int      		=  encoder_ffn_dim
      a__	:						List[Any]      		=  decoder_ffn_dim
      a__	:						List[str]      		=  encoder_layers
      a__	:						Any      		=  decoder_layers

      a__	:						List[str]      		=  dropout
      a__	:						int      		=  attention_dropout
      a__	:						List[Any]      		=  activation_dropout
      a__	:						Optional[int]      		=  encoder_layerdrop
      a__	:						Tuple      		=  decoder_layerdrop

      a__	:						Any      		=  activation_function
      a__	:						Tuple      		=  init_std

      a__	:						Optional[int]      		=  use_cache

      # Informer
      a__	:						Union[str, Any]      		=  attention_type
      a__	:						List[str]      		=  sampling_factor
      a__	:						Optional[int]      		=  distil

      super().__init__(is_encoder_decoder=snake_case					,			**snake_case      )





   @property
   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      return (
          sum(self.embedding_dimension      )
          + self.num_dynamic_real_features
          + self.num_time_features
          + self.num_static_real_features
          + self.input_size * 2  # the log1p(abs(loc)) and log(scale) features
      )

