

# Copyright 2022 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import os
import subprocess

from packaging.version import Version, parse

from accelerate.commands.config.config_args import default_config_file, load_config_from_file


A_ :				List[Any]						 =    'Run commands across TPU VMs for initial setup before running `accelerate launch`.'

def 		UpperCamelCase      (lowercase_:				int=None      )  ->   Any:
	if subparsers is not None:
		A__							:					int       =	subparsers.add_parser("""tpu-config"""					,   description=_description      )
	else:
		A__							:					Any       =	argparse.ArgumentParser("""Accelerate tpu-config command"""					,   description=_description      )
	# Core arguments
	A__							:					Union[str, Any]       =	parser.add_argument_group(
	    """Config Arguments"""					,   """Arguments that can be configured through `accelerate config`."""      )
	config_args.add_argument(
	    """--config_file"""					,   type=lowercase_					,   default=lowercase_					,   help="""Path to the config file to use for accelerate."""					,   )
	config_args.add_argument(
	    """--tpu_name"""					,   default=lowercase_					,   help="""The name of the TPU to use. If not specified, will use the TPU specified in the config file."""					,   )
	config_args.add_argument(
	    """--tpu_zone"""					,   default=lowercase_					,   help="""The zone of the TPU to use. If not specified, will use the zone specified in the config file."""					,   )
	A__							:					int       =	parser.add_argument_group("""TPU Arguments"""					,   """Arguments for options ran inside the TPU."""      )
	pod_args.add_argument(
	    """--use_alpha"""					,   action="""store_true"""					,   help="""Whether to use `gcloud alpha` when running the TPU training script instead of `gcloud`."""					,   )
	pod_args.add_argument(
	    """--command_file"""					,   default=lowercase_					,   help="""The path to the file containing the commands to run on the pod on startup."""					,   )
	pod_args.add_argument(
	    """--command"""					,   action="""append"""					,   nargs="""+"""					,   help="""A command to run on the pod. Can be passed multiple times."""					,   )
	pod_args.add_argument(
	    """--install_accelerate"""					,   action="""store_true"""					,   help="""Whether to install accelerate on the pod. Defaults to False."""					,   )
	pod_args.add_argument(
	    """--accelerate_version"""					,   default="""latest"""					,   help="""The version of accelerate to install on the pod. If not specified, will use the latest pypi version. Specify 'dev' to install from GitHub."""					,   )
	pod_args.add_argument(
	    """--debug"""					,   action="""store_true"""					,   help="""If set, will print the command that would be run instead of running it."""      )

	if subparsers is not None:
		parser.set_defaults(func=lowercase_      )
	return parser

def 		UpperCamelCase      (lowercase_:				int      )  ->   int:
	A__							:					List[str]       =	None

	# Get the default from the config file if it exists.
	if args.config_file is not None or os.path.isfile(lowercase_      ):
		A__							:					str       =	load_config_from_file(args.config_file      )
		if not args.command_file and defaults.command_file is not None and not args.command:
			A__							:					List[str]       =	defaults.command_file
		if not args.command and defaults.commands is not None:
			A__							:					Tuple       =	defaults.commands
		if not args.tpu_name:
			A__							:					int       =	defaults.tpu_name
		if not args.tpu_zone:
			A__							:					Optional[int]       =	defaults.tpu_zone
	if args.accelerate_version == "dev":
		A__							:					List[Any]       =	"""git+https://github.com/huggingface/accelerate.git"""
	elif args.accelerate_version == "latest":
		A__							:					int       =	"""accelerate -U"""
	elif isinstance(parse(args.accelerate_version      )					,   lowercase_      ):
		A__							:					int       =	f"""accelerate=={args.accelerate_version}"""

	if not args.command_file and not args.command:
		raise ValueError("""You must specify either a command file or a command to run on the pod."""      )

	if args.command_file:
		with open(args.command_file					,   """r"""      ) as f:
			A__							:					List[str]       =	[f.read().splitlines()]

    # To turn list of lists into list of strings
	if isinstance(args.command[0]					,   lowercase_      ):
		A__							:					List[str]       =	[line for cmd in args.command for line in cmd]
	# Default to the shared folder and install accelerate
	A__							:					List[Any]       =	["""cd /usr/share"""]
	if args.install_accelerate:
		new_cmd += [f"""pip install {args.accelerate_version}"""]
	new_cmd += args.command
	A__							:					str       =	"""; """.join(lowercase_      )

	# Then send it to gcloud
	# Eventually try to use google-api-core to do this instead of subprocess
	A__							:					Union[str, Any]       =	["""gcloud"""]
	if args.use_alpha:
		cmd += ["alpha"]
	cmd += [
	    "compute",
	    "tpus",
	    "tpu-vm",
	    "ssh",
	    args.tpu_name,
	    "--zone",
	    args.tpu_zone,
	    "--command",
	    args.command,
	    "--worker",
	    "all",
	]
	if args.debug:
		print(f"""Running {' '.join(lowercase_      )}"""      )
		return
	subprocess.run(lowercase_      )
	print("""Successfully setup pod."""      )


def 		UpperCamelCase      ()  ->   Optional[Any]:
	A__							:					str       =	tpu_command_parser()
	A__							:					Dict       =	parser.parse_args()

	tpu_command_launcher(lowercase_      )






import os
import tempfile
import unittest
from pathlib import Path

from transformers import AutoConfig, is_torch_available
from transformers.testing_utils import require_torch, torch_device


if is_torch_available():
	from transformers import PyTorchBenchmark, PyTorchBenchmarkArguments



@require_torch
class 					_a		(unittest.TestCase       ):


					'''simple docstring'''






					def 						__A    (				self					, A__    ):
						for model_result in results.values():
							for batch_size, sequence_length in zip(model_result["""bs"""]					, model_result["""ss"""]    ):
								A__							:					str       =	model_result["""result"""][batch_size][sequence_length]
								self.assertIsNotNone(A__    )






					def 						__A    (				self    ):
						A__							:					Dict       =	"""sshleifer/tiny-gpt2"""
						A__							:					Tuple       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, multi_process=A__					, )
						A__							:					int       =	PyTorchBenchmark(A__    )
						A__							:					List[Any]       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_inference_result    )
						self.check_results_dict_not_empty(results.memory_inference_result    )






					def 						__A    (				self    ):
						A__							:					Dict       =	"""sgugger/tiny-distilbert-classification"""
						A__							:					Dict       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, multi_process=A__					, only_pretrain_model=A__					, )
						A__							:					str       =	PyTorchBenchmark(A__    )
						A__							:					List[str]       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_inference_result    )
						self.check_results_dict_not_empty(results.memory_inference_result    )






					def 						__A    (				self    ):
						A__							:					Any       =	"""sshleifer/tiny-gpt2"""
						A__							:					List[Any]       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, torchscript=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, multi_process=A__					, )
						A__							:					Tuple       =	PyTorchBenchmark(A__    )
						A__							:					str       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_inference_result    )
						self.check_results_dict_not_empty(results.memory_inference_result    )






					@unittest.skipIf(torch_device == """cpu"""					, """Cant do half precision"""    )
					def 						__A    (				self    ):
						A__							:					Optional[Any]       =	"""sshleifer/tiny-gpt2"""
						A__							:					Optional[Any]       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, fpaa=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, multi_process=A__					, )
						A__							:					str       =	PyTorchBenchmark(A__    )
						A__							:					Any       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_inference_result    )
						self.check_results_dict_not_empty(results.memory_inference_result    )






					def 						__A    (				self    ):
						A__							:					Optional[Any]       =	"""sshleifer/tiny-gpt2"""
						A__							:					Tuple       =	AutoConfig.from_pretrained(A__    )
						# set architectures equal to `None`
						A__							:					List[Any]       =	None
						A__							:					str       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, multi_process=A__					, )
						A__							:					List[str]       =	PyTorchBenchmark(A__					, configs=[config]    )
						A__							:					Any       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_inference_result    )
						self.check_results_dict_not_empty(results.memory_inference_result    )






					def 						__A    (				self    ):
						A__							:					Optional[int]       =	"""sshleifer/tiny-gpt2"""
						A__							:					Optional[int]       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, multi_process=A__					, )
						A__							:					Any       =	PyTorchBenchmark(A__    )
						A__							:					Dict       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_train_result    )
						self.check_results_dict_not_empty(results.memory_train_result    )






					@unittest.skipIf(torch_device == """cpu"""					, """Can't do half precision"""    )
					def 						__A    (				self    ):
						A__							:					Optional[int]       =	"""sshleifer/tiny-gpt2"""
						A__							:					List[str]       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, fpaa=A__					, multi_process=A__					, )
						A__							:					Dict       =	PyTorchBenchmark(A__    )
						A__							:					Optional[int]       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_train_result    )
						self.check_results_dict_not_empty(results.memory_train_result    )






					def 						__A    (				self    ):
						A__							:					int       =	"""sshleifer/tiny-gpt2"""
						A__							:					Optional[int]       =	AutoConfig.from_pretrained(A__    )
						A__							:					str       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, multi_process=A__					, )
						A__							:					int       =	PyTorchBenchmark(A__					, configs=[config]    )
						A__							:					Tuple       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_inference_result    )
						self.check_results_dict_not_empty(results.memory_inference_result    )






					def 						__A    (				self    ):
						A__							:					List[str]       =	"""sshleifer/tinier_bart"""
						A__							:					List[str]       =	AutoConfig.from_pretrained(A__    )
						A__							:					List[str]       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, multi_process=A__					, )
						A__							:					Union[str, Any]       =	PyTorchBenchmark(A__					, configs=[config]    )
						A__							:					str       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_inference_result    )
						self.check_results_dict_not_empty(results.memory_inference_result    )






					def 						__A    (				self    ):
						A__							:					Optional[int]       =	"""sshleifer/tiny-gpt2"""
						A__							:					Union[str, Any]       =	AutoConfig.from_pretrained(A__    )
						A__							:					Tuple       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, multi_process=A__					, )
						A__							:					int       =	PyTorchBenchmark(A__					, configs=[config]    )
						A__							:					Tuple       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_train_result    )
						self.check_results_dict_not_empty(results.memory_train_result    )






					def 						__A    (				self    ):
						A__							:					Dict       =	"""sshleifer/tinier_bart"""
						A__							:					int       =	AutoConfig.from_pretrained(A__    )
						A__							:					Union[str, Any]       =	PyTorchBenchmarkArguments(
						    models=[MODEL_ID]					, training=A__					, inference=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, multi_process=A__					, )
						A__							:					List[Any]       =	PyTorchBenchmark(A__					, configs=[config]    )
						A__							:					Optional[Any]       =	benchmark.run()
						self.check_results_dict_not_empty(results.time_train_result    )
						self.check_results_dict_not_empty(results.memory_train_result    )






					def 						__A    (				self    ):
						A__							:					int       =	"""sshleifer/tiny-gpt2"""
						with tempfile.TemporaryDirectory() as tmp_dir:
							A__							:					Dict       =	PyTorchBenchmarkArguments(
							    models=[MODEL_ID]					, training=A__					, inference=A__					, save_to_csv=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, inference_time_csv_file=os.path.join(A__					, """inf_time.csv"""    )					, train_memory_csv_file=os.path.join(A__					, """train_mem.csv"""    )					, inference_memory_csv_file=os.path.join(A__					, """inf_mem.csv"""    )					, train_time_csv_file=os.path.join(A__					, """train_time.csv"""    )					, env_info_csv_file=os.path.join(A__					, """env.csv"""    )					, multi_process=A__					, )
							A__							:					Optional[Any]       =	PyTorchBenchmark(A__    )
							benchmark.run()
							self.assertTrue(Path(os.path.join(A__					, """inf_time.csv"""    )    ).exists()    )
							self.assertTrue(Path(os.path.join(A__					, """train_time.csv"""    )    ).exists()    )
							self.assertTrue(Path(os.path.join(A__					, """inf_mem.csv"""    )    ).exists()    )
							self.assertTrue(Path(os.path.join(A__					, """train_mem.csv"""    )    ).exists()    )
							self.assertTrue(Path(os.path.join(A__					, """env.csv"""    )    ).exists()    )






					def 						__A    (				self    ):
						A__							:					Optional[int]       =	"""sshleifer/tiny-gpt2"""

						def _check_summary_is_not_empty(A__    ):
							self.assertTrue(hasattr(A__					, """sequential"""    )    )
							self.assertTrue(hasattr(A__					, """cumulative"""    )    )
							self.assertTrue(hasattr(A__					, """current"""    )    )
							self.assertTrue(hasattr(A__					, """total"""    )    )

						with tempfile.TemporaryDirectory() as tmp_dir:
							A__							:					Dict       =	PyTorchBenchmarkArguments(
							    models=[MODEL_ID]					, training=A__					, inference=A__					, sequence_lengths=[8]					, batch_sizes=[1]					, log_filename=os.path.join(A__					, """log.txt"""    )					, log_print=A__					, trace_memory_line_by_line=A__					, multi_process=A__					, )
							A__							:					Dict       =	PyTorchBenchmark(A__    )
							A__							:					str       =	benchmark.run()
							_check_summary_is_not_empty(result.inference_summary    )
							_check_summary_is_not_empty(result.train_summary    )
							self.assertTrue(Path(os.path.join(A__					, """log.txt"""    )    ).exists()    )



