


"""simple docstring"""


import argparse
import logging
import os
import sys
import tempfile
from pathlib import Path

import lightning_base
import pytest
import pytorch_lightning as pl
import torch
from convert_pl_checkpoint_to_hf import convert_pl_to_hf
from distillation import distill_main
from finetune import SummarizationModule, main
from huggingface_hub import list_models
from parameterized import parameterized
from run_eval import generate_summaries_or_translations
from torch import nn

from transformers import AutoConfig, AutoModelForSeqaSeqLM
from transformers.testing_utils import CaptureStderr, CaptureStdout, TestCasePlus, require_torch_gpu, slow
from utils import label_smoothed_nll_loss, lmap, load_json


logging.basicConfig(level=logging.DEBUG)

a_											=			logging.getLogger()
a_											=			torch.cuda.is_available()
a_											=			{
    "max_tokens_per_batch": None,
    "supervise_forward": True,
    "normalize_hidden": True,
    "label_smoothing": 0.2,
    "eval_max_gen_length": None,
    "eval_beams": 1,
    "val_metric": "loss",
    "save_top_k": 1,
    "adafactor": True,
    "early_stopping_patience": 2,
    "logger_name": "default",
    "length_penalty": 0.5,
    "cache_dir": "",
    "task": "summarization",
    "num_workers": 2,
    "alpha_hid": 0,
    "freeze_embeds": True,
    "enc_only": False,
    "tgt_suffix": "",
    "resume_from_checkpoint": None,
    "sortish_sampler": True,
    "student_decoder_layers": 1,
    "val_check_interval": 1.0,
    "output_dir": "",
    "fp16": False,  # TODO(SS): set this to CUDA_AVAILABLE if ci installs apex or start using native amp
    "no_teacher": False,
    "fp16_opt_level": "O1",
    "gpus": 1 if CUDA_AVAILABLE else 0,
    "n_tpu_cores": 0,
    "max_grad_norm": 1.0,
    "do_train": True,
    "do_predict": True,
    "accumulate_grad_batches": 1,
    "server_ip": "",
    "server_port": "",
    "seed": 42,
    "model_name_or_path": "sshleifer/bart-tiny-random",
    "config_name": "",
    "tokenizer_name": "facebook/bart-large",
    "do_lower_case": False,
    "learning_rate": 0.3,
    "lr_scheduler": "linear",
    "weight_decay": 0.0,
    "adam_epsilon": 1E-08,
    "warmup_steps": 0,
    "max_epochs": 1,
    "train_batch_size": 2,
    "eval_batch_size": 2,
    "max_source_length": 12,
    "max_target_length": 12,
    "val_max_target_length": 12,
    "test_max_target_length": 12,
    "fast_dev_run": False,
    "no_cache": False,
    "n_train": -1,
    "n_val": -1,
    "n_test": -1,
    "student_encoder_layers": 1,
    "freeze_encoder": False,
    "auto_scale_batch_size": False,
    "overwrite_output_dir": False,
    "student": None,
}






def    a__       ( __lowercase   ,	__lowercase      )		->		Union[str, Any]:
				_A         =	"\n".join(__lowercase      )
				Path(__lowercase      ).open("w"      ).writelines(__lowercase      )


a_											=			[" Sam ate lunch today.", "Sams lunch ingredients."]
a_											=			["A very interesting story about what I ate for lunch.", "Avocado, celery, turkey, coffee"]
a_											=			"patrickvonplaten/t5-tiny-random"
a_											=			"sshleifer/t5-tinier-random"
a_											=			"sshleifer/bart-tiny-random"
a_											=			"sshleifer/tiny-mbart"
a_											=			"sshleifer/tiny-marian-en-de"
a_											=			"stas/tiny-wmt19-en-de"


a_											=			logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)
logging.disable(logging.CRITICAL)  # remove noisy download output from tracebacks






def    a__       ( __lowercase      )		->		List[str]:
				for split in ["train", "val", "test"]:
								_dump_articles(os.path.join(__lowercase   ,	f"""{split}.source"""      )   ,	__lowercase      )
								_dump_articles(os.path.join(__lowercase   ,	f"""{split}.target"""      )   ,	__lowercase      )
				return tmp_dir
class snake_case					(     _UpperCamelCase):







		@classmethod
		def        a_       (    cls					:				List[Any]						)				->  Optional[Any]:





						'''simple docstring'''







						logging.disable(logging.CRITICAL						)  # remove noisy download output from tracebacks
						return cls







		@slow
		@require_torch_gpu
		def        a_       (    self					:				Union[str, Any]						)				->  Optional[int]:





						'''simple docstring'''







						_A         =	list_models()
						_A         =	"sshleifer"
						_A         =	[x.modelId for x in model_list if x.modelId.startswith(a__						)]
						_A         =	["sshleifer/blenderbot-3B", "sshleifer/blenderbot-90M"]
						_A         =	[]
						for m in model_ids:
										if m in allowed_to_be_broken:
														continue
										try:
														AutoConfig.from_pretrained(a__						)
										except Exception:
														failures.append(a__						)
						assert not failures, F"""The following models could not be loaded through AutoConfig: {failures}"""







		def        a_       (    self					:				Dict						)				->  Optional[int]:





						'''simple docstring'''







						_A         =	{"student_encoder_layers": 2, "student_decoder_layers": 1, "no_teacher": True}
						self._test_distiller_cli(a__						)







		def        a_       (    self					:				Tuple						)				->  Optional[Any]:





						'''simple docstring'''







						_A         =	{
						    "student_encoder_layers": 2,
						    "student_decoder_layers": 1,
						    "max_epochs": 4,
						    "val_check_interval": 0.2_5,
						    "alpha_hid": 2.0,
						    "model_name_or_path": "IGNORE_THIS_IT_DOESNT_GET_USED",
						}
						_A         =	self._test_distiller_cli(a__			,      check_contents=a__						)

						_A         =	list(Path(model.output_dir						).glob("*.ckpt"						)						)
						self.assertEqual(1			,      len(a__						)						)
						_A         =	list(Path(model.output_dir						).glob("**/*.bin"						)						)
						self.assertEqual(len(a__						)			,      2						)
						_A         =	lmap(str.strip			,      Path(model.hparams.data_dir						).joinpath("test.source"						).open().readlines()						)
						_A         =	tempfile.mktemp()  # XXX: not being cleaned up
						generate_summaries_or_translations(a__			,      a__			,      str(model.output_dir / "best_tfmr"						)						)
						self.assertTrue(Path(a__						).exists()						)

						_A         =	self.get_auto_remove_tmp_dir()
						convert_pl_to_hf(ckpts[0]			,      transformer_ckpts[0].parent			,      a__						)
						assert os.path.exists(os.path.join(a__			,      "pytorch_model.bin"						)						)







		def        a_       (    self					:				int						)				->  Union[str, Any]:





						'''simple docstring'''







						_A         =	AutoModelForSeqaSeqLM.from_pretrained(a__						)
						_A     ,						_A         =	model.dummy_inputs["input_ids"], model.dummy_inputs["attention_mask"]
						_A         =	torch.tensor([[0, 4, 8, 2], [0, 8, 2, 1]]			,      dtype=torch.long			,      device=model.device						)
						_A         =	target_ids[:, :-1].contiguous()  # Why this line?
						_A         =	target_ids[:, 1:].clone()  # why clone?
						_A         =	model(
						    a__			,      attention_mask=a__			,      decoder_input_ids=a__			,      labels=a__			,      use_cache=a__						).loss

						_A         =	model(a__			,      attention_mask=a__			,      decoder_input_ids=a__			,      use_cache=a__						).logits

						_A         =	nn.functional.log_softmax(a__			,      dim=-1						)
						_A     ,						_A         =	label_smoothed_nll_loss(
						    a__			,      a__			,      0.1			,      ignore_index=model.config.pad_token_id						)
						with self.assertRaises(a__						):
										# TODO: understand why this breaks
										self.assertEqual(a__			,      a__						)







		def        a_       (    self					:				List[str]						)				->  List[str]:





						'''simple docstring'''







						_A         =	{
						    "student_encoder_layers": 2,
						    "student_decoder_layers": 1,
						    "num_train_epochs": 4,
						    "val_check_interval": 0.2_5,
						    "alpha_hid": 2.0,
						    "task": "translation",
						    "model_name_or_path": "IGNORE_THIS_IT_DOESNT_GET_USED",
						    "tokenizer_name": MBART_TINY,
						    "teacher": MBART_TINY,
						    "src_lang": "en_XX",
						    "tgt_lang": "ro_RO",
						}
						_A         =	self._test_distiller_cli(a__			,      check_contents=a__						)
						assert model.model.config.model_type == "mbart"

						_A         =	list(Path(model.output_dir						).glob("*.ckpt"						)						)
						self.assertEqual(1			,      len(a__						)						)
						_A         =	list(Path(model.output_dir						).glob("**/*.bin"						)						)
						_A         =	list(Path(model.output_dir						).glob("best_tfmr/*"						)						)
						assert len(a__						) > 2
						self.assertEqual(len(a__						)			,      2						)







		def        a_       (    self					:				Dict						)				->  Dict:





						'''simple docstring'''







						_A         =	{
						    "student_encoder_layers": 1,
						    "student_decoder_layers": 1,
						    "alpha_hid": 2.0,
						    "teacher": T5_TINY,
						    "model_name_or_path": T5_TINY,
						    "tokenizer_name": T5_TINY,
						}
						self._test_distiller_cli(a__						)







		def        a_       (    self					:				Union[str, Any]						)				->  List[str]:





						'''simple docstring'''







						_A         =	{
						    "teacher": T5_TINY,
						    "student": T5_TINIER,
						    "model_name_or_path": T5_TINIER,
						    "tokenizer_name": T5_TINIER,
						}
						self._test_distiller_cli(a__						)







		def        a_       (    self					:				List[Any]			,      a__					:				List[str]			,      a__					:				Union[str, Any]=True						)				->  str:





						'''simple docstring'''







						_A         =	{
						    "label_smoothing": 0.0,
						    "early_stopping_patience": -1,
						    "train_batch_size": 1,
						    "eval_batch_size": 2,
						    "max_epochs": 2,
						    "alpha_mlm": 0.2,
						    "alpha_ce": 0.8,
						    "do_predict": True,
						    "model_name_or_path": "sshleifer/tinier_bart",
						    "teacher": CHEAP_ARGS["model_name_or_path"],
						    "val_check_interval": 0.5,
						}
						default_updates.update(a__						)
						_A         =	CHEAP_ARGS.copy()
						_A         =	make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()						)
						_A         =	self.get_auto_remove_tmp_dir()

						args_d.update(data_dir=a__			,      output_dir=a__			,      **a__						)
						_A         =	distill_main(argparse.Namespace(**a__						)						)
						if not check_contents:
										return model
						_A         =	os.listdir(a__						)
						_A         =	{os.path.basename(a__						) for p in contents}
						_A         =	[p for p in contents if p.endswith("ckpt"						)]
						assert len(a__						) > 0

						self.assertIn("test_generations.txt"			,      a__						)
						self.assertIn("test_results.txt"			,      a__						)

						_A         =	load_json(model.metrics_save_path						)
						_A         =	metrics["val"][-1]
						self.assertGreaterEqual(last_step_stats["val_avg_gen_time"]			,      0.0_1						)
						self.assertGreaterEqual(1.0			,      last_step_stats["val_avg_gen_time"]						)
						self.assertIsInstance(last_step_stats[F"""val_avg_{model.val_metric}"""]			,      a__						)
						_A         =	int(args_d["max_epochs"] * (1 / args_d["val_check_interval"]) + 1						)
						self.assertEqual(len(metrics["val"]						)			,      a__						)
						self.assertEqual(len(metrics["test"]						)			,      1						)
						return model





class snake_case					(     _UpperCamelCase):







		@parameterized.expand(
		    [T5_TINY, BART_TINY, MBART_TINY, MARIAN_TINY, FSMT_TINY]			,      )
		def        a_       (    self					:				Tuple			,      a__					:				List[str]						)				->  Tuple:





						'''simple docstring'''







						_A         =	CHEAP_ARGS.copy()
						_A         =	"translation" if model in [MBART_TINY, MARIAN_TINY, FSMT_TINY] else "summarization"
						_A         =	0.1 if task == "translation" else 0

						_A         =	make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()						)
						_A         =	self.get_auto_remove_tmp_dir()
						args_d.update(
						    data_dir=a__			,      model_name_or_path=a__			,      tokenizer_name=a__			,      train_batch_size=2			,      eval_batch_size=2			,      output_dir=a__			,      do_predict=a__			,      task=a__			,      src_lang="en_XX"			,      tgt_lang="ro_RO"			,      freeze_encoder=a__			,      freeze_embeds=a__			,      )
						assert "n_train" in args_d
						_A         =	argparse.Namespace(**a__						)
						_A         =	main(a__						)

						_A         =	module.model.get_input_embeddings()
						assert not input_embeds.weight.requires_grad
						if model == T5_TINY:
										_A         =	module.model.lm_head
										assert not lm_head.weight.requires_grad
										assert (lm_head.weight == input_embeds.weight).all().item()
						elif model == FSMT_TINY:
										_A         =	module.model.model
										_A         =	fsmt.decoder.embed_positions
										assert not embed_pos.weight.requires_grad
										assert not fsmt.decoder.embed_tokens.weight.requires_grad
										# check that embeds are not the same
										assert fsmt.decoder.embed_tokens != fsmt.encoder.embed_tokens
						else:
										_A         =	module.model.model
										_A         =	bart.decoder.embed_positions
										assert not embed_pos.weight.requires_grad
										assert not bart.shared.weight.requires_grad
										# check that embeds are the same
										assert bart.decoder.embed_tokens == bart.encoder.embed_tokens
										assert bart.decoder.embed_tokens == bart.shared

						_A         =	load_json(module.output_dir / "text_batch.json"						)
						assert isinstance(a__			,      a__						)
						assert len(a__						) >= 4







		def        a_       (    self					:				Tuple						)				->  Union[str, Any]:





						'''simple docstring'''







						_A         =	CHEAP_ARGS.copy()

						_A         =	"summarization"
						_A         =	make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()						)

						args_d.update(
						    data_dir=a__			,      tokenizer_name=a__			,      train_batch_size=2			,      eval_batch_size=2			,      do_predict=a__			,      task=a__			,      src_lang="en_XX"			,      tgt_lang="ro_RO"			,      freeze_encoder=a__			,      freeze_embeds=a__			,      )

						# test models whose config includes the extra_model_args
						_A         =	BART_TINY
						_A         =	self.get_auto_remove_tmp_dir()
						_A         =	args_d.copy()
						args_da.update(
						    model_name_or_path=a__			,      output_dir=a__			,      )
						_A         =	("encoder_layerdrop", "decoder_layerdrop", "dropout", "attention_dropout")
						for p in extra_model_params:
										_A         =	0.5
						_A         =	argparse.Namespace(**a__						)
						_A         =	main(a__						)
						for p in extra_model_params:
										assert getattr(model.config			,      a__						) == 0.5, F"""failed to override the model config for param {p}"""

						# test models whose config doesn't include the extra_model_args
						_A         =	T5_TINY
						_A         =	self.get_auto_remove_tmp_dir()
						_A         =	args_d.copy()
						args_da.update(
						    model_name_or_path=a__			,      output_dir=a__			,      )
						_A         =	"encoder_layerdrop"
						_A         =	0.5
						_A         =	argparse.Namespace(**a__						)
						with pytest.raises(a__						) as excinfo:
										_A         =	main(a__						)
						assert str(excinfo.value						) == F"""model config doesn't have a `{unsupported_param}` attribute"""







		def        a_       (    self					:				Any						)				->  Optional[int]:





						'''simple docstring'''







						_A         =	CHEAP_ARGS.copy()

						_A         =	"summarization"
						_A         =	make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()						)

						_A         =	BART_TINY
						_A         =	self.get_auto_remove_tmp_dir()

						args_d.update(
						    data_dir=a__			,      model_name_or_path=a__			,      output_dir=a__			,      tokenizer_name=a__			,      train_batch_size=2			,      eval_batch_size=2			,      do_predict=a__			,      task=a__			,      src_lang="en_XX"			,      tgt_lang="ro_RO"			,      freeze_encoder=a__			,      freeze_embeds=a__			,      )

						# emulate finetune.py
						_A         =	argparse.ArgumentParser()
						_A         =	pl.Trainer.add_argparse_args(a__						)
						_A         =	SummarizationModule.add_model_specific_args(a__			,      os.getcwd()						)
						_A         =	{"--help": True}

						# --help test
						with pytest.raises(a__						) as excinfo:
										with CaptureStdout() as cs:
														_A         =	parser.parse_args(a__						)
										assert False, "--help is expected to sys.exit"
						assert excinfo.type == SystemExit
						_A         =	lightning_base.arg_to_scheduler_metavar
						assert expected in cs.out, "--help is expected to list the supported schedulers"

						# --lr_scheduler=non_existing_scheduler test
						_A         =	"non_existing_scheduler"
						_A         =	{F"""--lr_scheduler={unsupported_param}"""}
						with pytest.raises(a__						) as excinfo:
										with CaptureStderr() as cs:
														_A         =	parser.parse_args(a__						)
										assert False, "invalid argument is expected to sys.exit"
						assert excinfo.type == SystemExit
						_A         =	F"""invalid choice: '{unsupported_param}'"""
						assert expected in cs.err, F"""should have bailed on invalid choice of scheduler {unsupported_param}"""

						# --lr_scheduler=existing_scheduler test
						_A         =	"cosine"
						_A         =	args_d.copy()
						_A         =	supported_param
						_A         =	argparse.Namespace(**a__						)
						_A         =	main(a__						)
						assert (
						    getattr(model.hparams			,      "lr_scheduler"						) == supported_param
						), F"""lr_scheduler={supported_param} shouldn't fail"""







from __future__ import annotations

import inspect
import unittest

import numpy as np

from transformers import SwinConfig
from transformers.testing_utils import require_tf, require_vision, slow, to_atuple
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.models.swin.modeling_tf_swin import (
							    TF_SWIN_PRETRAINED_MODEL_ARCHIVE_LIST,
							    TFSwinForImageClassification,
							    TFSwinForMaskedImageModeling,
							    TFSwinModel,
							)


if is_vision_available():
							from PIL import Image

							from transformers import AutoImageProcessor




class 						UpperCAmelCase_  :


						'''simple docstring'''




						def __init__(		self						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE=13						, _SCREAMING_SNAKE_CASE=32						, _SCREAMING_SNAKE_CASE=2						, _SCREAMING_SNAKE_CASE=3						, _SCREAMING_SNAKE_CASE=16						, _SCREAMING_SNAKE_CASE=[1, 2, 1]						, _SCREAMING_SNAKE_CASE=[2, 2, 4]						, _SCREAMING_SNAKE_CASE=2						, _SCREAMING_SNAKE_CASE=2.0						, _SCREAMING_SNAKE_CASE=True						, _SCREAMING_SNAKE_CASE=0.0						, _SCREAMING_SNAKE_CASE=0.0						, _SCREAMING_SNAKE_CASE=0.1						, _SCREAMING_SNAKE_CASE="gelu"						, _SCREAMING_SNAKE_CASE=False						, _SCREAMING_SNAKE_CASE=True						, _SCREAMING_SNAKE_CASE=0.02						, _SCREAMING_SNAKE_CASE=1e-5						, _SCREAMING_SNAKE_CASE=True						, _SCREAMING_SNAKE_CASE=None						, _SCREAMING_SNAKE_CASE=True						, _SCREAMING_SNAKE_CASE=10						, _SCREAMING_SNAKE_CASE=8						, )       -> None:
								snake_case_ :							List[str]         =				parent
								snake_case_ :							Dict         =				batch_size
								snake_case_ :							str         =				image_size
								snake_case_ :							Dict         =				patch_size
								snake_case_ :							Dict         =				num_channels
								snake_case_ :							Any         =				embed_dim
								snake_case_ :							int         =				depths
								snake_case_ :							Tuple         =				num_heads
								snake_case_ :							List[str]         =				window_size
								snake_case_ :							List[Any]         =				mlp_ratio
								snake_case_ :							Any         =				qkv_bias
								snake_case_ :							int         =				hidden_dropout_prob
								snake_case_ :							Tuple         =				attention_probs_dropout_prob
								snake_case_ :							str         =				drop_path_rate
								snake_case_ :							Any         =				hidden_act
								snake_case_ :							Tuple         =				use_absolute_embeddings
								snake_case_ :							Tuple         =				patch_norm
								snake_case_ :							Dict         =				layer_norm_eps
								snake_case_ :							Any         =				initializer_range
								snake_case_ :							str         =				is_training
								snake_case_ :							str         =				scope
								snake_case_ :							List[Any]         =				use_labels
								snake_case_ :							Dict         =				type_sequence_label_size
								snake_case_ :							List[Any]         =				encoder_stride






						def 					_lowerCAmelCase				(		self )       -> Optional[Any]:
								snake_case_ :							List[str]         =				floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size] )

								snake_case_ :							Dict         =				None
								if self.use_labels:
										snake_case_ :							List[str]         =				ids_tensor([self.batch_size]						, self.type_sequence_label_size )

								snake_case_ :							List[Any]         =				self.get_config()

								return config, pixel_values, labels






						def 					_lowerCAmelCase				(		self )       -> Optional[int]:
								return SwinConfig(
								    image_size=self.image_size						, patch_size=self.patch_size						, num_channels=self.num_channels						, embed_dim=self.embed_dim						, depths=self.depths						, num_heads=self.num_heads						, window_size=self.window_size						, mlp_ratio=self.mlp_ratio						, qkv_bias=self.qkv_bias						, hidden_dropout_prob=self.hidden_dropout_prob						, attention_probs_dropout_prob=self.attention_probs_dropout_prob						, drop_path_rate=self.drop_path_rate						, hidden_act=self.hidden_act						, use_absolute_embeddings=self.use_absolute_embeddings						, path_norm=self.patch_norm						, layer_norm_eps=self.layer_norm_eps						, initializer_range=self.initializer_range						, encoder_stride=self.encoder_stride						, )






						def 					_lowerCAmelCase				(		self						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE )       -> Union[str, Any]:
								snake_case_ :							List[Any]         =				TFSwinModel(config=_SCREAMING_SNAKE_CASE )
								snake_case_ :							str         =				model(_SCREAMING_SNAKE_CASE )

								snake_case_ :							Tuple         =				((config.image_size // config.patch_size) ** 2) // (4 ** (len(config.depths ) - 1))
								snake_case_ :							Optional[Any]         =				int(config.embed_dim * 2 ** (len(config.depths ) - 1) )

								self.parent.assertEqual(result.last_hidden_state.shape						, (self.batch_size, expected_seq_len, expected_dim) )






						def 					_lowerCAmelCase				(		self						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE )       -> List[str]:
								snake_case_ :							List[Any]         =				TFSwinForMaskedImageModeling(config=_SCREAMING_SNAKE_CASE )
								snake_case_ :							List[str]         =				model(_SCREAMING_SNAKE_CASE )
								self.parent.assertEqual(
								    result.logits.shape						, (self.batch_size, self.num_channels, self.image_size, self.image_size) )

								# test greyscale images
								snake_case_ :							Dict         =				1
								snake_case_ :							str         =				TFSwinForMaskedImageModeling(_SCREAMING_SNAKE_CASE )

								snake_case_ :							Optional[Any]         =				floats_tensor([self.batch_size, 1, self.image_size, self.image_size] )
								snake_case_ :							Tuple         =				model(_SCREAMING_SNAKE_CASE )
								self.parent.assertEqual(result.logits.shape						, (self.batch_size, 1, self.image_size, self.image_size) )






						def 					_lowerCAmelCase				(		self						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE )       -> Optional[Any]:
								snake_case_ :							str         =				self.type_sequence_label_size
								snake_case_ :							Dict         =				TFSwinForImageClassification(_SCREAMING_SNAKE_CASE )
								snake_case_ :							Optional[int]         =				model(_SCREAMING_SNAKE_CASE						, labels=_SCREAMING_SNAKE_CASE )
								self.parent.assertEqual(result.logits.shape						, (self.batch_size, self.type_sequence_label_size) )

								# test greyscale images
								snake_case_ :							List[str]         =				1
								snake_case_ :							Optional[int]         =				TFSwinForImageClassification(_SCREAMING_SNAKE_CASE )
								snake_case_ :							Optional[int]         =				floats_tensor([self.batch_size, 1, self.image_size, self.image_size] )
								snake_case_ :							Tuple         =				model(_SCREAMING_SNAKE_CASE )
								self.parent.assertEqual(result.logits.shape						, (self.batch_size, self.type_sequence_label_size) )






						def 					_lowerCAmelCase				(		self )       -> str:
								snake_case_ :							List[str]         =				self.prepare_config_and_inputs()
								snake_case_       ,				snake_case_       ,				snake_case_ :							Union[str, Any]         =				config_and_inputs
								snake_case_ :							str         =				{"pixel_values": pixel_values}
								return config, inputs_dict




@require_tf
class 						UpperCAmelCase_  (		SCREAMING_SNAKE_CASE__    ,     SCREAMING_SNAKE_CASE__    ,     unittest.TestCase    ):


						'''simple docstring'''




						A   :      Optional[int]						 =    (
						    (
						        TFSwinModel,
						        TFSwinForImageClassification,
						        TFSwinForMaskedImageModeling,
						    )
						    if is_tf_available()
						    else ()
						)
						A   :      Dict						 =    (
						    {'feature-extraction': TFSwinModel, 'image-classification': TFSwinForImageClassification}
						    if is_tf_available()
						    else {}
						)

						A   :      int						 =    False
						A   :      Tuple						 =    False
						A   :      Union[str, Any]						 =    False
						A   :      List[str]						 =    False






						def 					_lowerCAmelCase				(		self )       -> Optional[Any]:
								snake_case_ :							List[str]         =				TFSwinModelTester(self )
								snake_case_ :							Dict         =				ConfigTester(self						, config_class=_SCREAMING_SNAKE_CASE						, embed_dim=37 )






						def 					_lowerCAmelCase				(		self )       -> List[Any]:
								self.create_and_test_config_common_properties()
								self.config_tester.create_and_test_config_to_json_string()
								self.config_tester.create_and_test_config_to_json_file()
								self.config_tester.create_and_test_config_from_and_save_pretrained()
								self.config_tester.create_and_test_config_with_num_labels()
								self.config_tester.check_config_can_be_init_without_params()
								self.config_tester.check_config_arguments_init()






						def 					_lowerCAmelCase				(		self )       -> Any:
								return






						def 					_lowerCAmelCase				(		self )       -> Union[str, Any]:
								snake_case_ :							Dict         =				self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*_SCREAMING_SNAKE_CASE )






						def 					_lowerCAmelCase				(		self )       -> List[Any]:
								snake_case_ :							Any         =				self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_masked_image_modeling(*_SCREAMING_SNAKE_CASE )






						def 					_lowerCAmelCase				(		self )       -> int:
								snake_case_ :							str         =				self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_image_classification(*_SCREAMING_SNAKE_CASE )






						@unittest.skip(reason="Swin does not use inputs_embeds" )
						def 					_lowerCAmelCase				(		self )       -> Any:
								pass






						def 					_lowerCAmelCase				(		self )       -> List[Any]:
								snake_case_       ,				snake_case_ :							Optional[int]         =				self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
										snake_case_ :							Dict         =				model_class(_SCREAMING_SNAKE_CASE )
										self.assertIsInstance(model.get_input_embeddings()						, tf.keras.layers.Layer )
										snake_case_ :							int         =				model.get_output_embeddings()
										self.assertTrue(x is None or isinstance(_SCREAMING_SNAKE_CASE						, tf.keras.layers.Dense ) )






						def 					_lowerCAmelCase				(		self )       -> int:
								snake_case_       ,				snake_case_ :							List[Any]         =				self.model_tester.prepare_config_and_inputs_for_common()

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

										snake_case_ :							Optional[Any]         =				["pixel_values"]
										self.assertListEqual(arg_names[:1]						, _SCREAMING_SNAKE_CASE )






						def 					_lowerCAmelCase				(		self )       -> List[Any]:
								snake_case_       ,				snake_case_ :							Optional[int]         =				self.model_tester.prepare_config_and_inputs_for_common()
								snake_case_ :							Union[str, Any]         =				True

								for model_class in self.all_model_classes:
										snake_case_ :							str         =				True
										snake_case_ :							Tuple         =				False
										snake_case_ :							Any         =				True
										snake_case_ :							Union[str, Any]         =				model_class(_SCREAMING_SNAKE_CASE )
										snake_case_ :							str         =				model(**self._prepare_for_class(_SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE ) )
										snake_case_ :							Any         =				outputs.attentions
										snake_case_ :							str         =				len(self.model_tester.depths )
										self.assertEqual(len(_SCREAMING_SNAKE_CASE )						, _SCREAMING_SNAKE_CASE )

										# check that output_attentions also work using config
										del inputs_dict["output_attentions"]
										snake_case_ :							int         =				True
										snake_case_ :							List[Any]         =				config.window_size**2
										snake_case_ :							Optional[Any]         =				model_class(_SCREAMING_SNAKE_CASE )
										snake_case_ :							Dict         =				model(**self._prepare_for_class(_SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE ) )
										snake_case_ :							Optional[Any]         =				outputs.attentions
										self.assertEqual(len(_SCREAMING_SNAKE_CASE )						, _SCREAMING_SNAKE_CASE )

										self.assertListEqual(
										    list(attentions[0].shape[-3:] )						, [self.model_tester.num_heads[0], window_size_squared, window_size_squared]						, )
										snake_case_ :							Tuple         =				len(_SCREAMING_SNAKE_CASE )

										# Check attention is always last and order is fine
										snake_case_ :							str         =				True
										snake_case_ :							Optional[Any]         =				True
										snake_case_ :							Dict         =				model_class(_SCREAMING_SNAKE_CASE )
										snake_case_ :							str         =				model(**self._prepare_for_class(_SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE ) )

										if hasattr(self.model_tester						, "num_hidden_states_types" ):
												snake_case_ :							Optional[Any]         =				self.model_tester.num_hidden_states_types
										else:
												# also another +1 for reshaped_hidden_states
												snake_case_ :							Dict         =				2
										self.assertEqual(out_len + added_hidden_states						, len(_SCREAMING_SNAKE_CASE ) )

										snake_case_ :							List[str]         =				outputs.attentions

										self.assertEqual(len(_SCREAMING_SNAKE_CASE )						, _SCREAMING_SNAKE_CASE )

										self.assertListEqual(
										    list(self_attentions[0].shape[-3:] )						, [self.model_tester.num_heads[0], window_size_squared, window_size_squared]						, )






						def 					_lowerCAmelCase				(		self						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE )       -> List[Any]:
								snake_case_ :							Optional[int]         =				model_class(_SCREAMING_SNAKE_CASE )
								snake_case_ :							Optional[Any]         =				model(**self._prepare_for_class(_SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE ) )
								snake_case_ :							Optional[int]         =				outputs.hidden_states

								snake_case_ :							List[str]         =				getattr(
								    self.model_tester						, "expected_num_hidden_layers"						, len(self.model_tester.depths ) + 1 )
								self.assertEqual(len(_SCREAMING_SNAKE_CASE )						, _SCREAMING_SNAKE_CASE )

								# Swin has a different seq_length
								snake_case_ :							int         =				to_atuple(config.patch_size )

								snake_case_ :							Dict         =				(image_size[1] // patch_size[1]) * (image_size[0] // patch_size[0])

								self.assertListEqual(
								    list(hidden_states[0].shape[-2:] )						, [num_patches, self.model_tester.embed_dim]						, )

								snake_case_ :							Union[str, Any]         =				outputs.reshaped_hidden_states
								self.assertEqual(len(_SCREAMING_SNAKE_CASE )						, _SCREAMING_SNAKE_CASE )

								snake_case_       ,				snake_case_       ,				snake_case_       ,				snake_case_ :							Union[str, Any]         =				reshaped_hidden_states[0].shape

								snake_case_ :							Any         =				tf.reshape(reshaped_hidden_states[0]						, (batch_size, num_channels, height * width) )
								snake_case_ :							Tuple         =				tf.transpose(_SCREAMING_SNAKE_CASE						, (0, 2, 1) )

								self.assertListEqual(
								    list(reshaped_hidden_states.shape[-2:] )						, [num_patches, self.model_tester.embed_dim]						, )






						def 					_lowerCAmelCase				(		self )       -> Optional[Any]:
								snake_case_       ,				snake_case_ :							List[Any]         =				self.model_tester.prepare_config_and_inputs_for_common()

								snake_case_ :							Union[str, Any]         =				to_atuple(self.model_tester.image_size )

								for model_class in self.all_model_classes:
										snake_case_ :							Optional[Any]         =				True
										self.check_hidden_states_output(_SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE )

										# check that output_hidden_states also work using config
										del inputs_dict["output_hidden_states"]
										snake_case_ :							Union[str, Any]         =				True

										self.check_hidden_states_output(_SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE )






						def 					_lowerCAmelCase				(		self )       -> Optional[int]:
								snake_case_       ,				snake_case_ :							int         =				self.model_tester.prepare_config_and_inputs_for_common()
								snake_case_ :							Tuple         =				3

								snake_case_ :							Any         =				to_atuple(self.model_tester.image_size )
								snake_case_ :							Dict         =				to_atuple(config.patch_size )

								snake_case_ :							Union[str, Any]         =				image_size[0] + patch_size[0] - (image_size[0] % patch_size[0])
								snake_case_ :							int         =				image_size[1] + patch_size[1] - (image_size[1] % patch_size[1])

								for model_class in self.all_model_classes:
										snake_case_ :							List[str]         =				True
										self.check_hidden_states_output(_SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, (padded_height, padded_width) )

										# check that output_hidden_states also work using config
										del inputs_dict["output_hidden_states"]
										snake_case_ :							Tuple         =				True
										self.check_hidden_states_output(_SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE						, (padded_height, padded_width) )






						@slow
						def 					_lowerCAmelCase				(		self )       -> int:
								for model_name in TF_SWIN_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
										snake_case_ :							str         =				TFSwinModel.from_pretrained(_SCREAMING_SNAKE_CASE )
										self.assertIsNotNone(_SCREAMING_SNAKE_CASE )




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


						'''simple docstring'''




						@cached_property
						def 					_lowerCAmelCase				(		self )       -> Optional[Any]:
								return (
								    AutoImageProcessor.from_pretrained("microsoft/swin-tiny-patch4-window7-224" )
								    if is_vision_available()
								    else None
								)






						@slow
						def 					_lowerCAmelCase				(		self )       -> Tuple:
								snake_case_ :							Optional[Any]         =				TFSwinForImageClassification.from_pretrained("microsoft/swin-tiny-patch4-window7-224" )
								snake_case_ :							List[Any]         =				self.default_image_processor

								snake_case_ :							Union[str, Any]         =				Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png" )
								snake_case_ :							List[Any]         =				image_processor(images=_SCREAMING_SNAKE_CASE						, return_tensors="tf" )

								# forward pass
								snake_case_ :							Any         =				model(_SCREAMING_SNAKE_CASE )

								# verify the logits
								snake_case_ :							Any         =				tf.TensorShape((1, 1000) )
								self.assertEqual(outputs.logits.shape						, _SCREAMING_SNAKE_CASE )
								snake_case_ :							Union[str, Any]         =				tf.constant([-0.0948, -0.6454, -0.0921] )
								self.assertTrue(np.allclose(outputs.logits[0, :3]						, _SCREAMING_SNAKE_CASE						, atol=1e-4 ) )
