





'''simple docstring'''

import argparse
import logging
import os
from pathlib import Path
from typing import Any, Dict

import pytorch_lightning as pl
from pytorch_lightning.utilities import rank_zero_info

from transformers import (
    AdamW,
    AutoConfig,
    AutoModel,
    AutoModelForPreTraining,
    AutoModelForQuestionAnswering,
    AutoModelForSeqaSeqLM,
    AutoModelForSequenceClassification,
    AutoModelForTokenClassification,
    AutoModelWithLMHead,
    AutoTokenizer,
    PretrainedConfig,
    PreTrainedTokenizer,
)
from transformers.optimization import (
    Adafactor,
    get_cosine_schedule_with_warmup,
    get_cosine_with_hard_restarts_schedule_with_warmup,
    get_linear_schedule_with_warmup,
    get_polynomial_decay_schedule_with_warmup,
)
from transformers.utils.versions import require_version


lowerCAmelCase__	:					Tuple	     =       logging.getLogger(__name__)

require_version("pytorch_lightning>=1.0.4")

lowerCAmelCase__	:					Dict	     =       {
    'base': AutoModel,
    'sequence-classification': AutoModelForSequenceClassification,
    'question-answering': AutoModelForQuestionAnswering,
    'pretraining': AutoModelForPreTraining,
    'token-classification': AutoModelForTokenClassification,
    'language-modeling': AutoModelWithLMHead,
    'summarization': AutoModelForSeqaSeqLM,
    'translation': AutoModelForSeqaSeqLM,
}


# update this and the import above to support new schedulers from transformers.optimization
lowerCAmelCase__	:					int	     =       {
    'linear': get_linear_schedule_with_warmup,
    'cosine': get_cosine_schedule_with_warmup,
    'cosine_w_restarts': get_cosine_with_hard_restarts_schedule_with_warmup,
    'polynomial': get_polynomial_decay_schedule_with_warmup,
    # '': get_constant_schedule,             # not supported for now
    # '': get_constant_schedule_with_warmup, # not supported for now
}
lowerCAmelCase__	:					Tuple	     =       sorted(arg_to_scheduler.keys())
lowerCAmelCase__	:					Optional[Any]	     =       '{' + ', '.join(arg_to_scheduler_choices) + '}'


class 			SCREAMING_SNAKE_CASE__	(					pl.LightningModule						):







					"""simple docstring"""






					def __init__(					self    :	str						,     UpperCAmelCase_    :	argparse.Namespace						,     UpperCAmelCase_    :	Dict=None						,     UpperCAmelCase_    :	str="base"						,     UpperCAmelCase_    :	Optional[int]=None						,     UpperCAmelCase_    :	Optional[Any]=None						,     UpperCAmelCase_    :	Union[str, Any]=None						,     **UpperCAmelCase_    :	Any						,     ):







							"""simple docstring"""
							super().__init__()
							# TODO: move to self.save_hyperparameters()
							# self.save_hyperparameters()
							# can also expand arguments into trainer signature for easier reading

							self.save_hyperparameters(UpperCAmelCase_						)
							__UpperCAmelCase     :				List[Any]					   =			0
							__UpperCAmelCase     :				Dict					   =			Path(self.hparams.output_dir						)
							__UpperCAmelCase     :				Tuple					   =			self.hparams.cache_dir if self.hparams.cache_dir else None
							if config is None:
									__UpperCAmelCase     :				str					   =			AutoConfig.from_pretrained(
									    self.hparams.config_name if self.hparams.config_name else self.hparams.model_name_or_path						,     **({"num_labels": num_labels} if num_labels is not None else {})						,     cache_dir=UpperCAmelCase_						,     **UpperCAmelCase_						,     )
							else:
									__UpperCAmelCase     :				PretrainedConfig					   =			config

							__UpperCAmelCase     :				Union[str, Any]					   =			('encoder_layerdrop', 'decoder_layerdrop', 'dropout', 'attention_dropout')
							for p in extra_model_params:
									if getattr(self.hparams						,     UpperCAmelCase_						,     UpperCAmelCase_						):
											assert hasattr(self.config						,     UpperCAmelCase_						), f"model config doesn\'t have a `{p}` attribute"
											setattr(self.config						,     UpperCAmelCase_						,     getattr(self.hparams						,     UpperCAmelCase_						)						)

							if tokenizer is None:
									__UpperCAmelCase     :				str					   =			AutoTokenizer.from_pretrained(
									    self.hparams.tokenizer_name if self.hparams.tokenizer_name else self.hparams.model_name_or_path						,     cache_dir=UpperCAmelCase_						,     )
							else:
									__UpperCAmelCase     :				PreTrainedTokenizer					   =			tokenizer
							__UpperCAmelCase     :				Union[str, Any]					   =			MODEL_MODES[mode]
							if model is None:
									__UpperCAmelCase     :				Optional[Any]					   =			self.model_type.from_pretrained(
									    self.hparams.model_name_or_path						,     from_tf=bool(".ckpt" in self.hparams.model_name_or_path						)						,     config=self.config						,     cache_dir=UpperCAmelCase_						,     )
							else:
									__UpperCAmelCase     :				List[Any]					   =			model
					def 						lowerCamelCase_					(					self    :	Union[str, Any]						,     *UpperCAmelCase_    :	List[Any]						,     **UpperCAmelCase_    :	Optional[int]						):







							"""simple docstring"""
							__UpperCAmelCase     :				Optional[int]					   =			self.model_type.from_pretrained(*UpperCAmelCase_						,     **UpperCAmelCase_						)
					def 						lowerCamelCase_					(					self    :	Optional[int]						):







							"""simple docstring"""
							__UpperCAmelCase     :				Union[str, Any]					   =			arg_to_scheduler[self.hparams.lr_scheduler]
							__UpperCAmelCase     :				int					   =			get_schedule_func(
							    self.opt						,     num_warmup_steps=self.hparams.warmup_steps						,     num_training_steps=self.total_steps()						)
							__UpperCAmelCase     :				List[Any]					   =			{'scheduler': scheduler, 'interval': 'step', 'frequency': 1}
							return scheduler
					def 						lowerCamelCase_					(					self    :	Union[str, Any]						):







							"""simple docstring"""
							__UpperCAmelCase     :				List[str]					   =			self.model
							__UpperCAmelCase     :				List[str]					   =			['bias', 'LayerNorm.weight']
							__UpperCAmelCase     :				int					   =			[
							    {
							        'params': [p for n, p in model.named_parameters() if not any(nd in n for nd in no_decay						)],
							        'weight_decay': self.hparams.weight_decay,
							    },
							    {
							        'params': [p for n, p in model.named_parameters() if any(nd in n for nd in no_decay						)],
							        'weight_decay': 0.0,
							    },
							]
							if self.hparams.adafactor:
									__UpperCAmelCase     :				Union[str, Any]					   =			Adafactor(
									    UpperCAmelCase_						,     lr=self.hparams.learning_rate						,     scale_parameter=UpperCAmelCase_						,     relative_step=UpperCAmelCase_						)

							else:
									__UpperCAmelCase     :				Tuple					   =			AdamW(
									    UpperCAmelCase_						,     lr=self.hparams.learning_rate						,     eps=self.hparams.adam_epsilon						)
							__UpperCAmelCase     :				Optional[int]					   =			optimizer

							__UpperCAmelCase     :				str					   =			self.get_lr_scheduler()

							return [optimizer], [scheduler]
					def 						lowerCamelCase_					(					self    :	List[Any]						,     UpperCAmelCase_    :	Union[str, Any]						,     UpperCAmelCase_    :	Optional[Any]						):







							"""simple docstring"""
							return self.validation_step(UpperCAmelCase_						,     UpperCAmelCase_						)
					def 						lowerCamelCase_					(					self    :	List[str]						,     UpperCAmelCase_    :	str						):







							"""simple docstring"""
							return self.validation_end(UpperCAmelCase_						)
					def 						lowerCamelCase_					(					self    :	int						):







							"""simple docstring"""
							__UpperCAmelCase     :				List[str]					   =			max(1						,     self.hparams.gpus						)  # TODO: consider num_tpu_cores
							__UpperCAmelCase     :				str					   =			self.hparams.train_batch_size * self.hparams.accumulate_grad_batches * num_devices
							return (self.dataset_size / effective_batch_size) * self.hparams.max_epochs
					def 						lowerCamelCase_					(					self    :	Any						,     UpperCAmelCase_    :	Tuple						):







							"""simple docstring"""




							if stage == "test":
									__UpperCAmelCase     :				Any					   =			len(self.test_dataloader().dataset						)
							else:
									__UpperCAmelCase     :				List[str]					   =			self.get_dataloader("train"						,     self.hparams.train_batch_size						,     shuffle=UpperCAmelCase_						)
									__UpperCAmelCase     :				Dict					   =			len(self.train_dataloader().dataset						)
					def 						lowerCamelCase_					(					self    :	List[Any]						,     UpperCAmelCase_    :	str						,     UpperCAmelCase_    :	int						,     UpperCAmelCase_    :	bool = False						):







							"""simple docstring"""
							raise NotImplementedError("You must implement this for your task"						)
					def 						lowerCamelCase_					(					self    :	List[Any]						):







							"""simple docstring"""
							return self.train_loader
					def 						lowerCamelCase_					(					self    :	Tuple						):







							"""simple docstring"""
							return self.get_dataloader("dev"						,     self.hparams.eval_batch_size						,     shuffle=UpperCAmelCase_						)
					def 						lowerCamelCase_					(					self    :	Dict						):







							"""simple docstring"""
							return self.get_dataloader("test"						,     self.hparams.eval_batch_size						,     shuffle=UpperCAmelCase_						)
					def 						lowerCamelCase_					(					self    :	int						,     UpperCAmelCase_    :	Dict						):







							"""simple docstring"""
							return os.path.join(
							    self.hparams.data_dir						,     "cached_{}_{}_{}".format(
							        UpperCAmelCase_						,     list(filter(UpperCAmelCase_						,     self.hparams.model_name_or_path.split("/"						)						)						).pop()						,     str(self.hparams.max_seq_length						)						,     )						,     )
					@pl.utilities.rank_zero_only
					def 						lowerCamelCase_					(					self    :	List[Any]						,     UpperCAmelCase_    :	Dict[str, Any]						):







							"""simple docstring"""
							__UpperCAmelCase     :				Any					   =			self.output_dir.joinpath("best_tfmr"						)
							__UpperCAmelCase     :				Tuple					   =			self.step_count
							self.model.save_pretrained(UpperCAmelCase_						)
							self.tokenizer.save_pretrained(UpperCAmelCase_						)






					@staticmethod
					def 						lowerCamelCase_					(					UpperCAmelCase_    :	Optional[Any]						,     UpperCAmelCase_    :	List[Any]						):







							"""simple docstring"""
							parser.add_argument(
							    "--model_name_or_path"						,     default=UpperCAmelCase_						,     type=UpperCAmelCase_						,     required=UpperCAmelCase_						,     help="Path to pretrained model or model identifier from huggingface.co/models"						,     )
							parser.add_argument(
							    "--config_name"						,     default=""						,     type=UpperCAmelCase_						,     help="Pretrained config name or path if not the same as model_name"						)
							parser.add_argument(
							    "--tokenizer_name"						,     default=UpperCAmelCase_						,     type=UpperCAmelCase_						,     help="Pretrained tokenizer name or path if not the same as model_name"						,     )
							parser.add_argument(
							    "--cache_dir"						,     default=""						,     type=UpperCAmelCase_						,     help="Where do you want to store the pre-trained models downloaded from huggingface.co"						,     )
							parser.add_argument(
							    "--encoder_layerdrop"						,     type=UpperCAmelCase_						,     help="Encoder layer dropout probability (Optional). Goes into model.config"						,     )
							parser.add_argument(
							    "--decoder_layerdrop"						,     type=UpperCAmelCase_						,     help="Decoder layer dropout probability (Optional). Goes into model.config"						,     )
							parser.add_argument(
							    "--dropout"						,     type=UpperCAmelCase_						,     help="Dropout probability (Optional). Goes into model.config"						,     )
							parser.add_argument(
							    "--attention_dropout"						,     type=UpperCAmelCase_						,     help="Attention dropout probability (Optional). Goes into model.config"						,     )
							parser.add_argument("--learning_rate"						,     default=5e-5						,     type=UpperCAmelCase_						,     help="The initial learning rate for Adam."						)
							parser.add_argument(
							    "--lr_scheduler"						,     default="linear"						,     choices=UpperCAmelCase_						,     metavar=UpperCAmelCase_						,     type=UpperCAmelCase_						,     help="Learning rate scheduler"						,     )
							parser.add_argument("--weight_decay"						,     default=0.0						,     type=UpperCAmelCase_						,     help="Weight decay if we apply some."						)
							parser.add_argument("--adam_epsilon"						,     default=1e-8						,     type=UpperCAmelCase_						,     help="Epsilon for Adam optimizer."						)
							parser.add_argument("--warmup_steps"						,     default=0						,     type=UpperCAmelCase_						,     help="Linear warmup over warmup_steps."						)
							parser.add_argument("--num_workers"						,     default=4						,     type=UpperCAmelCase_						,     help="kwarg passed to DataLoader"						)
							parser.add_argument("--num_train_epochs"						,     dest="max_epochs"						,     default=3						,     type=UpperCAmelCase_						)
							parser.add_argument("--train_batch_size"						,     default=32						,     type=UpperCAmelCase_						)
							parser.add_argument("--eval_batch_size"						,     default=32						,     type=UpperCAmelCase_						)
							parser.add_argument("--adafactor"						,     action="store_true"						)






class 			SCREAMING_SNAKE_CASE__	(					pl.Callback						):







					"""simple docstring"""






					def 						lowerCamelCase_					(					self    :	Tuple						,     UpperCAmelCase_    :	Optional[int]						,     UpperCAmelCase_    :	str						):







							"""simple docstring"""




							if (
							    trainer.is_global_zero and trainer.global_rank == 0
							):  # we initialize the retriever only on master worker with RAY. In new pytorch-lightning accelorators are removed.
									pl_module.model.rag.retriever.init_retrieval()  # better to use hook functions.






class 			SCREAMING_SNAKE_CASE__	(					pl.Callback						):







					"""simple docstring"""






					def 						lowerCamelCase_					(					self    :	Optional[Any]						,     UpperCAmelCase_    :	Optional[Any]						,     UpperCAmelCase_    :	Optional[int]						):







							"""simple docstring"""
							__UpperCAmelCase     :				Dict					   =			trainer.lr_schedulers[0]['scheduler']
							__UpperCAmelCase     :				Optional[int]					   =			{f"lr_group_{i}": lr for i, lr in enumerate(lr_scheduler.get_lr()						)}
							pl_module.logger.log_metrics(UpperCAmelCase_						)
					def 						lowerCamelCase_					(					self    :	List[str]						,     UpperCAmelCase_    :	pl.Trainer						,     UpperCAmelCase_    :	pl.LightningModule						):







							"""simple docstring"""
							rank_zero_info("***** Validation results *****"						)
							__UpperCAmelCase     :				int					   =			trainer.callback_metrics
							# Log results
							for key in sorted(UpperCAmelCase_						):
									if key not in ["log", "progress_bar"]:
											rank_zero_info("{} = {}\n".format(UpperCAmelCase_						,     str(metrics[key]						)						)						)






					def 						lowerCamelCase_					(					self    :	int						,     UpperCAmelCase_    :	pl.Trainer						,     UpperCAmelCase_    :	pl.LightningModule						):







							"""simple docstring"""
							rank_zero_info("***** Test results *****"						)
							__UpperCAmelCase     :				Optional[Any]					   =			trainer.callback_metrics
							# Log and save results to file
							__UpperCAmelCase     :				List[Any]					   =			os.path.join(pl_module.hparams.output_dir						,     "test_results.txt"						)
							with open(UpperCAmelCase_						,     "w"						) as writer:
									for key in sorted(UpperCAmelCase_						):
											if key not in ["log", "progress_bar"]:
													rank_zero_info("{} = {}\n".format(UpperCAmelCase_						,     str(metrics[key]						)						)						)
													writer.write("{} = {}\n".format(UpperCAmelCase_						,     str(metrics[key]						)						)						)






def 		__UpperCamelCase		(       _UpperCAmelCase,    _UpperCAmelCase   ):
		parser.add_argument(
		    "--output_dir",    default=A_,    type=A_,    required=A_,    help="The output directory where the model predictions and checkpoints will be written.",    )
		parser.add_argument(
		    "--fp16",    action="store_true",    help="Whether to use 16-bit (mixed) precision (through NVIDIA apex) instead of 32-bit",    )

		parser.add_argument(
		    "--fp16_opt_level",    type=A_,    default="O2",    help=(
		        "For fp16: Apex AMP optimization level selected in [\'O0\', \'O1\', \'O2\', and \'O3\']."
		        "See details at https://nvidia.github.io/apex/amp.html"
		    ),    )
		parser.add_argument("--n_tpu_cores",    dest="tpu_cores",    type=A_   )
		parser.add_argument("--max_grad_norm",    dest="gradient_clip_val",    default=1.0,    type=A_,    help="Max gradient norm"   )
		parser.add_argument("--do_train",    action="store_true",    help="Whether to run training."   )
		parser.add_argument("--do_predict",    action="store_true",    help="Whether to run predictions on the test set."   )
		parser.add_argument(
		    "--gradient_accumulation_steps",    dest="accumulate_grad_batches",    type=A_,    default=1,    help="Number of updates steps to accumulate before performing a backward/update pass.",    )
		parser.add_argument("--seed",    type=A_,    default=42,    help="random seed for initialization"   )
		parser.add_argument(
		    "--data_dir",    default=A_,    type=A_,    required=A_,    help="The input data dir. Should contain the training files for the CoNLL-2003 NER task.",    )






def 		__UpperCamelCase		(       _UpperCAmelCase,    _UpperCAmelCase,    _UpperCAmelCase=None,    _UpperCAmelCase=True,    _UpperCAmelCase=None,    _UpperCAmelCase=[],    _UpperCAmelCase=None,    _UpperCAmelCase=None,    **_UpperCAmelCase,    ):
		pl.seed_everything(args.seed   )

		# init model
		__UpperCAmelCase     :				Union[str, Any]					   =			Path(model.hparams.output_dir   )
		odir.mkdir(exist_ok=A_   )

		# add custom checkpoints
		if checkpoint_callback is None:
				__UpperCAmelCase     :				List[Any]					   =			pl.callbacks.ModelCheckpoint(
				    filepath=args.output_dir,    prefix="checkpoint",    monitor="val_loss",    mode="min",    save_top_k=1   )
		if early_stopping_callback:
				extra_callbacks.append(A_   )
		if logging_callback is None:
				__UpperCAmelCase     :				List[Any]					   =			LoggingCallback()

		__UpperCAmelCase     :				Tuple					   =			{}

		# TODO: remove with PyTorch 1.6 since pl uses native amp
		if args.fpaa:
				__UpperCAmelCase     :				str					   =			16
				# train_params["amp_level"] = args.fp16_opt_level

		if args.gpus > 1:
				__UpperCAmelCase     :				List[str]					   =			'auto'  # "ddp"
				__UpperCAmelCase     :				Optional[int]					   =			'ddp'

		__UpperCAmelCase     :				List[Any]					   =			args.accumulate_grad_batches
		__UpperCAmelCase     :				Optional[Any]					   =			None  # extra_train_kwargs.get("profiler", None) #get unwanted logs
		__UpperCAmelCase     :				List[Any]					   =			'auto'

		__UpperCAmelCase     :				int					   =			pl.Trainer.from_argparse_args(
		    A_,    weights_summary=A_,    callbacks=[logging_callback] + extra_callbacks + [checkpoint_callback] + [InitCallback()],    logger=A_,    **A_,    )

		if args.do_train:
				trainer.fit(A_   )

		return trainer












"""simple docstring"""


import inspect
import unittest

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

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


if is_torch_available():
   import torch
   from torch import nn

   from transformers import SwinaSRForImageSuperResolution, SwinaSRModel
   from transformers.models.swinasr.modeling_swinasr import SWIN2SR_PRETRAINED_MODEL_ARCHIVE_LIST

if is_vision_available():
   from PIL import Image

   from transformers import SwinaSRImageProcessor






class 						snake_case  :





    def __init__(							self	:					Tuple				,		A	:					Any				,		A	:					List[Any]=1_3				,		A	:					Union[str, Any]=3_2				,		A	:					Optional[Any]=1				,		A	:					Union[str, Any]=3				,		A	:					List[str]=1_6				,		A	:					Optional[Any]=[1, 2, 1]				,		A	:					Any=[2, 2, 4]				,		A	:					int=2				,		A	:					List[str]=2.0				,		A	:					Dict=True				,		A	:					Union[str, Any]=0.0				,		A	:					Union[str, Any]=0.0				,		A	:					Optional[int]=0.1				,		A	:					Optional[int]="gelu"				,		A	:					Any=False				,		A	:					Dict=True				,		A	:					int=0.02				,		A	:					Tuple=1E-5				,		A	:					List[str]=True				,		A	:					List[str]=None				,		A	:					str=False				,		A	:					Any=2				,		):


      '''simple docstring'''




      a     :			Tuple				=   parent
      a     :			Union[str, Any]				=   batch_size
      a     :			Tuple				=   image_size
      a     :			List[str]				=   patch_size
      a     :			Union[str, Any]				=   num_channels
      a     :			str				=   embed_dim
      a     :			int				=   depths
      a     :			Tuple				=   num_heads
      a     :			Any				=   window_size
      a     :			List[Any]				=   mlp_ratio
      a     :			str				=   qkv_bias
      a     :			Any				=   hidden_dropout_prob
      a     :			List[Any]				=   attention_probs_dropout_prob
      a     :			int				=   drop_path_rate
      a     :			List[Any]				=   hidden_act
      a     :			Union[str, Any]				=   use_absolute_embeddings
      a     :			Dict				=   patch_norm
      a     :			Optional[Any]				=   layer_norm_eps
      a     :			Dict				=   initializer_range
      a     :			Optional[Any]				=   is_training
      a     :			str				=   scope
      a     :			List[Any]				=   use_labels
      a     :			Union[str, Any]				=   upscale

      # here we set some attributes to make tests pass
      a     :			str				=   len(A			)
      a     :			List[Any]				=   embed_dim
      a     :			List[Any]				=   (image_size // patch_size) ** 2





    def 				lowerCamelCase__      (							self	:					str			):


      '''simple docstring'''




      a     :			List[Any]				=   floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]			)

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

      a     :			Any				=   self.get_config()

      return config, pixel_values, labels





    def 				lowerCamelCase__      (							self	:					str			):


      '''simple docstring'''




      return SwinaSRConfig(
          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				,		upscale=self.upscale				,		)





    def 				lowerCamelCase__      (							self	:					Union[str, Any]				,		A	:					Dict				,		A	:					Union[str, Any]				,		A	:					List[str]			):


      '''simple docstring'''




      a     :			List[Any]				=   SwinaSRModel(config=A			)
      model.to(A			)
      model.eval()
      a     :			Optional[int]				=   model(A			)

      self.parent.assertEqual(
          result.last_hidden_state.shape				,		(self.batch_size, self.embed_dim, self.image_size, self.image_size)			)





    def 				lowerCamelCase__      (							self	:					List[str]				,		A	:					Tuple				,		A	:					str				,		A	:					Tuple			):


      '''simple docstring'''




      a     :			Optional[Any]				=   SwinaSRForImageSuperResolution(A			)
      model.to(A			)
      model.eval()
      a     :			List[str]				=   model(A			)

      a     :			Any				=   self.image_size * self.upscale
      self.parent.assertEqual(
          result.reconstruction.shape				,		(self.batch_size, self.num_channels, expected_image_size, expected_image_size)			)





    def 				lowerCamelCase__      (							self	:					Union[str, Any]			):


      '''simple docstring'''




      a     :			Tuple				=   self.prepare_config_and_inputs()
      a,     a,     a     :			Union[str, Any]				=   config_and_inputs
      a     :			List[str]				=   {'pixel_values': pixel_values}
      return config, inputs_dict






@require_torch
class 						snake_case  (  UpperCAmelCase   , UpperCAmelCase   , unittest.TestCase       ):
    __magic_name__         =     (SwinaSRModel, SwinaSRForImageSuperResolution) if is_torch_available() else ()
    __magic_name__         =     {'''feature-extraction''': SwinaSRModel} if is_torch_available() else {}

    __magic_name__         =     False
    __magic_name__         =     False
    __magic_name__         =     False
    __magic_name__         =     False
    __magic_name__         =     False





    def 				lowerCamelCase__      (							self	:					Optional[int]			):


      '''simple docstring'''




      a     :			List[str]				=   SwinaSRModelTester(self			)
      a     :			int				=   ConfigTester(self				,		config_class=A				,		embed_dim=3_7			)





    def 				lowerCamelCase__      (							self	:					Optional[int]			):


      '''simple docstring'''




      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	:					List[Any]			):


      '''simple docstring'''




      a     :			str				=   self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*A			)





    def 				lowerCamelCase__      (							self	:					Tuple			):


      '''simple docstring'''




      a     :			Any				=   self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_image_super_resolution(*A			)





    @unittest.skip(reason='Got `CUDA error: misaligned address` with PyTorch 2.0.0.'			)
    def 				lowerCamelCase__      (							self	:					str			):


      '''simple docstring'''




      pass





    @unittest.skip(reason='Swin2SR does not use inputs_embeds'			)
    def 				lowerCamelCase__      (							self	:					Tuple			):


      '''simple docstring'''




      pass





    @unittest.skip(reason='Swin2SR does not support training yet'			)
    def 				lowerCamelCase__      (							self	:					Optional[int]			):


      '''simple docstring'''




      pass





    @unittest.skip(reason='Swin2SR does not support training yet'			)
    def 				lowerCamelCase__      (							self	:					int			):


      '''simple docstring'''




      pass





    def 				lowerCamelCase__      (							self	:					int			):


      '''simple docstring'''




      a,     a     :			Optional[int]				=   self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a     :			Union[str, Any]				=   model_class(A			)
        self.assertIsInstance(model.get_input_embeddings()				,		(nn.Module)			)
        a     :			Optional[int]				=   model.get_output_embeddings()
        self.assertTrue(x is None or isinstance(A				,		nn.Linear			)			)





    def 				lowerCamelCase__      (							self	:					Optional[int]			):


      '''simple docstring'''




      a,     a     :			int				=   self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a     :			Dict				=   model_class(A			)
        a     :			Union[str, Any]				=   inspect.signature(model.forward			)
        # signature.parameters is an OrderedDict => so arg_names order is deterministic
        a     :			Any				=   [*signature.parameters.keys()]

        a     :			Tuple				=   ['pixel_values']
        self.assertListEqual(arg_names[:1]				,		A			)





    @slow
    def 				lowerCamelCase__      (							self	:					List[str]			):


      '''simple docstring'''




      for model_name in SWIN2SR_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        a     :			Optional[int]				=   SwinaSRModel.from_pretrained(A			)
        self.assertIsNotNone(A			)





    def 				lowerCamelCase__      (							self	:					Dict			):


      '''simple docstring'''




      a,     a     :			str				=   self.model_tester.prepare_config_and_inputs_for_common()

      a     :			int				=   _config_zero_init(A			)
      for model_class in self.all_model_classes:
        a     :			List[str]				=   model_class(config=A			)
        for name, param in model.named_parameters():
          if "logit_scale" in name:
            continue
          if param.requires_grad:
            self.assertIn(
                ((param.data.mean() * 1E9).round() / 1E9).item()				,		[0.0, 1.0]				,		msg=F'''Parameter {name} of model {model_class} seems not properly initialized'''				,		)





    def 				lowerCamelCase__      (							self	:					List[str]			):


      '''simple docstring'''




      a,     a     :			Union[str, Any]				=   self.model_tester.prepare_config_and_inputs_for_common()
      a     :			List[str]				=   True

      for model_class in self.all_model_classes:
        a     :			Dict				=   True
        a     :			int				=   False
        a     :			str				=   True
        a     :			Optional[int]				=   model_class(A			)
        model.to(A			)
        model.eval()
        with torch.no_grad():
          a     :			Tuple				=   model(**self._prepare_for_class(A				,		A			)			)
        a     :			Optional[Any]				=   outputs.attentions
        a     :			int				=   len(self.model_tester.depths			)
        self.assertEqual(len(A			)				,		A			)

        # check that output_attentions also work using config
        del inputs_dict["output_attentions"]
        a     :			List[str]				=   True
        a     :			List[Any]				=   config.window_size**2
        a     :			List[str]				=   model_class(A			)
        model.to(A			)
        model.eval()
        with torch.no_grad():
          a     :			str				=   model(**self._prepare_for_class(A				,		A			)			)
        a     :			int				=   outputs.attentions
        self.assertEqual(len(A			)				,		A			)

        self.assertListEqual(
            list(attentions[0].shape[-3:]			)				,		[self.model_tester.num_heads[0], window_size_squared, window_size_squared]				,		)
        a     :			int				=   len(A			)

        # Check attention is always last and order is fine
        a     :			List[str]				=   True
        a     :			List[Any]				=   True
        a     :			str				=   model_class(A			)
        model.to(A			)
        model.eval()
        with torch.no_grad():
          a     :			List[Any]				=   model(**self._prepare_for_class(A				,		A			)			)

        self.assertEqual(out_len + 1				,		len(A			)			)

        a     :			Tuple				=   outputs.attentions

        self.assertEqual(len(A			)				,		A			)

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






@require_vision
@require_torch
@slow
class 						snake_case  (  unittest.TestCase       ):





    def 				lowerCamelCase__      (							self	:					Optional[int]			):


      '''simple docstring'''




      a     :			Optional[int]				=   SwinaSRImageProcessor()
      a     :			Optional[int]				=   SwinaSRForImageSuperResolution.from_pretrained('caidas/swin2SR-classical-sr-x2-64'			).to(A			)

      a     :			str				=   Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'			)
      a     :			Union[str, Any]				=   processor(images=A				,		return_tensors='pt'			).to(A			)

      # forward pass
      with torch.no_grad():
        a     :			Tuple				=   model(**A			)

      # verify the logits
      a     :			Optional[int]				=   torch.Size([1, 3, 9_7_6, 1_2_9_6]			)
      self.assertEqual(outputs.reconstruction.shape				,		A			)
      a     :			Tuple				=   torch.tensor(
          [[0.54_58, 0.55_46, 0.56_38], [0.55_26, 0.55_65, 0.56_51], [0.53_96, 0.54_26, 0.56_21]]			).to(A			)
      self.assertTrue(torch.allclose(outputs.reconstruction[0, 0, :3, :3]				,		A				,		atol=1E-4			)			)




