





'''simple docstring'''




import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment



def 		__A		(  lowerCamelCase_			,			lowerCamelCase_=()			,			lowerCamelCase_=None			,			lowerCamelCase_="no"			,			lowerCamelCase_="29500"	):


		"""simple docstring"""



		SCREAMING_SNAKE_CASE :				Union[str, Any]		=			False
		SCREAMING_SNAKE_CASE :				Any		=			False
		if any(key.startswith("""KAGGLE"""	) for key in os.environ.keys()	):
				SCREAMING_SNAKE_CASE :				Union[str, Any]		=			True
		elif "IPython" in sys.modules:
				SCREAMING_SNAKE_CASE :				List[Any]		=			"""google.colab""" in str(sys.modules["""IPython"""].get_ipython()	)

		try:
				SCREAMING_SNAKE_CASE :				Optional[Any]		=			PrecisionType(mixed_precision.lower()	)
		except ValueError:
				raise ValueError(
				    f'''Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.'''	)

		if (in_colab or in_kaggle) and (os.environ.get("""TPU_NAME"""			,			lowerCamelCase_	) is not None):
				# TPU launch
				import torch_xla.distributed.xla_multiprocessing as xmp

				if len(AcceleratorState._shared_state	) > 0:
						raise ValueError(
						    """To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside """
						    """your training function. Restart your notebook and make sure no cells initializes an """
						    """`Accelerator`."""	)
				if num_processes is None:
						SCREAMING_SNAKE_CASE :				Optional[int]		=			8

				SCREAMING_SNAKE_CASE :				Union[str, Any]		=			PrepareForLaunch(lowerCamelCase_			,			distributed_type="""TPU"""	)
				print(f'''Launching a training on {num_processes} TPU cores.'''	)
				xmp.spawn(lowerCamelCase_			,			args=lowerCamelCase_			,			nprocs=lowerCamelCase_			,			start_method="""fork"""	)
		elif in_colab:
				# No need for a distributed launch otherwise as it's either CPU or one GPU.
				if torch.cuda.is_available():
						print("""Launching training on one GPU."""	)
				else:
						print("""Launching training on one CPU."""	)
				function(*lowerCamelCase_	)
		else:
				if num_processes is None:
						raise ValueError(
						    """You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call."""	)

				if num_processes > 1:
						# Multi-GPU launch
						from torch.multiprocessing import start_processes
						from torch.multiprocessing.spawn import ProcessRaisedException

						if len(AcceleratorState._shared_state	) > 0:
								raise ValueError(
								    """To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized """
								    """inside your training function. Restart your notebook and make sure no cells initializes an """
								    """`Accelerator`."""	)

						if torch.cuda.is_initialized():
								raise ValueError(
								    """To launch a multi-GPU training from your notebook, you need to avoid running any instruction """
								    """using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA """
								    """function."""	)

						# torch.distributed will expect a few environment variable to be here. We set the ones common to each
						# process here (the other ones will be set be the launcher).
						with patch_environment(
						    world_size=lowerCamelCase_			,			master_addr="""127.0.01"""			,			master_port=lowerCamelCase_			,			mixed_precision=lowerCamelCase_	):
								SCREAMING_SNAKE_CASE :				Optional[Any]		=			PrepareForLaunch(lowerCamelCase_			,			distributed_type="""MULTI_GPU"""	)
								print(f'''Launching training on {num_processes} GPUs.'''	)
								try:
										start_processes(lowerCamelCase_			,			args=lowerCamelCase_			,			nprocs=lowerCamelCase_			,			start_method="""fork"""	)
								except ProcessRaisedException as e:
										if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
												raise RuntimeError(
												    """CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. """
												    """This likely stems from an outside import causing issues once the `notebook_launcher()` is called. """
												    """Please review your imports and test them when running the `notebook_launcher()` to identify """
												    """which one is problematic."""	) from e

				else:
						# No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
						if is_mps_available():
								SCREAMING_SNAKE_CASE :				str		=			"""1"""
								print("""Launching training on MPS."""	)
						elif torch.cuda.is_available():
								print("""Launching training on one GPU."""	)
						else:
								print("""Launching training on CPU."""	)
						function(*lowerCamelCase_	)



def 		__A		(  lowerCamelCase_			,			lowerCamelCase_=()			,			lowerCamelCase_=2	):


		"""simple docstring"""



		from torch.multiprocessing import start_processes

		with tempfile.NamedTemporaryFile() as tmp_file:
				# torch.distributed will expect a few environment variable to be here. We set the ones common to each
				# process here (the other ones will be set be the launcher).
				with patch_environment(
				    world_size=lowerCamelCase_			,			master_addr="""127.0.01"""			,			master_port="""29500"""			,			accelerate_mixed_precision="""no"""			,			accelerate_debug_rdv_file=tmp_file.name			,			accelerate_use_cpu="""yes"""			,			):
						SCREAMING_SNAKE_CASE :				str		=			PrepareForLaunch(lowerCamelCase_			,			debug=lowerCamelCase_	)
						start_processes(lowerCamelCase_			,			args=lowerCamelCase_			,			nprocs=lowerCamelCase_			,			start_method="""fork"""	)










'''simple docstring'''




import unittest

from transformers import DonutProcessor


__UpperCAmelCase            =       """naver-clova-ix/donut-base"""




class      UpperCamelCase__					(					unittest.TestCase			):




					"""simple docstring"""


					def lowerCamelCase_			(						self				:			List[str]						):

							'''simple docstring'''



							SCREAMING_SNAKE_CASE :				Any		=			DonutProcessor.from_pretrained(lowerCamelCase_						)



					def lowerCamelCase_			(						self				:			Tuple						):

							'''simple docstring'''



							SCREAMING_SNAKE_CASE :				Optional[int]		=			{
							    """name""": """John Doe""",
							    """age""": """99""",
							    """city""": """Atlanta""",
							    """state""": """GA""",
							    """zip""": """30301""",
							    """phone""": """123-4567""",
							    """nicknames""": [{"""nickname""": """Johnny"""}, {"""nickname""": """JD"""}],
							}

							SCREAMING_SNAKE_CASE :				List[Any]		=			(
							    """<s_name>John Doe</s_name><s_age>99</s_age><s_city>Atlanta</s_city>"""
							    """<s_state>GA</s_state><s_zip>30301</s_zip><s_phone>123-4567</s_phone>"""
							    """<s_nicknames><s_nickname>Johnny</s_nickname>"""
							    """<sep/><s_nickname>JD</s_nickname></s_nicknames>"""
							)
							SCREAMING_SNAKE_CASE :				List[Any]		=			self.processor.tokenajson(lowerCamelCase_						)

							self.assertDictEqual(lowerCamelCase_      ,						lowerCamelCase_						)



