
import argparse
import dataclasses
import json
import logging
import os
import shutil
from typing import List, Optional

import datasets
from accelerate import Accelerator
from datasets import load_dataset
from finetuning import finetune
from tqdm.auto import tqdm

import transformers
from transformers import AutoConfig, set_seed
from transformers.trainer_utils import IntervalStrategy


lowercase   =	logging.getLogger(__name__)

lowercase   =	'''pytorch_model.bin'''





@dataclasses.dataclass
class    __A:
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    metadata={'''help''': '''Path to pretrained model or model identifier from huggingface.co/models.'''}							)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=UpperCAmelCase  ,		metadata={'''help''': '''Where do you want to store the pretrained models downloaded from huggingface.co.'''}  ,		)





@dataclasses.dataclass
class    __A:
					SCREAMING_SNAKE_CASE           =     dataclasses.field(metadata={'''help''': '''A csv or a json file containing the training data.'''}							)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(metadata={'''help''': '''A csv or a json file containing the data to predict on.'''}							)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=UpperCAmelCase  ,		metadata={'''help''': '''A csv or a json file containing the validation data.'''}							)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=UpperCAmelCase  ,		metadata={'''help''': '''The name of the task to train on.'''}  ,		)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=UpperCAmelCase  ,		metadata={'''help''': '''The list of labels for the task.'''}							)





@dataclasses.dataclass
class    __A:
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    metadata={'''help''': '''The output directory where the model predictions and checkpoints will be written.'''}							)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default='''accuracy'''  ,		metadata={'''help''': '''The evaluation metric used for the task.'''}							)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default='''no'''  ,		metadata={
					        '''help''': '''The evaluation strategy to adopt during training. Possible values are: ["no", "step", "epoch]'''
					    }  ,		)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=1_0  ,		metadata={'''help''': '''Number of evaluation calls with no improvement after which training will be stopped.'''}  ,		)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=0.0  ,		metadata={
					        '''help''': '''How much the specified evaluation metric must improve to satisfy early stopping conditions.'''
					    }  ,		)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=UpperCAmelCase  ,		metadata={'''help''': '''Whether to filter the pseudo-labeled data based on the confidence score.'''}  ,		)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=UpperCAmelCase  ,		metadata={'''help''': '''Whether to filter the pseudo-labeled data based on the validation performance.'''}  ,		)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=UpperCAmelCase  ,		metadata={'''help''': '''Whether to fine-tune on labeled data after pseudo training.'''}  ,		)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=0.0  ,		metadata={'''help''': '''Confidence threshold for pseudo-labeled data filtering.'''}  ,		)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=1_0_0  ,		metadata={'''help''': '''Number of evaluation calls with no improvement after which training will be stopped.'''}  ,		)
					SCREAMING_SNAKE_CASE           =     dataclasses.field(
					    default=UpperCAmelCase  ,		metadata={'''help''': '''Random seed for initialization.'''}  ,		)



def   __lowerCAmelCase    (				UpperCAmelCase__					:   Optional[int]      ,					UpperCAmelCase__					:   Tuple      ,					UpperCAmelCase__					:   int      ,					UpperCAmelCase__					:   Union[str, Any]      ,					UpperCAmelCase__					:   List[str]      ,					UpperCAmelCase__					:   str							)	->     List[str]:
							lowerCamelCase_					=       datasets.concatenate_datasets([infer_input, infer_output]      ,					axis=1							)

							if args.do_filter_by_confidence:
														lowerCamelCase_					=       dataset.filter(lambda UpperCAmelCase__							: example["probability"] > args.confidence_threshold							)

							if args.do_filter_by_val_performance:
														assert eval_result >= 0.0 and eval_result <= 1.0
														lowerCamelCase_					=       int(eval_result * len(UpperCAmelCase__							)							)
														print(UpperCAmelCase__							)
														lowerCamelCase_					=       dataset.sort("""probability"""      ,					reverse=UpperCAmelCase__							)
														lowerCamelCase_					=       dataset.select(range(UpperCAmelCase__							)							)

							lowerCamelCase_					=       dataset.remove_columns(["""label""", """probability"""]							)
							lowerCamelCase_					=       dataset.rename_column("""prediction"""      ,					"""label"""							)
							lowerCamelCase_					=       dataset.map(lambda UpperCAmelCase__							: {"label": idalabel[example["label"]]}							)
							lowerCamelCase_					=       dataset.shuffle(seed=args.seed							)

							lowerCamelCase_					=       os.path.join(UpperCAmelCase__      ,					F'''train_pseudo.{args.data_file_extension}'''							)
							if args.data_file_extension == "csv":
														dataset.to_csv(UpperCAmelCase__      ,					index=UpperCAmelCase__							)
							else:
														dataset.to_json(UpperCAmelCase__							)



def   __lowerCAmelCase    (				UpperCAmelCase__					:   Any      ,					UpperCAmelCase__					:   Any      ,					UpperCAmelCase__					:   Optional[int]      ,					UpperCAmelCase__					:   str      ,					**UpperCAmelCase__					:   Dict							)	->     Optional[int]:
							lowerCamelCase_					=       Accelerator()
							# Make one log on every process with the configuration for debugging.
							logging.basicConfig(
							    format="""%(asctime)s - %(levelname)s - %(name)s - %(message)s"""      ,					datefmt="""%m/%d/%Y %H:%M:%S"""      ,					level=logging.INFO      ,					)
							logger.info(accelerator.state							)

							# Setup logging, we only want one process per machine to log things on the
							# screen. accelerator.is_local_main_process is only True for one process per
							# machine.
							logger.setLevel(logging.INFO if accelerator.is_local_main_process else logging.ERROR							)

							if accelerator.is_local_main_process:
														datasets.utils.logging.set_verbosity_warning()
														transformers.utils.logging.set_verbosity_info()
							else:
														datasets.utils.logging.set_verbosity_error()
														transformers.utils.logging.set_verbosity_error()

							lowerCamelCase_					=       STModelArguments(model_name_or_path=UpperCAmelCase__							)
							lowerCamelCase_					=       STDataArguments(train_file=UpperCAmelCase__      ,					infer_file=UpperCAmelCase__							)
							lowerCamelCase_					=       STTrainingArguments(output_dir=UpperCAmelCase__							)
							lowerCamelCase_					=       argparse.Namespace()

							for arg_class in (model_args, data_args, training_args):
														for key, value in vars(UpperCAmelCase__							).items():
																					setattr(UpperCAmelCase__      ,					UpperCAmelCase__      ,					UpperCAmelCase__							)

							for key, value in kwargs.items():
														if hasattr(UpperCAmelCase__      ,					UpperCAmelCase__							):
																					setattr(UpperCAmelCase__      ,					UpperCAmelCase__      ,					UpperCAmelCase__							)

    # Sanity checks
							lowerCamelCase_					=       {}
							lowerCamelCase_					=       None

							# You need to provide the training data and the data to predict on
							assert args.train_file is not None
							assert args.infer_file is not None
							lowerCamelCase_					=       args.train_file
							lowerCamelCase_					=       args.infer_file

							if args.evaluation_strategy != IntervalStrategy.NO.value:
														assert args.eval_file is not None
														lowerCamelCase_					=       args.eval_file

							for key in data_files:
														lowerCamelCase_					=       data_files[key].split("""."""							)[-1]
														assert extension in ["csv", "json"], F'''`{key}_file` should be a csv or a json file.'''
														if args.data_file_extension is None:
																					lowerCamelCase_					=       extension
														else:
																					assert extension == args.data_file_extension, F'''`{key}_file` should be a {args.data_file_extension} file`.'''

							assert (
							    args.eval_metric in datasets.list_metrics()
							), F'''{args.eval_metric} not in the list of supported metrics {datasets.list_metrics()}.'''

							# If passed along, set the training seed now.
							if args.seed is not None:
														set_seed(args.seed							)

							logger.info("""Creating the initial data directory for self-training..."""							)
							lowerCamelCase_					=       F'''{args.output_dir}/self-train_iter-{{}}'''.format
							lowerCamelCase_					=       data_dir_format(0							)

							if accelerator.is_main_process:
														if args.output_dir is not None:
																					os.makedirs(args.output_dir      ,					exist_ok=UpperCAmelCase__							)
																					os.makedirs(UpperCAmelCase__      ,					exist_ok=UpperCAmelCase__							)
							accelerator.wait_for_everyone()

							lowerCamelCase_					=       None
							lowerCamelCase_					=       None
							lowerCamelCase_					=       0
							lowerCamelCase_					=       False
							# Show the progress bar
							lowerCamelCase_					=       tqdm(range(args.max_selftrain_iterations							)      ,					disable=not accelerator.is_local_main_process							)

							# Self-train
							for iteration in range(0      ,					int(args.max_selftrain_iterations							)							):
														lowerCamelCase_					=       data_dir_format(UpperCAmelCase__							)
														assert os.path.exists(UpperCAmelCase__							)

														# Stage 1: initial fine-tuning for iteration = 0 or pseudo-training for
														# iteration > 0
														lowerCamelCase_					=       os.path.join(UpperCAmelCase__      ,					"""stage-1"""							)
														lowerCamelCase_					=       {
														    """accelerator""": accelerator,
														    """model_name_or_path""": args.model_name_or_path,
														    """cache_dir""": args.cache_dir,
														    """do_train""": True,
														    """train_file""": data_files["""train"""] if iteration == 0 else data_files["""train_pseudo"""],
														    """do_eval""": True if args.eval_file is not None else False,
														    """eval_file""": data_files["""eval"""],
														    """do_predict""": True,
														    """infer_file""": data_files["""infer"""],
														    """task_name""": args.task_name,
														    """label_list""": args.label_list,
														    """output_dir""": current_output_dir,
														    """eval_metric""": args.eval_metric,
														    """evaluation_strategy""": args.evaluation_strategy,
														    """early_stopping_patience""": args.early_stopping_patience,
														    """early_stopping_threshold""": args.early_stopping_threshold,
														    """seed""": args.seed,
														}
														# Add additional training arguments
														for key, value in kwargs.items():
																					if key not in arguments_dict and not hasattr(UpperCAmelCase__      ,					UpperCAmelCase__							):
																												arguments_dict.update({key: value}							)

														lowerCamelCase_					=       os.path.join(UpperCAmelCase__      ,					"""best-checkpoint"""      ,					UpperCAmelCase__							)
														if os.path.exists(UpperCAmelCase__							):
																					logger.info(
																					    """Found existing model checkpoint at %s. Skipping self-training: iteration: %d, stage: 1."""      ,					UpperCAmelCase__      ,					UpperCAmelCase__      ,					)
														else:
																					logger.info("""***** Running self-training: iteration: %d, stage: 1 *****"""      ,					UpperCAmelCase__							)
																					finetune(**UpperCAmelCase__							)
																					accelerator.wait_for_everyone()
																					assert os.path.exists(UpperCAmelCase__							)
																					logger.info("""Self-training job completed: iteration: %d, stage: 1."""      ,					UpperCAmelCase__							)

														if iteration > 0 and args.finetune_on_labeled_data:
																					# Stage 2 (optional): fine-tuning on the original labeled data
																					lowerCamelCase_					=       os.path.join(UpperCAmelCase__      ,					"""best-checkpoint"""							)
																					lowerCamelCase_					=       os.path.join(UpperCAmelCase__      ,					"""stage-2"""							)
																					# Update arguments_dict
																					lowerCamelCase_					=       model_path
																					lowerCamelCase_					=       data_files["""train"""]
																					lowerCamelCase_					=       current_output_dir

																					lowerCamelCase_					=       os.path.join(UpperCAmelCase__      ,					"""best-checkpoint"""      ,					UpperCAmelCase__							)
																					if os.path.exists(UpperCAmelCase__							):
																												logger.info(
																												    """Found existing model checkpoint at %s. Skipping self-training: iteration: %d, stage: 2."""      ,					UpperCAmelCase__      ,					UpperCAmelCase__      ,					)
																					else:
																												logger.info("""***** Running self-training: iteration: %d, stage: 2 *****"""      ,					UpperCAmelCase__							)
																												finetune(**UpperCAmelCase__							)
																												accelerator.wait_for_everyone()
																												assert os.path.exists(UpperCAmelCase__							)
																												logger.info("""Self-training job completed: iteration: %d, stage: 2."""      ,					UpperCAmelCase__							)

														lowerCamelCase_					=       iteration
														lowerCamelCase_					=       data_dir_format(iteration + 1							)

														lowerCamelCase_					=       AutoConfig.from_pretrained(os.path.join(UpperCAmelCase__      ,					"""best-checkpoint"""							)							)
														lowerCamelCase_					=       config.idalabel
														lowerCamelCase_					=       os.path.join(UpperCAmelCase__      ,					"""eval_results_best-checkpoint.json"""							)
														lowerCamelCase_					=       os.path.join(UpperCAmelCase__      ,					"""test_results_best-checkpoint.json"""							)
														assert os.path.exists(UpperCAmelCase__							)

														with open(UpperCAmelCase__      ,					"""r"""							) as f:
																					lowerCamelCase_					=       float(json.load(UpperCAmelCase__							)[args.eval_metric]							)
														lowerCamelCase_					=       os.path.join(UpperCAmelCase__      ,					"""infer_output_best-checkpoint.csv"""							)
														assert os.path.exists(UpperCAmelCase__							)
														# Loading the dataset from local csv or json files.
														lowerCamelCase_					=       load_dataset(args.data_file_extension      ,					data_files={"""data""": data_files["""infer"""]}							)["""data"""]
														lowerCamelCase_					=       load_dataset("""csv"""      ,					data_files={"""data""": infer_output_file}							)["""data"""]

														if accelerator.is_main_process:
																					os.makedirs(UpperCAmelCase__      ,					exist_ok=UpperCAmelCase__							)
																					shutil.copy(UpperCAmelCase__      ,					os.path.join(UpperCAmelCase__      ,					F'''eval_results_iter-{iteration}.json'''							)							)
																					if os.path.exists(UpperCAmelCase__							):
																												shutil.copy(UpperCAmelCase__      ,					os.path.join(UpperCAmelCase__      ,					F'''test_results_iter-{iteration}.json'''							)							)
																					create_pseudo_labeled_data(UpperCAmelCase__      ,					UpperCAmelCase__      ,					UpperCAmelCase__      ,					UpperCAmelCase__      ,					UpperCAmelCase__      ,					UpperCAmelCase__							)
														accelerator.wait_for_everyone()

														lowerCamelCase_					=       os.path.join(UpperCAmelCase__      ,					F'''train_pseudo.{args.data_file_extension}'''							)

														if args.evaluation_strategy != IntervalStrategy.NO.value:
																					lowerCamelCase_					=       eval_result

																					if best_iteration is None:
																												lowerCamelCase_					=       new_iteration
																												lowerCamelCase_					=       new_eval_result
																					else:
																												if new_eval_result - best_eval_result > args.early_stopping_threshold:
																																			lowerCamelCase_					=       new_iteration
																																			lowerCamelCase_					=       new_eval_result
																																			lowerCamelCase_					=       0
																												else:
																																			if new_eval_result == best_eval_result:
																																										lowerCamelCase_					=       new_iteration
																																										lowerCamelCase_					=       new_eval_result
																																			early_stopping_patience_counter += 1

																												if early_stopping_patience_counter >= args.early_stopping_patience:
																																			lowerCamelCase_					=       True

														progress_bar.update(1							)

														if should_training_stop:
																					break

							if best_iteration is not None:
														# Save the best iteration
														logger.info("""Best iteration: %d"""      ,					UpperCAmelCase__							)
														logger.info("""Best evaluation result: %s = %f"""      ,					args.eval_metric      ,					UpperCAmelCase__							)
														accelerator.wait_for_everyone()
														if accelerator.is_main_process:
																					shutil.copy(
																					    os.path.join(UpperCAmelCase__      ,					F'''eval_results_iter-{iteration}.json'''							)      ,					os.path.join(UpperCAmelCase__      ,					"""eval_results_best-iteration.json"""							)      ,					)
							else:
														# Assume that the last iteration is the best
														logger.info("""Best iteration: %d"""      ,					args.max_selftrain_iterations - 1							)
														logger.info("""Best evaluation result: %s = %f"""      ,					args.eval_metric      ,					UpperCAmelCase__							)
														accelerator.wait_for_everyone()
														if accelerator.is_main_process:
																					shutil.copy(
																					    os.path.join(UpperCAmelCase__      ,					F'''eval_results_iter-{args.max_selftrain_iterations - 1}.json'''							)      ,					os.path.join(UpperCAmelCase__      ,					"""eval_results_best-iteration.json"""							)      ,					)







import collections
import inspect
import unittest

from transformers import FocalNetConfig
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_backbone_common import BackboneTesterMixin
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 (
		    FocalNetBackbone,
		    FocalNetForImageClassification,
		    FocalNetForMaskedImageModeling,
		    FocalNetModel,
		)
		from transformers.models.focalnet.modeling_focalnet import FOCALNET_PRETRAINED_MODEL_ARCHIVE_LIST

if is_vision_available():
		from PIL import Image

		from transformers import AutoImageProcessor





class    __A:

					def __init__(						self       :  Tuple ,       __UpperCamelCase       :  Optional[Any] ,       __UpperCamelCase       :  Optional[Any]=1_3 ,       __UpperCamelCase       :  str=3_2 ,       __UpperCamelCase       :  List[Any]=2 ,       __UpperCamelCase       :  Tuple=3 ,       __UpperCamelCase       :  List[Any]=1_6 ,       __UpperCamelCase       :  int=[3_2, 6_4, 1_2_8] ,       __UpperCamelCase       :  Tuple=[1, 2, 1] ,       __UpperCamelCase       :  List[Any]=[2, 2, 4] ,       __UpperCamelCase       :  int=2 ,       __UpperCamelCase       :  Tuple=2.0 ,       __UpperCamelCase       :  List[str]=True ,       __UpperCamelCase       :  List[Any]=0.0 ,       __UpperCamelCase       :  List[Any]=0.0 ,       __UpperCamelCase       :  Tuple=0.1 ,       __UpperCamelCase       :  Any="gelu" ,       __UpperCamelCase       :  int=False ,       __UpperCamelCase       :  List[Any]=True ,       __UpperCamelCase       :  Tuple=0.02 ,       __UpperCamelCase       :  Union[str, Any]=1E-5 ,       __UpperCamelCase       :  Dict=True ,       __UpperCamelCase       :  List[Any]=None ,       __UpperCamelCase       :  Optional[Any]=True ,       __UpperCamelCase       :  Optional[int]=1_0 ,       __UpperCamelCase       :  Union[str, Any]=8 ,       __UpperCamelCase       :  Optional[Any]=["stage1", "stage2"] ,       __UpperCamelCase       :  Tuple=[1, 2] ,       ):
												lowerCamelCase_					=       parent
												lowerCamelCase_					=       batch_size
												lowerCamelCase_					=       image_size
												lowerCamelCase_					=       patch_size
												lowerCamelCase_					=       num_channels
												lowerCamelCase_					=       embed_dim
												lowerCamelCase_					=       hidden_sizes
												lowerCamelCase_					=       depths
												lowerCamelCase_					=       num_heads
												lowerCamelCase_					=       window_size
												lowerCamelCase_					=       mlp_ratio
												lowerCamelCase_					=       qkv_bias
												lowerCamelCase_					=       hidden_dropout_prob
												lowerCamelCase_					=       attention_probs_dropout_prob
												lowerCamelCase_					=       drop_path_rate
												lowerCamelCase_					=       hidden_act
												lowerCamelCase_					=       use_absolute_embeddings
												lowerCamelCase_					=       patch_norm
												lowerCamelCase_					=       layer_norm_eps
												lowerCamelCase_					=       initializer_range
												lowerCamelCase_					=       is_training
												lowerCamelCase_					=       scope
												lowerCamelCase_					=       use_labels
												lowerCamelCase_					=       type_sequence_label_size
												lowerCamelCase_					=       encoder_stride
												lowerCamelCase_					=       out_features
												lowerCamelCase_					=       out_indices

					def  lowercase__   (						self       :  Tuple ):
												lowerCamelCase_					=       floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size] )

												lowerCamelCase_					=       None
												if self.use_labels:
																			lowerCamelCase_					=       ids_tensor([self.batch_size] ,       self.type_sequence_label_size )

												lowerCamelCase_					=       self.get_config()

												return config, pixel_values, labels

					def  lowercase__   (						self       :  Dict ):
												return FocalNetConfig(
												    image_size=self.image_size ,       patch_size=self.patch_size ,       num_channels=self.num_channels ,       embed_dim=self.embed_dim ,       hidden_sizes=self.hidden_sizes ,       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 ,       out_features=self.out_features ,       out_indices=self.out_indices ,       )

					def  lowercase__   (						self       :  List[str] ,       __UpperCamelCase       :  Union[str, Any] ,       __UpperCamelCase       :  Tuple ,       __UpperCamelCase       :  Union[str, Any] ):
												lowerCamelCase_					=       FocalNetModel(config=__UpperCamelCase )
												model.to(__UpperCamelCase )
												model.eval()
												lowerCamelCase_					=       model(__UpperCamelCase )

												lowerCamelCase_					=       ((config.image_size // config.patch_size) ** 2) // (4 ** (len(config.depths ) - 1))
												lowerCamelCase_					=       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  lowercase__   (						self       :  Optional[Any] ,       __UpperCamelCase       :  str ,       __UpperCamelCase       :  int ,       __UpperCamelCase       :  Union[str, Any] ):
												lowerCamelCase_					=       FocalNetBackbone(config=__UpperCamelCase )
												model.to(__UpperCamelCase )
												model.eval()
												lowerCamelCase_					=       model(__UpperCamelCase )

												# verify feature maps
												self.parent.assertEqual(len(result.feature_maps ) ,       len(config.out_features ) )
												self.parent.assertListEqual(list(result.feature_maps[0].shape ) ,       [self.batch_size, self.image_size, 8, 8] )

												# verify channels
												self.parent.assertEqual(len(model.channels ) ,       len(config.out_features ) )
												self.parent.assertListEqual(model.channels ,       config.hidden_sizes[:-1] )

												# verify backbone works with out_features=None
												lowerCamelCase_					=       None
												lowerCamelCase_					=       FocalNetBackbone(config=__UpperCamelCase )
												model.to(__UpperCamelCase )
												model.eval()
												lowerCamelCase_					=       model(__UpperCamelCase )

												# verify feature maps
												self.parent.assertEqual(len(result.feature_maps ) ,       1 )
												self.parent.assertListEqual(list(result.feature_maps[0].shape ) ,       [self.batch_size, self.image_size * 2, 4, 4] )

												# verify channels
												self.parent.assertEqual(len(model.channels ) ,       1 )
												self.parent.assertListEqual(model.channels ,       [config.hidden_sizes[-1]] )

					def  lowercase__   (						self       :  Optional[Any] ,       __UpperCamelCase       :  Union[str, Any] ,       __UpperCamelCase       :  str ,       __UpperCamelCase       :  Any ):
												lowerCamelCase_					=       FocalNetForMaskedImageModeling(config=__UpperCamelCase )
												model.to(__UpperCamelCase )
												model.eval()
												lowerCamelCase_					=       model(__UpperCamelCase )
												self.parent.assertEqual(
												    result.reconstruction.shape ,       (self.batch_size, self.num_channels, self.image_size, self.image_size) )

												# test greyscale images
												lowerCamelCase_					=       1
												lowerCamelCase_					=       FocalNetForMaskedImageModeling(__UpperCamelCase )
												model.to(__UpperCamelCase )
												model.eval()

												lowerCamelCase_					=       floats_tensor([self.batch_size, 1, self.image_size, self.image_size] )
												lowerCamelCase_					=       model(__UpperCamelCase )
												self.parent.assertEqual(result.reconstruction.shape ,       (self.batch_size, 1, self.image_size, self.image_size) )

					def  lowercase__   (						self       :  str ,       __UpperCamelCase       :  str ,       __UpperCamelCase       :  Optional[int] ,       __UpperCamelCase       :  Union[str, Any] ):
												lowerCamelCase_					=       self.type_sequence_label_size
												lowerCamelCase_					=       FocalNetForImageClassification(__UpperCamelCase )
												model.to(__UpperCamelCase )
												model.eval()
												lowerCamelCase_					=       model(__UpperCamelCase ,       labels=__UpperCamelCase )
												self.parent.assertEqual(result.logits.shape ,       (self.batch_size, self.type_sequence_label_size) )

												# test greyscale images
												lowerCamelCase_					=       1
												lowerCamelCase_					=       FocalNetForImageClassification(__UpperCamelCase )
												model.to(__UpperCamelCase )
												model.eval()

												lowerCamelCase_					=       floats_tensor([self.batch_size, 1, self.image_size, self.image_size] )
												lowerCamelCase_					=       model(__UpperCamelCase )
												self.parent.assertEqual(result.logits.shape ,       (self.batch_size, self.type_sequence_label_size) )



					def  lowercase__   (						self       :  Tuple ):
												lowerCamelCase_					=       self.prepare_config_and_inputs()

												lowerCamelCase_	, lowerCamelCase_	, lowerCamelCase_					=       config_and_inputs
												lowerCamelCase_					=       {"""pixel_values""": pixel_values}
												return config, inputs_dict





@require_torch
class    __A(					UpperCAmelCase  ,		UpperCAmelCase  ,		unittest.TestCase							):
					SCREAMING_SNAKE_CASE           =     (
					    (
					        FocalNetModel,
					        FocalNetForImageClassification,
					        FocalNetForMaskedImageModeling,
					        FocalNetBackbone,
					    )
					    if is_torch_available()
					    else ()
					)
					SCREAMING_SNAKE_CASE           =     (
					    {'''feature-extraction''': FocalNetModel, '''image-classification''': FocalNetForImageClassification}
					    if is_torch_available()
					    else {}
					)
					SCREAMING_SNAKE_CASE           =     False

					SCREAMING_SNAKE_CASE           =     False
					SCREAMING_SNAKE_CASE           =     False
					SCREAMING_SNAKE_CASE           =     False
					SCREAMING_SNAKE_CASE           =     False

					def  lowercase__   (						self       :  str ):
												lowerCamelCase_					=       FocalNetModelTester(self )
												lowerCamelCase_					=       ConfigTester(self ,       config_class=__UpperCamelCase ,       embed_dim=3_7 ,       has_text_modality=__UpperCamelCase )

					def  lowercase__   (						self       :  Optional[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  lowercase__   (						self       :  Tuple ):
												return

					def  lowercase__   (						self       :  Optional[int] ):
												lowerCamelCase_					=       self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_model(*__UpperCamelCase )

					def  lowercase__   (						self       :  Any ):
												lowerCamelCase_					=       self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_backbone(*__UpperCamelCase )

					def  lowercase__   (						self       :  List[Any] ):
												lowerCamelCase_					=       self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_for_masked_image_modeling(*__UpperCamelCase )

					def  lowercase__   (						self       :  List[Any] ):
												lowerCamelCase_					=       self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_for_image_classification(*__UpperCamelCase )

					@unittest.skip(reason="""FocalNet does not use inputs_embeds""" )
					def  lowercase__   (						self       :  str ):
												pass

					@unittest.skip(reason="""FocalNet does not use feedforward chunking""" )
					def  lowercase__   (						self       :  List[str] ):
												pass

					def  lowercase__   (						self       :  Optional[int] ):
												lowerCamelCase_	, lowerCamelCase_					=       self.model_tester.prepare_config_and_inputs_for_common()

												for model_class in self.all_model_classes[:-1]:
																			lowerCamelCase_					=       model_class(__UpperCamelCase )
																			self.assertIsInstance(model.get_input_embeddings() ,       (nn.Module) )
																			lowerCamelCase_					=       model.get_output_embeddings()
																			self.assertTrue(x is None or isinstance(__UpperCamelCase ,       nn.Linear ) )

					def  lowercase__   (						self       :  Tuple ):
												lowerCamelCase_	, lowerCamelCase_					=       self.model_tester.prepare_config_and_inputs_for_common()

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

																			lowerCamelCase_					=       ["""pixel_values"""]
																			self.assertListEqual(arg_names[:1] ,       __UpperCamelCase )

					def  lowercase__   (						self       :  Dict ,       __UpperCamelCase       :  Tuple ,       __UpperCamelCase       :  List[str] ,       __UpperCamelCase       :  str ,       __UpperCamelCase       :  Optional[Any] ):
												lowerCamelCase_					=       model_class(__UpperCamelCase )
												model.to(__UpperCamelCase )
												model.eval()

												with torch.no_grad():
																			lowerCamelCase_					=       model(**self._prepare_for_class(__UpperCamelCase ,       __UpperCamelCase ) )

												lowerCamelCase_					=       outputs.hidden_states

												lowerCamelCase_					=       getattr(
												    self.model_tester ,       """expected_num_hidden_layers""" ,       len(self.model_tester.depths ) + 1 )
												self.assertEqual(len(__UpperCamelCase ) ,       __UpperCamelCase )

												# FocalNet has a different seq_length
												lowerCamelCase_					=       (
												    config.patch_size
												    if isinstance(config.patch_size ,       collections.abc.Iterable )
												    else (config.patch_size, config.patch_size)
												)

												lowerCamelCase_					=       (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] ,       )

												lowerCamelCase_					=       outputs.reshaped_hidden_states
												self.assertEqual(len(__UpperCamelCase ) ,       __UpperCamelCase )

												lowerCamelCase_	, lowerCamelCase_	, lowerCamelCase_	, lowerCamelCase_					=       reshaped_hidden_states[0].shape
												lowerCamelCase_					=       (
												    reshaped_hidden_states[0].view(__UpperCamelCase ,       __UpperCamelCase ,       height * width ).permute(0 ,       2 ,       1 )
												)
												self.assertListEqual(
												    list(reshaped_hidden_states.shape[-2:] ) ,       [num_patches, self.model_tester.embed_dim] ,       )

					def  lowercase__   (						self       :  int ):
												lowerCamelCase_	, lowerCamelCase_					=       self.model_tester.prepare_config_and_inputs_for_common()

												lowerCamelCase_					=       (
												    self.model_tester.image_size
												    if isinstance(self.model_tester.image_size ,       collections.abc.Iterable )
												    else (self.model_tester.image_size, self.model_tester.image_size)
												)

												for model_class in self.all_model_classes[:-1]:
																			lowerCamelCase_					=       True
																			self.check_hidden_states_output(__UpperCamelCase ,       __UpperCamelCase ,       __UpperCamelCase ,       __UpperCamelCase )

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

																			self.check_hidden_states_output(__UpperCamelCase ,       __UpperCamelCase ,       __UpperCamelCase ,       __UpperCamelCase )

					def  lowercase__   (						self       :  List[str] ):
												lowerCamelCase_	, lowerCamelCase_					=       self.model_tester.prepare_config_and_inputs_for_common()
												lowerCamelCase_					=       3

												lowerCamelCase_					=       (
												    self.model_tester.image_size
												    if isinstance(self.model_tester.image_size ,       collections.abc.Iterable )
												    else (self.model_tester.image_size, self.model_tester.image_size)
												)
												lowerCamelCase_					=       (
												    config.patch_size
												    if isinstance(config.patch_size ,       collections.abc.Iterable )
												    else (config.patch_size, config.patch_size)
												)

												lowerCamelCase_					=       image_size[0] + patch_size[0] - (image_size[0] % patch_size[0])
												lowerCamelCase_					=       image_size[1] + patch_size[1] - (image_size[1] % patch_size[1])

												for model_class in self.all_model_classes[:-1]:
																			lowerCamelCase_					=       True
																			self.check_hidden_states_output(__UpperCamelCase ,       __UpperCamelCase ,       __UpperCamelCase ,       (padded_height, padded_width) )

																			# check that output_hidden_states also work using config
																			del inputs_dict["output_hidden_states"]
																			lowerCamelCase_					=       True
																			self.check_hidden_states_output(__UpperCamelCase ,       __UpperCamelCase ,       __UpperCamelCase ,       (padded_height, padded_width) )

					@slow
					def  lowercase__   (						self       :  Dict ):
												for model_name in FOCALNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																			lowerCamelCase_					=       FocalNetModel.from_pretrained(__UpperCamelCase )
																			self.assertIsNotNone(__UpperCamelCase )



					def  lowercase__   (						self       :  int ):
												lowerCamelCase_	, lowerCamelCase_					=       self.model_tester.prepare_config_and_inputs_for_common()

												lowerCamelCase_					=       _config_zero_init(__UpperCamelCase )
												for model_class in self.all_model_classes:
																			lowerCamelCase_					=       model_class(config=__UpperCamelCase )
																			for name, param in model.named_parameters():
																										if "embeddings" not in name and 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''' ,       )





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

					@cached_property
					def  lowercase__   (						self       :  int ):
												# TODO update organization
												return AutoImageProcessor.from_pretrained("""microsoft/focalnet-tiny""" ) if is_vision_available() else None



					@slow
					def  lowercase__   (						self       :  str ):
												lowerCamelCase_					=       FocalNetForImageClassification.from_pretrained("""microsoft/focalnet-tiny""" ).to(__UpperCamelCase )
												lowerCamelCase_					=       self.default_image_processor

												lowerCamelCase_					=       Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png""" )
												lowerCamelCase_					=       image_processor(images=__UpperCamelCase ,       return_tensors="""pt""" ).to(__UpperCamelCase )

												# forward pass
												with torch.no_grad():
																			lowerCamelCase_					=       model(**__UpperCamelCase )

												# verify the logits
												lowerCamelCase_					=       torch.Size((1, 1_0_0_0) )
												self.assertEqual(outputs.logits.shape ,       __UpperCamelCase )
												lowerCamelCase_					=       torch.tensor([0.2166, -0.4368, 0.2191] ).to(__UpperCamelCase )
												self.assertTrue(torch.allclose(outputs.logits[0, :3] ,       __UpperCamelCase ,       atol=1E-4 ) )
												self.assertTrue(outputs.logits.argmax(dim=-1 ).item() ,       2_8_1 )





@require_torch
class    __A(					UpperCAmelCase  ,		unittest.TestCase							):
					SCREAMING_SNAKE_CASE           =     (FocalNetBackbone,) if is_torch_available() else ()
					SCREAMING_SNAKE_CASE           =     FocalNetConfig

					SCREAMING_SNAKE_CASE           =     False

					def  lowercase__   (						self       :  Any ):
												lowerCamelCase_					=       FocalNetModelTester(self )





