
import os
import tempfile
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
		import torch
		from torch import nn

		from transformers import (
		    Adafactor,
		    AdamW,
		    get_constant_schedule,
		    get_constant_schedule_with_warmup,
		    get_cosine_schedule_with_warmup,
		    get_cosine_with_hard_restarts_schedule_with_warmup,
		    get_inverse_sqrt_schedule,
		    get_linear_schedule_with_warmup,
		    get_polynomial_decay_schedule_with_warmup,
		)
def      UpperCAmelCase__				(							__magic_name__    :  List[Any]				,	__magic_name__    :  Optional[int]=10	):


					'''simple docstring'''

					lowerCAmelCase		:       Dict							=  []
					for _ in range(__magic_name__	):
										lrs.append(scheduler.get_lr()[0]	)
										scheduler.step()
					return lrs
def      UpperCAmelCase__				(							__magic_name__    :  Optional[int]				,	__magic_name__    :  List[Any]=10	):


					'''simple docstring'''

					lowerCAmelCase		:       Tuple							=  []
					for step in range(__magic_name__	):
										lrs.append(scheduler.get_lr()[0]	)
										scheduler.step()
										if step == num_steps // 2:
															with tempfile.TemporaryDirectory() as tmpdirname:
																				lowerCAmelCase		:       Tuple							=  os.path.join(__magic_name__				,	'''schedule.bin'''	)
																				torch.save(scheduler.state_dict()				,	__magic_name__	)

																				lowerCAmelCase		:       int							=  torch.load(__magic_name__	)
																				scheduler.load_state_dict(__magic_name__	)
					return lrs


@require_torch
class 		__magic_name__ (    unittest.TestCase     ):





		def    _A				(				self     :						Optional[Any]    ,	lowerCamelCase__     :						List[str]    ,	lowerCamelCase__     :						Tuple    ,	lowerCamelCase__     :						Dict					):
							self.assertEqual(len(lowerCAmelCase__					)    ,	len(lowerCAmelCase__					)					)
							for a, b in zip(lowerCAmelCase__    ,	lowerCAmelCase__					):
												self.assertAlmostEqual(lowerCAmelCase__    ,	lowerCAmelCase__    ,	delta=lowerCAmelCase__					)





		def    _A				(				self     :						Optional[int]					):
							lowerCAmelCase		:       List[Any]							=  torch.tensor([0.1, -0.2, -0.1]    ,	requires_grad=lowerCAmelCase__					)
							lowerCAmelCase		:       List[Any]							=  torch.tensor([0.4, 0.2, -0.5]					)
							lowerCAmelCase		:       List[Any]							=  nn.MSELoss()
							# No warmup, constant schedule, no gradient clipping
							lowerCAmelCase		:       List[str]							=  AdamW(params=[w]    ,	lr=2E-1    ,	weight_decay=0.0					)
							for _ in range(1_0_0					):
												lowerCAmelCase		:       Dict							=  criterion(lowerCAmelCase__    ,	lowerCAmelCase__					)
												loss.backward()
												optimizer.step()
												w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
												w.grad.zero_()
							self.assertListAlmostEqual(w.tolist()    ,	[0.4, 0.2, -0.5]    ,	tol=1E-2					)





		def    _A				(				self     :						int					):
							lowerCAmelCase		:       Tuple							=  torch.tensor([0.1, -0.2, -0.1]    ,	requires_grad=lowerCAmelCase__					)
							lowerCAmelCase		:       List[Any]							=  torch.tensor([0.4, 0.2, -0.5]					)
							lowerCAmelCase		:       int							=  nn.MSELoss()
							# No warmup, constant schedule, no gradient clipping
							lowerCAmelCase		:       Dict							=  Adafactor(
							    params=[w]    ,	lr=1E-2    ,	eps=(1E-30, 1E-3)    ,	clip_threshold=1.0    ,	decay_rate=-0.8    ,	betaa=lowerCAmelCase__    ,	weight_decay=0.0    ,	relative_step=lowerCAmelCase__    ,	scale_parameter=lowerCAmelCase__    ,	warmup_init=lowerCAmelCase__    ,	)
							for _ in range(1_0_0_0					):
												lowerCAmelCase		:       Tuple							=  criterion(lowerCAmelCase__    ,	lowerCAmelCase__					)
												loss.backward()
												optimizer.step()
												w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
												w.grad.zero_()
							self.assertListAlmostEqual(w.tolist()    ,	[0.4, 0.2, -0.5]    ,	tol=1E-2					)



@require_torch
class 		__magic_name__ (    unittest.TestCase     ):
		_lowerCAmelCase		=						nn.Linear(50,					50     ) if is_torch_available() else None
		_lowerCAmelCase		=						AdamW(m.parameters(),					lr=10.0     ) if is_torch_available() else None
		_lowerCAmelCase		=						10





		def    _A				(				self     :						Any    ,	lowerCamelCase__     :						Optional[Any]    ,	lowerCamelCase__     :						Optional[Any]    ,	lowerCamelCase__     :						Any    ,	lowerCamelCase__     :						Any=None					):
							self.assertEqual(len(lowerCAmelCase__					)    ,	len(lowerCAmelCase__					)					)
							for a, b in zip(lowerCAmelCase__    ,	lowerCAmelCase__					):
												self.assertAlmostEqual(lowerCAmelCase__    ,	lowerCAmelCase__    ,	delta=lowerCAmelCase__    ,	msg=lowerCAmelCase__					)





		def    _A				(				self     :						Any					):
							lowerCAmelCase		:       Dict							=  {'''num_warmup_steps''': 2, '''num_training_steps''': 1_0}
							# schedulers doct format
							# function: (sched_args_dict, expected_learning_rates)
							lowerCAmelCase		:       Tuple							=  {
							    get_constant_schedule: ({}, [1_0.0] * self.num_steps),
							    get_constant_schedule_with_warmup: (
							        {'''num_warmup_steps''': 4},
							        [0.0, 2.5, 5.0, 7.5, 1_0.0, 1_0.0, 1_0.0, 1_0.0, 1_0.0, 1_0.0],
							    ),
							    get_linear_schedule_with_warmup: (
							        {**common_kwargs},
							        [0.0, 5.0, 1_0.0, 8.7_5, 7.5, 6.2_5, 5.0, 3.7_5, 2.5, 1.2_5],
							    ),
							    get_cosine_schedule_with_warmup: (
							        {**common_kwargs},
							        [0.0, 5.0, 1_0.0, 9.6_1, 8.5_3, 6.9_1, 5.0, 3.0_8, 1.4_6, 0.3_8],
							    ),
							    get_cosine_with_hard_restarts_schedule_with_warmup: (
							        {**common_kwargs, '''num_cycles''': 2},
							        [0.0, 5.0, 1_0.0, 8.5_3, 5.0, 1.4_6, 1_0.0, 8.5_3, 5.0, 1.4_6],
							    ),
							    get_polynomial_decay_schedule_with_warmup: (
							        {**common_kwargs, '''power''': 2.0, '''lr_end''': 1E-7},
							        [0.0, 5.0, 1_0.0, 7.6_5_6, 5.6_2_5, 3.9_0_6, 2.5, 1.4_0_6, 0.6_2_5, 0.1_5_6],
							    ),
							    get_inverse_sqrt_schedule: (
							        {'''num_warmup_steps''': 2},
							        [0.0, 5.0, 1_0.0, 8.1_6_5, 7.0_7_1, 6.3_2_5, 5.7_7_4, 5.3_4_5, 5.0, 4.7_1_4],
							    ),
							}

							for scheduler_func, data in scheds.items():
												lowerCAmelCase						,				lowerCAmelCase		:       List[Any]							=  data

												lowerCAmelCase		:       List[str]							=  scheduler_func(self.optimizer    ,	**lowerCAmelCase__					)
												self.assertEqual(len([scheduler.get_lr()[0]]					)    ,	1					)
												lowerCAmelCase		:       Optional[int]							=  unwrap_schedule(lowerCAmelCase__    ,	self.num_steps					)
												self.assertListAlmostEqual(
												    lowerCAmelCase__    ,	lowerCAmelCase__    ,	tol=1E-2    ,	msg=f'''failed for {scheduler_func} in normal scheduler'''    ,	)

												lowerCAmelCase		:       str							=  scheduler_func(self.optimizer    ,	**lowerCAmelCase__					)
												if scheduler_func.__name__ != "get_constant_schedule":
																	LambdaScheduleWrapper.wrap_scheduler(lowerCAmelCase__					)  # wrap to test picklability of the schedule
												lowerCAmelCase		:       Tuple							=  unwrap_and_save_reload_schedule(lowerCAmelCase__    ,	self.num_steps					)
												self.assertListEqual(lowerCAmelCase__    ,	lowerCAmelCase__    ,	msg=f'''failed for {scheduler_func} in save and reload'''					)



class 		__magic_name__ :





		def __init__(				self     :						List[str]    ,	lowerCamelCase__     :						int					):
							lowerCAmelCase		:       int							=  fn





		def __call__(				self     :						Optional[Any]    ,	*lowerCamelCase__     :						int    ,	**lowerCamelCase__     :						List[str]					):
							return self.fn(*lowerCAmelCase__    ,	**lowerCAmelCase__					)





		@classmethod
		def    _A				(				self     :						List[Any]    ,	lowerCamelCase__     :						Optional[Any]					):
							lowerCAmelCase		:       str							=  list(map(self    ,	scheduler.lr_lambdas					)					)


import inspect
import unittest

from transformers import MobileNetVaConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
						import torch

						from transformers import MobileNetVaForImageClassification, MobileNetVaForSemanticSegmentation, MobileNetVaModel
						from transformers.models.mobilenet_va.modeling_mobilenet_va import MOBILENET_V2_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
						from PIL import Image

						from transformers import MobileNetVaImageProcessor





class   UpperCamelCase_      (      __a		):







						'''simple docstring'''



						def        SCREAMING_SNAKE_CASE(			self							:Tuple     )  ->Optional[int]:
										lowercase         =      self.config_class(**self.inputs_dict     )
										self.parent.assertTrue(hasattr(lowerCAmelCase__		,    "tf_padding"     )     )
										self.parent.assertTrue(hasattr(lowerCAmelCase__		,    "depth_multiplier"     )     )





class   UpperCamelCase_      :







						'''simple docstring'''



						def __init__(			self							:Dict		,    lowerCAmelCase__							:Any		,    lowerCAmelCase__							:Union[str, Any]=13		,    lowerCAmelCase__							:Tuple=3		,    lowerCAmelCase__							:Optional[int]=32		,    lowerCAmelCase__							:Any=0.25		,    lowerCAmelCase__							:Dict=8		,    lowerCAmelCase__							:Optional[int]=8		,    lowerCAmelCase__							:List[str]=6		,    lowerCAmelCase__							:List[Any]=32		,    lowerCAmelCase__							:Optional[Any]=True		,    lowerCAmelCase__							:str=True		,    lowerCAmelCase__							:Optional[int]=True		,    lowerCAmelCase__							:Dict="relu6"		,    lowerCAmelCase__							:Tuple=1280		,    lowerCAmelCase__							:Dict=0.1		,    lowerCAmelCase__							:List[str]=0.02		,    lowerCAmelCase__							:Dict=True		,    lowerCAmelCase__							:List[Any]=True		,    lowerCAmelCase__							:Tuple=10		,    lowerCAmelCase__							:int=None		,    )  ->List[str]:
										lowercase         =      parent
										lowercase         =      batch_size
										lowercase         =      num_channels
										lowercase         =      image_size
										lowercase         =      depth_multiplier
										lowercase         =      depth_divisible_by
										lowercase         =      min_depth
										lowercase         =      expand_ratio
										lowercase         =      tf_padding
										lowercase         =      output_stride
										lowercase         =      first_layer_is_expansion
										lowercase         =      finegrained_output
										lowercase         =      hidden_act
										lowercase         =      last_hidden_size if finegrained_output else int(last_hidden_size * depth_multiplier     )
										lowercase         =      classifier_dropout_prob
										lowercase         =      use_labels
										lowercase         =      is_training
										lowercase         =      num_labels
										lowercase         =      initializer_range
										lowercase         =      scope





						def        SCREAMING_SNAKE_CASE(			self							:str     )  ->Dict:
										lowercase         =      floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]     )

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

										lowercase         =      self.get_config()

										return config, pixel_values, labels, pixel_labels





						def        SCREAMING_SNAKE_CASE(			self							:Optional[int]     )  ->List[Any]:
										return MobileNetVaConfig(
										    num_channels=self.num_channels		,    image_size=self.image_size		,    depth_multiplier=self.depth_multiplier		,    depth_divisible_by=self.depth_divisible_by		,    min_depth=self.min_depth		,    expand_ratio=self.expand_ratio		,    output_stride=self.output_stride		,    first_layer_is_expansion=self.first_layer_is_expansion		,    finegrained_output=self.finegrained_output		,    hidden_act=self.hidden_act		,    tf_padding=self.tf_padding		,    classifier_dropout_prob=self.classifier_dropout_prob		,    initializer_range=self.initializer_range		,    )





						def        SCREAMING_SNAKE_CASE(			self							:Dict		,    lowerCAmelCase__							:Dict		,    lowerCAmelCase__							:Optional[Any]		,    lowerCAmelCase__							:Optional[Any]		,    lowerCAmelCase__							:Tuple     )  ->Any:
										lowercase         =      MobileNetVaModel(config=lowerCAmelCase__     )
										model.to(lowerCAmelCase__     )
										model.eval()
										lowercase         =      model(lowerCAmelCase__     )
										self.parent.assertEqual(
										    result.last_hidden_state.shape		,    (
										        self.batch_size,
										        self.last_hidden_size,
										        self.image_size // self.output_stride,
										        self.image_size // self.output_stride,
										    )		,    )
										self.parent.assertEqual(
										    result.pooler_output.shape		,    (self.batch_size, self.last_hidden_size)		,    )





						def        SCREAMING_SNAKE_CASE(			self							:Tuple		,    lowerCAmelCase__							:int		,    lowerCAmelCase__							:List[str]		,    lowerCAmelCase__							:str		,    lowerCAmelCase__							:Optional[Any]     )  ->Union[str, Any]:
										lowercase         =      self.num_labels
										lowercase         =      MobileNetVaForImageClassification(lowerCAmelCase__     )
										model.to(lowerCAmelCase__     )
										model.eval()
										lowercase         =      model(lowerCAmelCase__		,    labels=lowerCAmelCase__     )
										self.parent.assertEqual(result.logits.shape		,    (self.batch_size, self.num_labels)     )





						def        SCREAMING_SNAKE_CASE(			self							:Tuple		,    lowerCAmelCase__							:List[str]		,    lowerCAmelCase__							:Any		,    lowerCAmelCase__							:Union[str, Any]		,    lowerCAmelCase__							:Optional[int]     )  ->Union[str, Any]:
										lowercase         =      self.num_labels
										lowercase         =      MobileNetVaForSemanticSegmentation(lowerCAmelCase__     )
										model.to(lowerCAmelCase__     )
										model.eval()
										lowercase         =      model(lowerCAmelCase__     )
										self.parent.assertEqual(
										    result.logits.shape		,    (
										        self.batch_size,
										        self.num_labels,
										        self.image_size // self.output_stride,
										        self.image_size // self.output_stride,
										    )		,    )
										lowercase         =      model(lowerCAmelCase__		,    labels=lowerCAmelCase__     )
										self.parent.assertEqual(
										    result.logits.shape		,    (
										        self.batch_size,
										        self.num_labels,
										        self.image_size // self.output_stride,
										        self.image_size // self.output_stride,
										    )		,    )





						def        SCREAMING_SNAKE_CASE(			self							:Any     )  ->int:
										lowercase         =      self.prepare_config_and_inputs()
										lowercase     ,	lowercase     ,	lowercase     ,	lowercase         =      config_and_inputs
										lowercase         =      {"pixel_values": pixel_values}
										return config, inputs_dict





@require_torch
class   UpperCamelCase_      (      __a		,    __a		,    unittest.TestCase		):







						'''simple docstring'''



						UpperCamelCase      :  Optional[int]				   =					(
						    (MobileNetVaModel, MobileNetVaForImageClassification, MobileNetVaForSemanticSegmentation)
						    if is_torch_available()
						    else ()
						)
						UpperCamelCase      :  List[Any]				   =					(
						    {
						        '''feature-extraction''': MobileNetVaModel,
						        '''image-classification''': MobileNetVaForImageClassification,
						        '''image-segmentation''': MobileNetVaForSemanticSegmentation,
						    }
						    if is_torch_available()
						    else {}
						)

						UpperCamelCase      :  Union[str, Any]				   =					False
						UpperCamelCase      :  int				   =					False
						UpperCamelCase      :  str				   =					False
						UpperCamelCase      :  Dict				   =					False





						def        SCREAMING_SNAKE_CASE(			self							:Optional[int]     )  ->Dict:
										lowercase         =      MobileNetVaModelTester(self     )
										lowercase         =      MobileNetVaConfigTester(self		,    config_class=lowerCAmelCase__		,    has_text_modality=lowerCAmelCase__     )





						def        SCREAMING_SNAKE_CASE(			self							:Dict     )  ->Any:
										self.config_tester.run_common_tests()





						@unittest.skip(reason="MobileNetV2 does not use inputs_embeds"     )
						def        SCREAMING_SNAKE_CASE(			self							:Tuple     )  ->Optional[int]:
										pass





						@unittest.skip(reason="MobileNetV2 does not support input and output embeddings"     )
						def        SCREAMING_SNAKE_CASE(			self							:Any     )  ->str:
										pass





						@unittest.skip(reason="MobileNetV2 does not output attentions"     )
						def        SCREAMING_SNAKE_CASE(			self							:List[Any]     )  ->str:
										pass





						def        SCREAMING_SNAKE_CASE(			self							:Union[str, Any]     )  ->Dict:
										lowercase     ,	lowercase         =      self.model_tester.prepare_config_and_inputs_for_common()

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

														lowercase         =      ["pixel_values"]
														self.assertListEqual(arg_names[:1]		,    lowerCAmelCase__     )





						def        SCREAMING_SNAKE_CASE(			self							:int     )  ->List[str]:
										lowercase         =      self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_model(*lowerCAmelCase__     )





						def        SCREAMING_SNAKE_CASE(			self							:int     )  ->Optional[int]:
										def check_hidden_states_output(lowerCAmelCase__							:str		,    lowerCAmelCase__							:Union[str, Any]		,    lowerCAmelCase__							:int     ):
														lowercase         =      model_class(lowerCAmelCase__     )
														model.to(lowerCAmelCase__     )
														model.eval()

														with torch.no_grad():
																		lowercase         =      model(**self._prepare_for_class(lowerCAmelCase__		,    lowerCAmelCase__     )     )

														lowercase         =      outputs.hidden_states

														lowercase         =      16
														self.assertEqual(len(lowerCAmelCase__     )		,    lowerCAmelCase__     )

										lowercase     ,	lowercase         =      self.model_tester.prepare_config_and_inputs_for_common()

										for model_class in self.all_model_classes:
														lowercase         =      True
														check_hidden_states_output(lowerCAmelCase__		,    lowerCAmelCase__		,    lowerCAmelCase__     )

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

														check_hidden_states_output(lowerCAmelCase__		,    lowerCAmelCase__		,    lowerCAmelCase__     )





						def        SCREAMING_SNAKE_CASE(			self							:Any     )  ->Tuple:
										lowercase         =      self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_for_image_classification(*lowerCAmelCase__     )





						def        SCREAMING_SNAKE_CASE(			self							:int     )  ->str:
										lowercase         =      self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_for_semantic_segmentation(*lowerCAmelCase__     )





						@slow
						def        SCREAMING_SNAKE_CASE(			self							:Union[str, Any]     )  ->str:
										for model_name in MOBILENET_V2_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														lowercase         =      MobileNetVaModel.from_pretrained(lowerCAmelCase__     )
														self.assertIsNotNone(lowerCAmelCase__     )




def  __snake_case							(	):







				'''simple docstring'''


				lowercase         =      Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"						)
				return image





@require_torch
@require_vision
class   UpperCamelCase_      (      unittest.TestCase		):







						'''simple docstring'''



						@cached_property
						def        SCREAMING_SNAKE_CASE(			self							:str     )  ->Any:
										return (
										    MobileNetVaImageProcessor.from_pretrained("google/mobilenet_v2_1.0_224"     ) if is_vision_available() else None
										)





						@slow
						def        SCREAMING_SNAKE_CASE(			self							:Tuple     )  ->Union[str, Any]:
										lowercase         =      MobileNetVaForImageClassification.from_pretrained("google/mobilenet_v2_1.0_224"     ).to(lowerCAmelCase__     )

										lowercase         =      self.default_image_processor
										lowercase         =      prepare_img()
										lowercase         =      image_processor(images=lowerCAmelCase__		,    return_tensors="pt"     ).to(lowerCAmelCase__     )

										# forward pass
										with torch.no_grad():
														lowercase         =      model(**lowerCAmelCase__     )

										# verify the logits
										lowercase         =      torch.Size((1, 1001)     )
										self.assertEqual(outputs.logits.shape		,    lowerCAmelCase__     )

										lowercase         =      torch.tensor([0.24_45, -1.19_93, 0.19_05]     ).to(lowerCAmelCase__     )

										self.assertTrue(torch.allclose(outputs.logits[0, :3]		,    lowerCAmelCase__		,    atol=1E-4     )     )





						@slow
						def        SCREAMING_SNAKE_CASE(			self							:Tuple     )  ->List[Any]:
										lowercase         =      MobileNetVaForSemanticSegmentation.from_pretrained("google/deeplabv3_mobilenet_v2_1.0_513"     )
										lowercase         =      model.to(lowerCAmelCase__     )

										lowercase         =      MobileNetVaImageProcessor.from_pretrained("google/deeplabv3_mobilenet_v2_1.0_513"     )

										lowercase         =      prepare_img()
										lowercase         =      image_processor(images=lowerCAmelCase__		,    return_tensors="pt"     ).to(lowerCAmelCase__     )

										# forward pass
										with torch.no_grad():
														lowercase         =      model(**lowerCAmelCase__     )
										lowercase         =      outputs.logits

										# verify the logits
										lowercase         =      torch.Size((1, 21, 65, 65)     )
										self.assertEqual(logits.shape		,    lowerCAmelCase__     )

										lowercase         =      torch.tensor(
										    [
										        [[17.57_90, 17.75_81, 18.33_55], [18.32_57, 18.42_30, 18.89_73], [18.61_69, 18.86_50, 19.21_87]],
										        [[-2.15_95, -2.09_77, -2.37_41], [-2.42_26, -2.30_28, -2.68_35], [-2.78_19, -2.59_91, -2.77_06]],
										        [[4.20_58, 4.83_17, 4.76_38], [4.41_36, 5.03_61, 4.93_83], [4.50_28, 4.96_44, 4.87_34]],
										    ]		,    device=lowerCAmelCase__		,    )

										self.assertTrue(torch.allclose(logits[0, :3, :3, :3]		,    lowerCAmelCase__		,    atol=1E-4     )     )
