
import gzip
import hashlib
import json
import multiprocessing
import os
import re
import shutil
import time
from pathlib import Path

import numpy as np
from arguments import PreprocessingArguments
from datasets import load_dataset
from minhash_deduplication import deduplicate_dataset

from transformers import AutoTokenizer, HfArgumentParser


__snake_case:  Union[str, Any]								=			re.compile(R"\s+")






def 	_snake_case						(      A_       :    int   ):
  """simple docstring"""







  return {"hash": hashlib.mda(re.sub(A_     ,		""""""     ,		example["""content"""]   ).encode("""utf-8"""   )   ).hexdigest()}






def 	_snake_case						(      A_       :    List[str]   ):
  """simple docstring"""







  a_      :	Tuple      =							[len(A_   ) for line in example["""content"""].splitlines()]
  return {"line_mean": np.mean(A_   ), "line_max": max(A_   )}






def 	_snake_case						(      A_       :    Dict   ):
  """simple docstring"""







  a_      :	int      =							np.mean([c.isalnum() for c in example["""content"""]]   )
  return {"alpha_frac": alpha_frac}






def 	_snake_case						(      A_       :    Optional[int]     ,		A_       :    List[Any]   ):
  """simple docstring"""







  if example["hash"] in uniques:
    uniques.remove(example["""hash"""]   )
    return True
  else:
    return False






def 	_snake_case						(      A_       :    Dict     ,		A_       :    Dict=5   ):
  """simple docstring"""







  a_      :	Tuple      =							["""auto-generated""", """autogenerated""", """automatically generated"""]
  a_      :	Any      =							example["""content"""].splitlines()
  for _, line in zip(range(A_   )     ,		A_   ):
    for keyword in keywords:
      if keyword in line.lower():
        return {"autogenerated": True}
  else:
    return {"autogenerated": False}






def 	_snake_case						(      A_       :    Tuple     ,		A_       :    str=5     ,		A_       :    Union[str, Any]=0.05   ):
  """simple docstring"""







  a_      :	Optional[int]      =							["""unit tests""", """test file""", """configuration file"""]
  a_      :	Tuple      =							example["""content"""].splitlines()
  a_      :	List[str]      =							0
  a_      :	Dict      =							0
  # first test
  for _, line in zip(range(A_   )     ,		A_   ):
    for keyword in keywords:
      if keyword in line.lower():
        return {"config_or_test": True}
    # second test
  a_      :	Optional[Any]      =							example["""content"""].count("""\n"""   )
  a_      :	Any      =							int(coeff * nlines   )
  for line in lines:
    count_config += line.lower().count("""config"""   )
    count_test += line.lower().count("""test"""   )
    if count_config > threshold or count_test > threshold:
      return {"config_or_test": True}
  return {"config_or_test": False}






def 	_snake_case						(      A_       :    Any   ):
  """simple docstring"""







  a_      :	Any      =							["""def """, """class """, """for """, """while """]
  a_      :	Any      =							example["""content"""].splitlines()
  for line in lines:
    for keyword in keywords:
      if keyword in line.lower():
        return {"has_no_keywords": False}
  return {"has_no_keywords": True}






def 	_snake_case						(      A_       :    Tuple     ,		A_       :    List[Any]=4   ):
  """simple docstring"""







  a_      :	int      =							example["""content"""].splitlines()
  a_      :	Any      =							0
  for line in lines:
    counter += line.lower().count("""="""   )
    if counter > minimum:
      return {"has_few_assignments": False}
  return {"has_few_assignments": True}






def 	_snake_case						(      A_       :    Optional[Any]   ):
  """simple docstring"""







  a_      :	List[str]      =							tokenizer(example["""content"""]     ,		truncation=A_   )["""input_ids"""]
  a_      :	str      =							len(example["""content"""]   ) / len(A_   )
  return {"ratio": ratio}






def 	_snake_case						(      A_       :    Union[str, Any]   ):
  """simple docstring"""







  a_      :	Tuple      =							{}
  results.update(get_hash(A_   )   )
  results.update(line_stats(A_   )   )
  results.update(alpha_stats(A_   )   )
  results.update(char_token_ratio(A_   )   )
  results.update(is_autogenerated(A_   )   )
  results.update(is_config_or_test(A_   )   )
  results.update(has_no_keywords(A_   )   )
  results.update(has_few_assignments(A_   )   )
  return results






def 	_snake_case						(      A_       :    Union[str, Any]     ,		A_       :    Tuple     ,		A_       :    List[Any]   ):
  """simple docstring"""







  if not check_uniques(A_     ,		A_   ):
    return False
  elif example["autogenerated"]:
    return False
  elif example["line_max"] > args.line_max:
    return False
  elif example["line_mean"] > args.line_mean:
    return False
  elif example["alpha_frac"] < args.alpha_frac:
    return False
  elif example["ratio"] < args.min_token_ratio:
    return False
  elif example["config_or_test"] and np.random.rand() <= args.filter_proba:
    return False
  elif example["has_no_keywords"] and np.random.rand() <= args.filter_proba:
    return False
  elif example["has_few_assignments"]:
    return False
  else:
    return True






def 	_snake_case						(      A_       :    Any   ):
  """simple docstring"""







  with open(A_     ,		"""rb"""   ) as f_in:
    with gzip.open(str(A_   ) + """.gz"""     ,		"""wb"""     ,		compresslevel=6   ) as f_out:
      shutil.copyfileobj(A_     ,		A_   )
  os.unlink(A_   )


# Settings
__snake_case:  int								=			HfArgumentParser(PreprocessingArguments)
__snake_case:  Tuple								=			parser.parse_args()
if args.num_workers is None:
    __snake_case:  Union[str, Any]								=			multiprocessing.cpu_count()
__snake_case:  str								=			AutoTokenizer.from_pretrained(args.tokenizer_dir)

# Load dataset
__snake_case:  Union[str, Any]								=			time.time()
__snake_case:  Any								=			load_dataset(args.dataset_name, split="train")
print(F"""Time to load dataset: {time.time()-t_start:.2f}""")

# Run preprocessing
__snake_case:  Optional[Any]								=			time.time()
__snake_case:  Any								=			ds.map(preprocess, num_proc=args.num_workers)
print(F"""Time to preprocess dataset: {time.time()-t_start:.2f}""")

# Deduplicate hashes
__snake_case:  List[str]								=			set(ds.unique("hash"))
__snake_case:  Optional[Any]								=			len(uniques) / len(ds)
print(F"""Fraction of duplicates: {1-frac:.2%}""")

# Deduplicate data and apply heuristics
__snake_case:  Tuple								=			time.time()
__snake_case:  Optional[int]								=			ds.filter(filter, fn_kwargs={"uniques": uniques, "args": args})
print(F"""Time to filter dataset: {time.time()-t_start:.2f}""")
print(F"""Size of filtered dataset: {len(ds_filter)}""")

# Deduplicate with minhash and jaccard similarity
if args.near_deduplication:
    __snake_case:  Union[str, Any]								=			time.time()
    __snake_case							,__snake_case:  str								=			deduplicate_dataset(ds_filter, args.jaccard_threshold)
    print(F"""Time to deduplicate dataset: {time.time()-t_start:.2f}""")
    print(F"""Size of deduplicate dataset: {len(ds_filter)}""")

# Save data in batches of samples_per_file
__snake_case:  str								=			Path(args.output_dir)
output_dir.mkdir(exist_ok=True)

# save duplicate_clusters in the output_dir as artifacts
# not sure it is the right place the save it
if args.near_deduplication:
    with open(output_dir / "duplicate_clusters.json", "w") as f:
        json.dump(duplicate_clusters, f)

__snake_case:  Optional[Any]								=			output_dir / "data"
data_dir.mkdir(exist_ok=True)

__snake_case:  Any								=			time.time()
for file_number, index in enumerate(range(0, len(ds_filter), args.samples_per_file)):
    __snake_case:  Any								=			str(data_dir / F"""file-{file_number+1:012}.json""")
    __snake_case:  Optional[Any]								=			min(len(ds_filter), index + args.samples_per_file)
    ds_filter.select(list(range(index, end_index))).to_json(file_path)
    compress_file(file_path)




print(F"""Time to save dataset: {time.time()-t_start:.2f}""")




'''simple docstring'''

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  _UpperCAmelCase :

     """simple docstring"""

     def __init__(      self							,		lowerCAmelCase_							,		lowerCAmelCase_=13							,		lowerCAmelCase_=32							,		lowerCAmelCase_=2							,		lowerCAmelCase_=3							,		lowerCAmelCase_=16							,		lowerCAmelCase_=[32, 64, 1_28]							,		lowerCAmelCase_=[1, 2, 1]							,		lowerCAmelCase_=[2, 2, 4]							,		lowerCAmelCase_=2							,		lowerCAmelCase_=2.0							,		lowerCAmelCase_=True							,		lowerCAmelCase_=0.0							,		lowerCAmelCase_=0.0							,		lowerCAmelCase_=0.1							,		lowerCAmelCase_="gelu"							,		lowerCAmelCase_=False							,		lowerCAmelCase_=True							,		lowerCAmelCase_=0.02							,		lowerCAmelCase_=1E-5							,		lowerCAmelCase_=True							,		lowerCAmelCase_=None							,		lowerCAmelCase_=True							,		lowerCAmelCase_=10							,		lowerCAmelCase_=8							,		lowerCAmelCase_=["stage1", "stage2"]							,		lowerCAmelCase_=[1, 2]							,		):

       '''simple docstring'''

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


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	int      =							floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size] )

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

       a_      :	Any      =							self.get_config()

       return config, pixel_values, labels


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       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 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							FocalNetModel(config=lowerCAmelCase_ )
       model.to(lowerCAmelCase_ )
       model.eval()
       a_      :	Optional[Any]      =							model(lowerCAmelCase_ )

       a_      :	Union[str, Any]      =							((config.image_size // config.patch_size) ** 2) // (4 ** (len(config.depths ) - 1))
       a_      :	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							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							FocalNetBackbone(config=lowerCAmelCase_ )
       model.to(lowerCAmelCase_ )
       model.eval()
       a_      :	Tuple      =							model(lowerCAmelCase_ )

       # 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
       a_      :	Optional[Any]      =							None
       a_      :	List[Any]      =							FocalNetBackbone(config=lowerCAmelCase_ )
       model.to(lowerCAmelCase_ )
       model.eval()
       a_      :	List[str]      =							model(lowerCAmelCase_ )

       # 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 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							FocalNetForMaskedImageModeling(config=lowerCAmelCase_ )
       model.to(lowerCAmelCase_ )
       model.eval()
       a_      :	str      =							model(lowerCAmelCase_ )
       self.parent.assertEqual(
           result.reconstruction.shape							,		(self.batch_size, self.num_channels, self.image_size, self.image_size) )

       # test greyscale images
       a_      :	List[str]      =							1
       a_      :	Optional[Any]      =							FocalNetForMaskedImageModeling(lowerCAmelCase_ )
       model.to(lowerCAmelCase_ )
       model.eval()

       a_      :	Tuple      =							floats_tensor([self.batch_size, 1, self.image_size, self.image_size] )
       a_      :	int      =							model(lowerCAmelCase_ )
       self.parent.assertEqual(result.reconstruction.shape							,		(self.batch_size, 1, self.image_size, self.image_size) )


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	Any      =							self.type_sequence_label_size
       a_      :	str      =							FocalNetForImageClassification(lowerCAmelCase_ )
       model.to(lowerCAmelCase_ )
       model.eval()
       a_      :	List[Any]      =							model(lowerCAmelCase_							,		labels=lowerCAmelCase_ )
       self.parent.assertEqual(result.logits.shape							,		(self.batch_size, self.type_sequence_label_size) )

       # test greyscale images
       a_      :	Optional[Any]      =							1
       a_      :	Dict      =							FocalNetForImageClassification(lowerCAmelCase_ )
       model.to(lowerCAmelCase_ )
       model.eval()

       a_      :	List[Any]      =							floats_tensor([self.batch_size, 1, self.image_size, self.image_size] )
       a_      :	List[Any]      =							model(lowerCAmelCase_ )
       self.parent.assertEqual(result.logits.shape							,		(self.batch_size, self.type_sequence_label_size) )






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	int      =							self.prepare_config_and_inputs()

       a_      , a_      , a_      :	List[str]      =							config_and_inputs
       a_      :	Optional[Any]      =							{"""pixel_values""": pixel_values}
       return config, inputs_dict







@require_torch
class  _UpperCAmelCase ( lowerCAmelCase__	,lowerCAmelCase__	,unittest.TestCase    ):

     """simple docstring"""

     a_    	=					(
         (
             FocalNetModel,
             FocalNetForImageClassification,
             FocalNetForMaskedImageModeling,
             FocalNetBackbone,
         )
         if is_torch_available()
         else ()
     )
     a_    	=					(
         {"feature-extraction": FocalNetModel, "image-classification": FocalNetForImageClassification}
         if is_torch_available()
         else {}
     )
     a_    	=					False

     a_    	=					False
     a_    	=					False
     a_    	=					False
     a_    	=					False


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	List[str]      =							FocalNetModelTester(self )
       a_      :	Union[str, Any]      =							ConfigTester(self							,		config_class=lowerCAmelCase_							,		embed_dim=37							,		has_text_modality=lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       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 ):

       '''simple docstring'''

       return


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Tuple      =							self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	List[Any]      =							self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_backbone(*lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_masked_image_modeling(*lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	str      =							self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_image_classification(*lowerCAmelCase_ )


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

       '''simple docstring'''

       pass


     @unittest.skip(reason="""FocalNet does not use feedforward chunking""" )
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       pass


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      :	List[str]      =							self.model_tester.prepare_config_and_inputs_for_common()

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


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      :	Union[str, Any]      =							self.model_tester.prepare_config_and_inputs_for_common()

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

         a_      :	Optional[int]      =							["""pixel_values"""]
         self.assertListEqual(arg_names[:1]							,		lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	str      =							model_class(lowerCAmelCase_ )
       model.to(lowerCAmelCase_ )
       model.eval()

       with torch.no_grad():
         a_      :	str      =							model(**self._prepare_for_class(lowerCAmelCase_							,		lowerCAmelCase_ ) )

       a_      :	Optional[Any]      =							outputs.hidden_states

       a_      :	int      =							getattr(
           self.model_tester							,		"""expected_num_hidden_layers"""							,		len(self.model_tester.depths ) + 1 )
       self.assertEqual(len(lowerCAmelCase_ )							,		lowerCAmelCase_ )

       # FocalNet has a different seq_length
       a_      :	List[str]      =							(
           config.patch_size
           if isinstance(config.patch_size							,		collections.abc.Iterable )
           else (config.patch_size, config.patch_size)
       )

       a_      :	List[Any]      =							(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]							,		)

       a_      :	str      =							outputs.reshaped_hidden_states
       self.assertEqual(len(lowerCAmelCase_ )							,		lowerCAmelCase_ )

       a_      , a_      , a_      , a_      :	Optional[Any]      =							reshaped_hidden_states[0].shape
       a_      :	Dict      =							(
           reshaped_hidden_states[0].view(lowerCAmelCase_							,		lowerCAmelCase_							,		height * width ).permute(0							,		2							,		1 )
       )
       self.assertListEqual(
           list(reshaped_hidden_states.shape[-2:] )							,		[num_patches, self.model_tester.embed_dim]							,		)


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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

       a_      :	Optional[Any]      =							(
           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]:
         a_      :	Any      =							True
         self.check_hidden_states_output(lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ )

         # check that output_hidden_states also work using config
         del inputs_dict["output_hidden_states"]
         a_      :	Optional[int]      =							True

         self.check_hidden_states_output(lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      :	Any      =							self.model_tester.prepare_config_and_inputs_for_common()
       a_      :	Optional[Any]      =							3

       a_      :	Union[str, Any]      =							(
           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)
       )
       a_      :	Optional[int]      =							(
           config.patch_size
           if isinstance(config.patch_size							,		collections.abc.Iterable )
           else (config.patch_size, config.patch_size)
       )

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

       for model_class in self.all_model_classes[:-1]:
         a_      :	List[str]      =							True
         self.check_hidden_states_output(lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		(padded_height, padded_width) )

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


     @slow
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       for model_name in FOCALNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
         a_      :	List[Any]      =							FocalNetModel.from_pretrained(lowerCAmelCase_ )
         self.assertIsNotNone(lowerCAmelCase_ )






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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

       a_      :	Union[str, Any]      =							_config_zero_init(lowerCAmelCase_ )
       for model_class in self.all_model_classes:
         a_      :	Dict      =							model_class(config=lowerCAmelCase_ )
         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  _UpperCAmelCase ( unittest.TestCase    ):

     """simple docstring"""

     @cached_property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return AutoImageProcessor.from_pretrained("""microsoft/focalnet-tiny""" ) if is_vision_available() else None






     @slow
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Optional[Any]      =							FocalNetForImageClassification.from_pretrained("""microsoft/focalnet-tiny""" ).to(lowerCAmelCase_ )
       a_      :	Dict      =							self.default_image_processor

       a_      :	int      =							Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png""" )
       a_      :	List[Any]      =							image_processor(images=lowerCAmelCase_							,		return_tensors="""pt""" ).to(lowerCAmelCase_ )

       # forward pass
       with torch.no_grad():
         a_      :	int      =							model(**lowerCAmelCase_ )

       # verify the logits
       a_      :	Optional[Any]      =							torch.Size((1, 10_00) )
       self.assertEqual(outputs.logits.shape							,		lowerCAmelCase_ )
       a_      :	Dict      =							torch.tensor([0.2166, -0.4368, 0.2191] ).to(lowerCAmelCase_ )
       self.assertTrue(torch.allclose(outputs.logits[0, :3]							,		lowerCAmelCase_							,		atol=1E-4 ) )
       self.assertTrue(outputs.logits.argmax(dim=-1 ).item()							,		2_81 )







@require_torch
class  _UpperCAmelCase ( lowerCAmelCase__	,unittest.TestCase    ):

     """simple docstring"""

     a_    	=					(FocalNetBackbone,) if is_torch_available() else ()
     a_    	=					FocalNetConfig

     a_    	=					False


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	int      =							FocalNetModelTester(self )


