


'''simple docstring'''


import re
import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							A_				: int       =    ["""image_processor""", """tokenizer"""]
							A_				: Optional[int]       =    """AutoImageProcessor"""
							A_				: Optional[Any]       =    """AutoTokenizer"""






							def __init__(       self       ,   __snake_case=None       ,   __snake_case=None       ,   **__snake_case    ):
											_SCREAMING_SNAKE_CASE					:			Tuple							= None
											if "feature_extractor" in kwargs:
															warnings.warn(
															    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
															    """ instead."""       ,   __snake_case       ,   )
															_SCREAMING_SNAKE_CASE					:			Tuple							= kwargs.pop("""feature_extractor"""    )

											_SCREAMING_SNAKE_CASE					:			int							= image_processor if image_processor is not None else feature_extractor
											if image_processor is None:
															raise ValueError("""You need to specify an `image_processor`."""    )
											if tokenizer is None:
															raise ValueError("""You need to specify a `tokenizer`."""    )

											super().__init__(__snake_case       ,   __snake_case    )
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= self.image_processor
											_SCREAMING_SNAKE_CASE					:			List[Any]							= False






							def __call__(       self       ,   *__snake_case       ,   **__snake_case    ):
											# For backward compatibility
											if self._in_target_context_manager:
															return self.current_processor(*__snake_case       ,   **__snake_case    )

											_SCREAMING_SNAKE_CASE					:			List[str]							= kwargs.pop("""images"""       ,   __snake_case    )
											_SCREAMING_SNAKE_CASE					:			Dict							= kwargs.pop("""text"""       ,   __snake_case    )
											if len(__snake_case    ) > 0:
															_SCREAMING_SNAKE_CASE					:			int							= args[0]
															_SCREAMING_SNAKE_CASE					:			Any							= args[1:]

											if images is None and text is None:
															raise ValueError("""You need to specify either an `images` or `text` input to process."""    )

											if images is not None:
															_SCREAMING_SNAKE_CASE					:			Any							= self.image_processor(__snake_case       ,   *__snake_case       ,   **__snake_case    )
											if text is not None:
															_SCREAMING_SNAKE_CASE					:			str							= self.tokenizer(__snake_case       ,   **__snake_case    )

											if text is None:
															return inputs
											elif images is None:
															return encodings
											else:
															_SCREAMING_SNAKE_CASE					:			Dict							= encodings["""input_ids"""]
															return inputs






							def UpperCAmelCase_      (       self       ,   *__snake_case       ,   **__snake_case    ):
											return self.tokenizer.batch_decode(*__snake_case       ,   **__snake_case    )






							def UpperCAmelCase_      (       self       ,   *__snake_case       ,   **__snake_case    ):
											return self.tokenizer.decode(*__snake_case       ,   **__snake_case    )






							@contextmanager
							def UpperCAmelCase_      (       self    ):
											warnings.warn(
											    """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
											    """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
											    """your images inputs, or in a separate call."""    )
											_SCREAMING_SNAKE_CASE					:			List[Any]							= True
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= self.tokenizer
											yield
											_SCREAMING_SNAKE_CASE					:			Tuple							= self.image_processor
											_SCREAMING_SNAKE_CASE					:			List[str]							= False






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case=False       ,   __snake_case=None    ):
											if added_vocab is None:
															_SCREAMING_SNAKE_CASE					:			int							= self.tokenizer.get_added_vocab()

											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= {}

											while tokens:
															_SCREAMING_SNAKE_CASE					:			List[Any]							= re.search(r"""<s_(.*?)>"""       ,   __snake_case       ,   re.IGNORECASE    )
															if start_token is None:
																			break
															_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= start_token.group(1    )
															_SCREAMING_SNAKE_CASE					:			List[str]							= re.search(rf"""</s_{key}>"""       ,   __snake_case       ,   re.IGNORECASE    )
															_SCREAMING_SNAKE_CASE					:			List[Any]							= start_token.group()
															if end_token is None:
																			_SCREAMING_SNAKE_CASE					:			Optional[Any]							= tokens.replace(__snake_case       ,   """"""    )
															else:
																			_SCREAMING_SNAKE_CASE					:			int							= end_token.group()
																			_SCREAMING_SNAKE_CASE					:			Dict							= re.escape(__snake_case    )
																			_SCREAMING_SNAKE_CASE					:			List[Any]							= re.escape(__snake_case    )
																			_SCREAMING_SNAKE_CASE					:			str							= re.search(f"""{start_token_escaped}(.*?){end_token_escaped}"""       ,   __snake_case       ,   re.IGNORECASE    )
																			if content is not None:
																							_SCREAMING_SNAKE_CASE					:			Any							= content.group(1    ).strip()
																							if r"<s_" in content and r"</s_" in content:  # non-leaf node
																											_SCREAMING_SNAKE_CASE					:			str							= self.tokenajson(__snake_case       ,   is_inner_value=__snake_case       ,   added_vocab=__snake_case    )
																											if value:
																															if len(__snake_case    ) == 1:
																																			_SCREAMING_SNAKE_CASE					:			Optional[int]							= value[0]
																															_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= value
																							else:  # leaf nodes
																											_SCREAMING_SNAKE_CASE					:			List[Any]							= []
																											for leaf in content.split(r"""<sep/>"""    ):
																															_SCREAMING_SNAKE_CASE					:			List[str]							= leaf.strip()
																															if leaf in added_vocab and leaf[0] == "<" and leaf[-2:] == "/>":
																																			_SCREAMING_SNAKE_CASE					:			Tuple							= leaf[1:-2]  # for categorical special tokens
																															output[key].append(__snake_case    )
																											if len(output[key]    ) == 1:
																															_SCREAMING_SNAKE_CASE					:			Optional[int]							= output[key][0]

																			_SCREAMING_SNAKE_CASE					:			int							= tokens[tokens.find(__snake_case    ) + len(__snake_case    ) :].strip()
																			if tokens[:6] == r"<sep/>":  # non-leaf nodes
																							return [output] + self.tokenajson(tokens[6:]       ,   is_inner_value=__snake_case       ,   added_vocab=__snake_case    )

											if len(__snake_case    ):
															return [output] if is_inner_value else output
											else:
															return [] if is_inner_value else {"text_sequence": tokens}






							@property
							def UpperCAmelCase_      (       self    ):
											warnings.warn(
											    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""       ,   __snake_case       ,   )
											return self.image_processor_class






							@property
							def UpperCAmelCase_      (       self    ):
											warnings.warn(
											    """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""       ,   __snake_case       ,   )
											return self.image_processor








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




