


'''simple docstring'''


from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


UpperCAmelCase_						:  List[Any]				  =						logging.get_logger(__name__)


UpperCAmelCase_						:  Union[str, Any]				  =						TypeVar('DatasetType', Dataset, IterableDataset)







def      snake_case_  (    SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__ = None				, SCREAMING_SNAKE_CASE__ = None				, SCREAMING_SNAKE_CASE__ = None				, SCREAMING_SNAKE_CASE__ = None				, SCREAMING_SNAKE_CASE__ = "first_exhausted"				, ):







				"""simple docstring"""





				from .arrow_dataset import Dataset
				from .iterable_dataset import IterableDataset

				if not datasets:
								raise ValueError("""Unable to interleave an empty list of datasets."""	)
				for i, dataset in enumerate(SCREAMING_SNAKE_CASE__	):
								if not isinstance(SCREAMING_SNAKE_CASE__				, (Dataset, IterableDataset)	):
												if isinstance(SCREAMING_SNAKE_CASE__				, (DatasetDict, IterableDatasetDict)	):
																if not dataset:
																				raise ValueError(
																				    f"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
																				    """is an empty dataset dictionary."""	)
																raise ValueError(
																    f"""Dataset at position {i} has at least one split: {list(SCREAMING_SNAKE_CASE__	)}\n"""
																    f"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(SCREAMING_SNAKE_CASE__	)	)}']"""	)
												raise ValueError(
												    f"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(SCREAMING_SNAKE_CASE__	).__name__}."""	)
								if i == 0:
												_SCREAMING_SNAKE_CASE						,			_SCREAMING_SNAKE_CASE					:			Dict							= (
												    (Dataset, IterableDataset) if isinstance(SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__	) else (IterableDataset, Dataset)
												)
								elif not isinstance(SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__	):
												raise ValueError(
												    f"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""	)
				if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
								raise ValueError(f"""{stopping_strategy} is not supported. Please enter a valid stopping_strategy."""	)
				if dataset_type is Dataset:
								return _interleave_map_style_datasets(
								    SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__				, info=SCREAMING_SNAKE_CASE__				, split=SCREAMING_SNAKE_CASE__				, stopping_strategy=SCREAMING_SNAKE_CASE__	)
				else:
								return _interleave_iterable_datasets(
								    SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__				, info=SCREAMING_SNAKE_CASE__				, split=SCREAMING_SNAKE_CASE__				, stopping_strategy=SCREAMING_SNAKE_CASE__	)







def      snake_case_  (    SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__ = None				, SCREAMING_SNAKE_CASE__ = None				, SCREAMING_SNAKE_CASE__ = 0				, ):







				"""simple docstring"""





				if not dsets:
								raise ValueError("""Unable to concatenate an empty list of datasets."""	)
				for i, dataset in enumerate(SCREAMING_SNAKE_CASE__	):
								if not isinstance(SCREAMING_SNAKE_CASE__				, (Dataset, IterableDataset)	):
												if isinstance(SCREAMING_SNAKE_CASE__				, (DatasetDict, IterableDatasetDict)	):
																if not dataset:
																				raise ValueError(
																				    f"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
																				    """is an empty dataset dictionary."""	)
																raise ValueError(
																    f"""Dataset at position {i} has at least one split: {list(SCREAMING_SNAKE_CASE__	)}\n"""
																    f"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(SCREAMING_SNAKE_CASE__	)	)}']"""	)
												raise ValueError(
												    f"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(SCREAMING_SNAKE_CASE__	).__name__}."""	)
								if i == 0:
												_SCREAMING_SNAKE_CASE						,			_SCREAMING_SNAKE_CASE					:			Any							= (
												    (Dataset, IterableDataset) if isinstance(SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__	) else (IterableDataset, Dataset)
												)
								elif not isinstance(SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__	):
												raise ValueError(
												    f"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""	)
				if dataset_type is Dataset:
								return _concatenate_map_style_datasets(SCREAMING_SNAKE_CASE__				, info=SCREAMING_SNAKE_CASE__				, split=SCREAMING_SNAKE_CASE__				, axis=SCREAMING_SNAKE_CASE__	)
				else:
								return _concatenate_iterable_datasets(SCREAMING_SNAKE_CASE__				, info=SCREAMING_SNAKE_CASE__				, split=SCREAMING_SNAKE_CASE__				, axis=SCREAMING_SNAKE_CASE__	)








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




