


'''simple docstring'''


from typing import List, Optional, Union

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


UpperCAmelCase_						:  int				  =						logging.get_logger(__name__)

UpperCAmelCase_						:  List[str]				  =						{
    'huggingface/time-series-transformer-tourism-monthly': (
        'https://huggingface.co/huggingface/time-series-transformer-tourism-monthly/resolve/main/config.json'
    ),
    # See all TimeSeriesTransformer models at https://huggingface.co/models?filter=time_series_transformer
}





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							A_				: List[Any]       =    """time_series_transformer"""
							A_				: int       =    {
							    """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 = [1, 2, 3, 4, 5, 6, 7]       ,   __snake_case = "mean"       ,   __snake_case = 0       ,   __snake_case = 0       ,   __snake_case = 0       ,   __snake_case = 0       ,   __snake_case = None       ,   __snake_case = None       ,   __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 = 64       ,   __snake_case = 0.1       ,   __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       ,   ):
											# time series specific configuration
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= prediction_length
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= context_length or prediction_length
											_SCREAMING_SNAKE_CASE					:			Any							= distribution_output
											_SCREAMING_SNAKE_CASE					:			List[Any]							= loss
											_SCREAMING_SNAKE_CASE					:			List[Any]							= input_size
											_SCREAMING_SNAKE_CASE					:			int							= num_time_features
											_SCREAMING_SNAKE_CASE					:			str							= lags_sequence
											_SCREAMING_SNAKE_CASE					:			List[str]							= scaling
											_SCREAMING_SNAKE_CASE					:			Any							= num_dynamic_real_features
											_SCREAMING_SNAKE_CASE					:			Any							= num_static_real_features
											_SCREAMING_SNAKE_CASE					:			List[str]							= num_static_categorical_features
											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`"""    )
															_SCREAMING_SNAKE_CASE					:			List[Any]							= cardinality
											else:
															_SCREAMING_SNAKE_CASE					:			Any							= [0]
											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`"""    )
															_SCREAMING_SNAKE_CASE					:			Dict							= embedding_dimension
											else:
															_SCREAMING_SNAKE_CASE					:			Tuple							= [min(50       ,   (cat + 1) // 2    ) for cat in self.cardinality]
											_SCREAMING_SNAKE_CASE					:			Dict							= num_parallel_samples

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

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

											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= activation_function
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= init_std

											_SCREAMING_SNAKE_CASE					:			Dict							= use_cache

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






							@property
							def UpperCAmelCase_      (       self    ):
											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
											)








'''simple docstring'''


import subprocess
import sys

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





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							@require_torch
							def UpperCAmelCase_      (       self    ):
											# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
											# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
											# while running an external program

											# python one-liner segments

											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

											_SCREAMING_SNAKE_CASE					:			Dict							= """
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\")
        """

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

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






							@require_torch
							def UpperCAmelCase_      (       self    ):
											# python one-liner segments
											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

											_SCREAMING_SNAKE_CASE					:			Any							= """
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\")
        """

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

											# should succeed
											_SCREAMING_SNAKE_CASE					:			str							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			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 UpperCAmelCase_      (       self    ):
											# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
											# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
											# while running an external program

											# python one-liner segments

											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Tuple							= """
from transformers import BertConfig, BertModel, BertTokenizer
        """

											_SCREAMING_SNAKE_CASE					:			str							= """
mname = \"hf-internal-testing/tiny-random-bert-sharded\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print(\"success\")
        """

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

											# should succeed
											_SCREAMING_SNAKE_CASE					:			List[str]							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Optional[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()    )

											# next emulate no network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [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
											_SCREAMING_SNAKE_CASE					:			Dict							= """1"""
											_SCREAMING_SNAKE_CASE					:			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 UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import pipeline
        """
											_SCREAMING_SNAKE_CASE					:			int							= """
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

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

											# should succeed
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			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()    )

											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """1"""
											_SCREAMING_SNAKE_CASE					:			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()    )




