
from __future__ import annotations

import inspect
import unittest
from typing import List, Tuple

from transformers import RegNetConfig
from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import cached_property, is_tf_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
							import tensorflow as tf

							from transformers import TF_REGNET_PRETRAINED_MODEL_ARCHIVE_LIST, TFRegNetForImageClassification, TFRegNetModel


if is_vision_available():
							from PIL import Image

							from transformers import AutoImageProcessor







class        lowercase	:




							def __init__(       self     ,A__     ,A__=3     ,A__=3_2     ,A__=3     ,A__=1_0     ,A__=[1_0, 2_0, 3_0, 4_0]     ,A__=[1, 1, 2, 1]     ,A__=True     ,A__=True     ,A__="relu"     ,A__=3     ,A__=None     ,):
												lowercase					=		parent
												lowercase					=		batch_size
												lowercase					=		image_size
												lowercase					=		num_channels
												lowercase					=		embeddings_size
												lowercase					=		hidden_sizes
												lowercase					=		depths
												lowercase					=		is_training
												lowercase					=		use_labels
												lowercase					=		hidden_act
												lowercase					=		num_labels
												lowercase					=		scope
												lowercase					=		len(A__)




							def 				A__   (       self):
												lowercase					=		floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size])

												lowercase					=		None
												if self.use_labels:
																	lowercase					=		ids_tensor([self.batch_size]     ,self.num_labels)

												lowercase					=		self.get_config()
												return config, pixel_values, labels




							def 				A__   (       self):
												return RegNetConfig(
												    num_channels=self.num_channels     ,embeddings_size=self.embeddings_size     ,hidden_sizes=self.hidden_sizes     ,depths=self.depths     ,hidden_act=self.hidden_act     ,num_labels=self.num_labels     ,)




							def 				A__   (       self     ,A__     ,A__     ,A__):
												lowercase					=		TFRegNetModel(config=A__)
												lowercase					=		model(A__     ,training=A__)
												# expected last hidden states: B, C, H // 32, W // 32
												self.parent.assertEqual(
												    result.last_hidden_state.shape     ,(self.batch_size, self.hidden_sizes[-1], self.image_size // 3_2, self.image_size // 3_2)     ,)




							def 				A__   (       self     ,A__     ,A__     ,A__):
												lowercase					=		self.num_labels
												lowercase					=		TFRegNetForImageClassification(A__)
												lowercase					=		model(A__     ,labels=A__     ,training=A__)
												self.parent.assertEqual(result.logits.shape     ,(self.batch_size, self.num_labels))





							def 				A__   (       self):
												lowercase					=		self.prepare_config_and_inputs()
												lowercase			,		lowercase			,		lowercase					=		config_and_inputs
												lowercase					=		{'''pixel_values''': pixel_values}
												return config, inputs_dict







@require_tf
class        lowercase	(      SCREAMING_SNAKE_CASE__ ,						SCREAMING_SNAKE_CASE__ ,						unittest.TestCase	):
							lowercase_ : Any       =(TFRegNetModel, TFRegNetForImageClassification) if is_tf_available() else ()
							lowercase_ : Optional[int]       =(
							    {'''feature-extraction''': TFRegNetModel, '''image-classification''': TFRegNetForImageClassification}
							    if is_tf_available()
							    else {}
							)

							lowercase_ : Any       =False
							lowercase_ : Optional[Any]       =False
							lowercase_ : Union[str, Any]       =False
							lowercase_ : Tuple       =False
							lowercase_ : Any       =False




							def 				A__   (       self):
												lowercase					=		TFRegNetModelTester(self)
												lowercase					=		ConfigTester(self     ,config_class=A__     ,has_text_modality=A__)




							def 				A__   (       self):
												return




							@unittest.skip(reason='''RegNet does not use inputs_embeds''')
							def 				A__   (       self):
												pass




							@unittest.skipIf(
							    not is_tf_available() or len(tf.config.list_physical_devices('''GPU''')) == 0     ,reason='''TF does not support backprop for grouped convolutions on CPU.'''     ,)
							@slow
							def 				A__   (       self):
												super().test_keras_fit()




							@unittest.skip(reason='''RegNet does not support input and output embeddings''')
							def 				A__   (       self):
												pass




							def 				A__   (       self):
												lowercase			,		lowercase					=		self.model_tester.prepare_config_and_inputs_for_common()

												for model_class in self.all_model_classes:
																	lowercase					=		model_class(A__)
																	lowercase					=		inspect.signature(model.call)
																	# signature.parameters is an OrderedDict => so arg_names order is deterministic
																	lowercase					=		[*signature.parameters.keys()]

																	lowercase					=		['''pixel_values''']
																	self.assertListEqual(arg_names[:1]     ,A__)




							def 				A__   (       self):
												lowercase					=		self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_model(*A__)




							def 				A__   (       self):
												def check_hidden_states_output(A__     ,A__     ,A__):
																	lowercase					=		model_class(A__)
																	lowercase					=		model(**self._prepare_for_class(A__     ,A__)     ,training=A__)

																	lowercase					=		outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

																	lowercase					=		self.model_tester.num_stages
																	self.assertEqual(len(A__)     ,expected_num_stages + 1)

																	# RegNet's feature maps are of shape (batch_size, num_channels, height, width)
																	self.assertListEqual(
																	    list(hidden_states[0].shape[-2:])     ,[self.model_tester.image_size // 2, self.model_tester.image_size // 2]     ,)

												lowercase			,		lowercase					=		self.model_tester.prepare_config_and_inputs_for_common()
												lowercase					=		['''basic''', '''bottleneck''']
												for model_class in self.all_model_classes:
																	for layer_type in layers_type:
																						lowercase					=		layer_type
																						lowercase					=		True
																						check_hidden_states_output(A__     ,A__     ,A__)

																						# check that output_hidden_states also work using config
																						del inputs_dict["output_hidden_states"]
																						lowercase					=		True

																						check_hidden_states_output(A__     ,A__     ,A__)




							def 				A__   (       self):
												lowercase			,		lowercase					=		self.model_tester.prepare_config_and_inputs_for_common()

												def check_equivalence(A__     ,A__     ,A__     ,A__={}):
																	lowercase					=		model(A__     ,return_dict=A__     ,**A__)
																	lowercase					=		model(A__     ,return_dict=A__     ,**A__).to_tuple()

																	def recursive_check(A__     ,A__):
																						if isinstance(A__     ,(List, Tuple)):
																											for tuple_iterable_value, dict_iterable_value in zip(A__     ,A__):
																																recursive_check(A__     ,A__)
																						elif tuple_object is None:
																											return
																						else:
																											self.assertTrue(
																											    all(tf.equal(A__     ,A__))     ,msg=(
																											        '''Tuple and dict output are not equal. Difference:'''
																											        f' {tf.math.reduce_max(tf.abs(tuple_object - dict_object))}'
																											    )     ,)

																						recursive_check(A__     ,A__)

												for model_class in self.all_model_classes:
																	lowercase					=		model_class(A__)

																	lowercase					=		self._prepare_for_class(A__     ,A__)
																	lowercase					=		self._prepare_for_class(A__     ,A__)
																	check_equivalence(A__     ,A__     ,A__)

																	lowercase					=		self._prepare_for_class(A__     ,A__     ,return_labels=A__)
																	lowercase					=		self._prepare_for_class(A__     ,A__     ,return_labels=A__)
																	check_equivalence(A__     ,A__     ,A__)

																	lowercase					=		self._prepare_for_class(A__     ,A__)
																	lowercase					=		self._prepare_for_class(A__     ,A__)
																	check_equivalence(A__     ,A__     ,A__     ,{'''output_hidden_states''': True})

																	lowercase					=		self._prepare_for_class(A__     ,A__     ,return_labels=A__)
																	lowercase					=		self._prepare_for_class(A__     ,A__     ,return_labels=A__)
																	check_equivalence(A__     ,A__     ,A__     ,{'''output_hidden_states''': True})




							def 				A__   (       self):
												lowercase					=		self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_for_image_classification(*A__)





							@slow
							def 				A__   (       self):
												for model_name in TF_REGNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																	lowercase					=		TFRegNetModel.from_pretrained(A__)
																	self.assertIsNotNone(A__)



def  UpperCamelCase					(							):
					'''simple docstring'''
					lowercase					=		Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''	)
					return image







@require_tf
@require_vision
class        lowercase	(      unittest.TestCase	):




							@cached_property
							def 				A__   (       self):
												return (
												    AutoImageProcessor.from_pretrained(TF_REGNET_PRETRAINED_MODEL_ARCHIVE_LIST[0])
												    if is_vision_available()
												    else None
												)





							@slow
							def 				A__   (       self):
												lowercase					=		TFRegNetForImageClassification.from_pretrained(TF_REGNET_PRETRAINED_MODEL_ARCHIVE_LIST[0])

												lowercase					=		self.default_image_processor
												lowercase					=		prepare_img()
												lowercase					=		image_processor(images=A__     ,return_tensors='''tf''')

												# forward pass
												lowercase					=		model(**A__     ,training=A__)

												# verify the logits
												lowercase					=		tf.TensorShape((1, 1_0_0_0))
												self.assertEqual(outputs.logits.shape     ,A__)

												lowercase					=		tf.constant([-0.4180, -1.5051, -3.4836])

												tf.debugging.assert_near(outputs.logits[0, :3]     ,A__     ,atol=1E-4)






# 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


lowercase__	:Any										=					"Run commands across TPU VMs for initial setup before running `accelerate launch`."



def  UpperCamelCase					(							lowerCAmelCase__=None	):
					'''simple docstring'''
					if subparsers is not None:
										lowercase					=		subparsers.add_parser('''tpu-config'''				,    description=_description	)
					else:
										lowercase					=		argparse.ArgumentParser('''Accelerate tpu-config command'''				,    description=_description	)
					# Core arguments
					lowercase					=		parser.add_argument_group(
					    '''Config Arguments'''				,    '''Arguments that can be configured through `accelerate config`.'''	)
					config_args.add_argument(
					    '''--config_file'''				,    type=lowerCAmelCase__				,    default=lowerCAmelCase__				,    help='''Path to the config file to use for accelerate.'''				,    )
					config_args.add_argument(
					    '''--tpu_name'''				,    default=lowerCAmelCase__				,    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=lowerCAmelCase__				,    help='''The zone of the TPU to use. If not specified, will use the zone specified in the config file.'''				,    )
					lowercase					=		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=lowerCAmelCase__				,    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=lowerCAmelCase__	)
					return parser



def  UpperCamelCase					(							lowerCAmelCase__	):
					'''simple docstring'''
					lowercase					=		None

					# Get the default from the config file if it exists.
					if args.config_file is not None or os.path.isfile(lowerCAmelCase__	):
										lowercase					=		load_config_from_file(args.config_file	)
										if not args.command_file and defaults.command_file is not None and not args.command:
															lowercase					=		defaults.command_file
										if not args.command and defaults.commands is not None:
															lowercase					=		defaults.commands
										if not args.tpu_name:
															lowercase					=		defaults.tpu_name
										if not args.tpu_zone:
															lowercase					=		defaults.tpu_zone
					if args.accelerate_version == "dev":
										lowercase					=		'''git+https://github.com/huggingface/accelerate.git'''
					elif args.accelerate_version == "latest":
										lowercase					=		'''accelerate -U'''
					elif isinstance(parse(args.accelerate_version	)				,    lowerCAmelCase__	):
										lowercase					=		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:
															lowercase					=		[f.read().splitlines()]

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

					# Then send it to gcloud
					# Eventually try to use google-api-core to do this instead of subprocess
					lowercase					=		['''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(lowerCAmelCase__	)}'	)
										return
					subprocess.run(lowerCAmelCase__	)
					print('''Successfully setup pod.'''	)



def  UpperCamelCase					(							):
					'''simple docstring'''
					lowercase					=		tpu_command_parser()
					lowercase					=		parser.parse_args()

					tpu_command_launcher(lowerCAmelCase__	)




