

import logging
import os
import time
from argparse import Namespace
from pathlib import Path

import datasets
import torch
from accelerate import Accelerator, DistributedType
from accelerate.utils import ProjectConfiguration
from arguments import TrainingArguments
from datasets import load_dataset
from huggingface_hub import Repository
from torch.optim import AdamW
from torch.utils.data import IterableDataset
from torch.utils.data.dataloader import DataLoader
from torch.utils.data.datapipes.iter.combinatorics import ShufflerIterDataPipe

import transformers
from transformers import AutoModelForCausalLM, AutoTokenizer, HfArgumentParser, get_scheduler, set_seed







class     a__    (				UpperCAmelCase    ):







  """simple docstring"""




  def __init__(     self      :	Any			,			UpperCAmelCase__      :	List[str]			,			UpperCAmelCase__      :	str			,			UpperCAmelCase__      :	str=False			,			UpperCAmelCase__      :	Union[str, Any]=1_0_2_4			,			UpperCAmelCase__      :	Tuple=1_0_2_4			,			UpperCAmelCase__      :	List[Any]=3.6			,			UpperCAmelCase__      :	Optional[Any]=False			,			)   ->str:
        """simple docstring"""







        SCREAMING_SNAKE_CASE	: Any					       =				tokenizer
        SCREAMING_SNAKE_CASE	: int					       =				tokenizer.bos_token_id
        SCREAMING_SNAKE_CASE	: Dict					       =				dataset
        SCREAMING_SNAKE_CASE	: Any					       =				seq_length
        SCREAMING_SNAKE_CASE	: Union[str, Any]					       =				0
        SCREAMING_SNAKE_CASE	: Optional[int]					       =				infinite
        SCREAMING_SNAKE_CASE	: str					       =				0
        SCREAMING_SNAKE_CASE	: Dict					       =				tokenized

        if self.tokenized:
              SCREAMING_SNAKE_CASE	: str					       =				seq_length * num_of_sequences
              SCREAMING_SNAKE_CASE	: Union[str, Any]					       =				"""input_ids"""
        else:
              SCREAMING_SNAKE_CASE	: Dict					       =				seq_length * chars_per_token * num_of_sequences
              SCREAMING_SNAKE_CASE	: Any					       =				"""content"""


  def __iter__(     self      :	List[str]						)   ->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE	: str					       =				iter(self.dataset						)
        SCREAMING_SNAKE_CASE	: List[str]					       =				True
        while more_examples:
              SCREAMING_SNAKE_CASE       ,		SCREAMING_SNAKE_CASE	: Dict					       =				[], 0
              while True:
                    if buffer_len >= self.max_buffer_size:
                          break
                    try:
                          buffer.append(next(UpperCAmelCase__						)[self.content_field]						)
                          buffer_len += len(buffer[-1]						)
                    except StopIteration:
                          if self.infinite:
                                SCREAMING_SNAKE_CASE	: Optional[Any]					       =				iter(self.dataset						)
                                self.epoch += 1
                                logger.info(f"Dataset epoch: {self.epoch}"						)
                          else:
                                SCREAMING_SNAKE_CASE	: List[str]					       =				False
                                break
              if self.tokenized:
                    SCREAMING_SNAKE_CASE	: Optional[Any]					       =				buffer
              else:
                    SCREAMING_SNAKE_CASE	: str					       =				self.tokenizer(UpperCAmelCase__			,			truncation=UpperCAmelCase__						)["""input_ids"""]
              SCREAMING_SNAKE_CASE	: str					       =				[]
              for tokenized_input in tokenized_inputs:
                    all_token_ids.extend(tokenized_input + [self.concat_token_id]						)
              for i in range(0			,			len(UpperCAmelCase__						)			,			self.seq_length						):
                    SCREAMING_SNAKE_CASE	: Optional[Any]					       =				all_token_ids[i : i + self.seq_length]
                    if len(UpperCAmelCase__						) == self.seq_length:
                          self.current_size += 1
                          yield torch.tensor(UpperCAmelCase__						)




  def 	_lowercase (     self      :	Dict			,			UpperCAmelCase__      :	List[str]=1_0_0_0						)   ->List[Any]:
        """simple docstring"""







        return ShufflerIterDataPipe(self			,			buffer_size=UpperCAmelCase__						)






def    __lowercase (       _A     )       ->  int:
      SCREAMING_SNAKE_CASE	: Optional[Any]					       =				args.model_ckpt.split("""/"""     )[-1]
      SCREAMING_SNAKE_CASE	: Any					       =				logging.getLogger(__name__     )
      SCREAMING_SNAKE_CASE	: int					       =				Path(args.save_dir     ) / """log/"""
      log_dir.mkdir(exist_ok=_A     )
      SCREAMING_SNAKE_CASE	: Optional[Any]					       =				F"debug_{accelerator.process_index}.log"
      logging.basicConfig(
          format="""%(asctime)s - %(levelname)s - %(name)s - %(message)s"""  ,      datefmt="""%m/%d/%Y %H:%M:%S"""  ,      level=logging.INFO  ,      handlers=[logging.FileHandler(log_dir / filename     ), logging.StreamHandler()]  ,      )
      if accelerator.is_main_process:  # we only want to setup logging once
            accelerator.init_trackers(_A  ,      vars(_A     )     )
            SCREAMING_SNAKE_CASE	: Any					       =				accelerator.trackers[0].run.name
            logger.setLevel(logging.INFO     )
            datasets.utils.logging.set_verbosity_info()
            transformers.utils.logging.set_verbosity_info()
      else:
            SCREAMING_SNAKE_CASE	: Tuple					       =				""""""
            logger.setLevel(logging.ERROR     )
            datasets.utils.logging.set_verbosity_error()
            transformers.utils.logging.set_verbosity_error()
      return logger, run_name






def    __lowercase (       _A     )       ->  List[str]:
      SCREAMING_SNAKE_CASE	: Union[str, Any]					       =				{"""streaming""": True}
      SCREAMING_SNAKE_CASE	: Dict					       =				load_dataset(args.dataset_name_train  ,      split="""train"""  ,      **_A     )
      SCREAMING_SNAKE_CASE	: Union[str, Any]					       =				train_data.shuffle(buffer_size=args.shuffle_buffer  ,      seed=args.seed     )
      SCREAMING_SNAKE_CASE	: Tuple					       =				load_dataset(args.dataset_name_valid  ,      split="""train"""  ,      **_A     )
      SCREAMING_SNAKE_CASE	: Optional[Any]					       =				ConstantLengthDataset(
          _A  ,      _A  ,      infinite=_A  ,      seq_length=args.seq_length  ,      tokenized=args.tokenized     )
      SCREAMING_SNAKE_CASE	: Tuple					       =				ConstantLengthDataset(
          _A  ,      _A  ,      infinite=_A  ,      seq_length=args.seq_length  ,      tokenized=args.tokenized     )
      SCREAMING_SNAKE_CASE	: Dict					       =				train_dataset.shuffle(buffer_size=args.shuffle_buffer     )
      SCREAMING_SNAKE_CASE	: int					       =				DataLoader(_A  ,      batch_size=args.train_batch_size  ,      shuffle=_A     )
      SCREAMING_SNAKE_CASE	: Any					       =				DataLoader(_A  ,      batch_size=args.valid_batch_size     )
      return train_dataloader, eval_dataloader






def    __lowercase (       _A  ,      _A  ,      _A=["bias", "ln_1.weight", "ln_2.weight", "ln_f.weight"]     )       ->  Dict:
      SCREAMING_SNAKE_CASE       ,		SCREAMING_SNAKE_CASE	: int					       =				[], []
      for n, p in model.named_parameters():
            if any(nd in n for nd in no_decay     ):
                  params_without_wd.append(_A     )
            else:
                  params_with_wd.append(_A     )
      return [
          {"params": params_with_wd, "weight_decay": args.weight_decay},
          {"params": params_without_wd, "weight_decay": 0.0},
      ]






def    __lowercase (       _A  ,      _A     )       ->  Union[str, Any]:
      logger.info(F"Step {step}: {metrics}"     )
      if accelerator.is_main_process:
            accelerator.log(_A  ,      _A     )






def    __lowercase (       _A  ,      _A  ,      _A     )       ->  str:
      # TFLOPs formula (from Equation 3 in Section 5.1 of https://arxiv.org/pdf/2104.04473.pdf).
      SCREAMING_SNAKE_CASE	: Dict					       =				accelerator.unwrap_model(_A     ).config
      SCREAMING_SNAKE_CASE	: str					       =				4 if args.gradient_checkpointing else 3
      SCREAMING_SNAKE_CASE	: Dict					       =				args.train_batch_size * accelerator.state.num_processes * args.gradient_accumulation_steps
      SCREAMING_SNAKE_CASE	: int					       =				24 * checkpoint_factor * batch_size * args.seq_length * config_model.n_layer * (config_model.n_embd**2)
      SCREAMING_SNAKE_CASE	: Union[str, Any]					       =				factor * (
          1.0
          + (args.seq_length / (6.0 * config_model.n_embd))
          + (tokenizer.vocab_size / (16.0 * config_model.n_layer * config_model.n_embd))
      )
      SCREAMING_SNAKE_CASE	: Any					       =				flops_per_iteration / (elapsed_time * accelerator.state.num_processes * (10**12))
      return tflops






def    __lowercase (       _A     )       ->  Any:
      model.eval()
      SCREAMING_SNAKE_CASE	: Optional[int]					       =				[]
      for step, batch in enumerate(_A     ):
            with torch.no_grad():
                  SCREAMING_SNAKE_CASE	: List[str]					       =				model(_A  ,      labels=_A     )
            SCREAMING_SNAKE_CASE	: Tuple					       =				outputs.loss.repeat(args.valid_batch_size     )
            losses.append(accelerator.gather(_A     )     )
            if args.max_eval_steps > 0 and step >= args.max_eval_steps:
                  break
      SCREAMING_SNAKE_CASE	: Tuple					       =				torch.cat(_A     )
      SCREAMING_SNAKE_CASE	: int					       =				losses[: eval_dataloader.dataset.current_size].mean()
      try:
            SCREAMING_SNAKE_CASE	: List[str]					       =				torch.exp(_A     )
      except OverflowError:
            SCREAMING_SNAKE_CASE	: str					       =				float("""inf"""     )
      return loss.item(), perplexity.item()


# Settings
UpperCAmelCase__	:       List[Any]			      =      HfArgumentParser(TrainingArguments)
UpperCAmelCase__	:       List[str]			      =      parser.parse_args()

# Accelerator
UpperCAmelCase__	:       Any			      =      ProjectConfiguration(project_dir=args.save_dir, logging_dir="""log""")
UpperCAmelCase__	:       str			      =      Accelerator(log_with=["""wandb""", """tensorboard"""], project_config=config)
UpperCAmelCase__	:       int			      =      {str(k): str(v) for k, v in accelerator.state.__dict__.items()}

UpperCAmelCase__	:       str			      =      Namespace(**vars(args), **acc_state)
UpperCAmelCase__	:       Tuple			      =      accelerator.state.num_processes * args.train_batch_size
set_seed(args.seed)

# Clone model repository
if accelerator.is_main_process:
  UpperCAmelCase__	:       str			      =      Repository(args.save_dir, clone_from=args.model_ckpt)

# Logging
UpperCAmelCase__      ,				UpperCAmelCase__	:       List[Any]			      =      setup_logging(args)
logger.info(accelerator.state)

# Checkout new branch on repo
if accelerator.is_main_process:
  hf_repo.git_checkout(run_name, create_branch_ok=True)

# Load model and tokenizer
UpperCAmelCase__	:       Any			      =      AutoModelForCausalLM.from_pretrained(args.save_dir)
if args.gradient_checkpointing:
  model.gradient_checkpointing_enable()
UpperCAmelCase__	:       Optional[Any]			      =      AutoTokenizer.from_pretrained(args.save_dir)

# Load dataset and dataloader
UpperCAmelCase__      ,				UpperCAmelCase__	:       Optional[Any]			      =      create_dataloaders(args)

# Prepare the optimizer and learning rate scheduler
UpperCAmelCase__	:       Tuple			      =      AdamW(get_grouped_params(model, args), lr=args.learning_rate)
UpperCAmelCase__	:       str			      =      get_scheduler(
    name=args.lr_scheduler_type,
    optimizer=optimizer,
    num_warmup_steps=args.num_warmup_steps,
    num_training_steps=args.max_train_steps,
)
accelerator.register_for_checkpointing(lr_scheduler)






def    __lowercase (       )       ->  str:
      return optimizer.param_groups[0]["lr"]


# Prepare everything with our `accelerator`.
UpperCAmelCase__      ,				UpperCAmelCase__      ,				UpperCAmelCase__      ,				UpperCAmelCase__	:       Tuple			      =      accelerator.prepare(
    model, optimizer, train_dataloader, eval_dataloader
)

# load in the weights and states from a previous save
if args.resume_from_checkpoint:
  if args.resume_from_checkpoint is not None or args.resume_from_checkpoint != "":
    accelerator.print(F"""Resumed from checkpoint: {args.resume_from_checkpoint}""")
    accelerator.load_state(args.resume_from_checkpoint)
    UpperCAmelCase__	:       Union[str, Any]			      =      os.path.basename(args.resume_from_checkpoint)
  else:
    # Get the most recent checkpoint
    UpperCAmelCase__	:       Any			      =      [f.name for f in os.scandir(args.save_dir) if f.is_dir() and """step""" in str(f)]
    dirs.sort(key=os.path.getctime)
    UpperCAmelCase__	:       Tuple			      =      dirs[-1]  # Sorts folders by date modified, most recent checkpoint is the last
  # Extract the step of the checkpoint to continue from there
  UpperCAmelCase__	:       Optional[int]			      =      os.path.splitext(path)[0]
  UpperCAmelCase__	:       List[Any]			      =      int(training_difference.replace("""step_""", """"""))

# Train model
model.train()
UpperCAmelCase__	:       str			      =      0
UpperCAmelCase__	:       Any			      =      time.time()
UpperCAmelCase__	:       Optional[int]			      =      0
for step, batch in enumerate(train_dataloader, start=1):
  if args.resume_from_checkpoint and step < resume_step:
    continue  # we need to skip steps until we reach the resumed step
  UpperCAmelCase__	:       str			      =      model(batch, labels=batch, use_cache=False).loss
  UpperCAmelCase__	:       List[str]			      =      accelerator.gather(loss.repeat(args.train_batch_size)).mean()
  loss_tracking += avg_loss.item() / args.gradient_accumulation_steps
  log_metrics(step, {"""samples""": step * samples_per_step, """loss_per_step/train""": loss.item()})
  UpperCAmelCase__	:       Tuple			      =      loss / args.gradient_accumulation_steps
  if step % args.gradient_accumulation_steps != 0:
    # Prevent backward from doing gradient all_reduce in every step
    if accelerator.distributed_type == DistributedType.MULTI_GPU:
      with model.no_sync():
        accelerator.backward(loss)
    else:
      accelerator.backward(loss)
  else:
    UpperCAmelCase__	:       Any			      =      get_lr()
    accelerator.backward(loss)
    accelerator.clip_grad_norm_(model.parameters(), 1.0)
    optimizer.step()
    lr_scheduler.step()
    optimizer.zero_grad()
    UpperCAmelCase__	:       Dict			      =      time.time() - t_start
    UpperCAmelCase__	:       Optional[int]			      =      compute_tflops(elapsed_time, accelerator, args)
    log_metrics(
        step,
        {
            """steps""": completed_steps,
            """loss/train""": loss_tracking,
            """lr""": lr,
            """tflops""": tflops,
            """time_per_iteration""": elapsed_time,
        },
    )
    UpperCAmelCase__	:       List[str]			      =      time.time()
    UpperCAmelCase__	:       List[str]			      =      0
    completed_steps += 1
  if step % args.save_checkpoint_steps == 0:
    logger.info("""Evaluating and saving model checkpoint""")
    UpperCAmelCase__      ,				UpperCAmelCase__	:       Optional[Any]			      =      evaluate(args)
    log_metrics(step, {"""loss/eval""": eval_loss, """perplexity""": perplexity})
    accelerator.wait_for_everyone()
    UpperCAmelCase__	:       List[Any]			      =      os.path.join(args.save_dir, F"""step_{step}""")
    accelerator.save_state(save_dir)
    if accelerator.is_main_process:
      hf_repo.push_to_hub(commit_message=F"""step {step}""")
    model.train()
  if completed_steps >= args.max_train_steps:
    break

# Evaluate and save the last checkpoint
logger.info("""Evaluating and saving model after training""")
UpperCAmelCase__      ,				UpperCAmelCase__	:       Optional[Any]			      =      evaluate(args)
log_metrics(step, {"""loss/eval""": eval_loss, """perplexity""": perplexity})
accelerator.wait_for_everyone()
UpperCAmelCase__	:       List[Any]			      =      accelerator.unwrap_model(model)
unwrapped_model.save_pretrained(args.save_dir, save_function=accelerator.save)
UpperCAmelCase__	:       str			      =      os.path.join(args.save_dir, F"""step_{step}""")
accelerator.save_state(save_dir)
if accelerator.is_main_process:
  hf_repo.push_to_hub(commit_message="""final model""")







import inspect
import unittest

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

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


if is_torch_available():
  import torch

  from transformers import EfficientNetForImageClassification, EfficientNetModel
  from transformers.models.efficientnet.modeling_efficientnet import EFFICIENTNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
  from PIL import Image

  from transformers import AutoImageProcessor







class     a__    :







  """simple docstring"""




  def __init__(     self      :	Any			,			UpperCAmelCase__      :	Union[str, Any]			,			UpperCAmelCase__      :	List[str]=1_3			,			UpperCAmelCase__      :	Any=3_2			,			UpperCAmelCase__      :	Union[str, Any]=3			,			UpperCAmelCase__      :	List[str]=[3, 3, 5]			,			UpperCAmelCase__      :	int=[3_2, 1_6, 2_4]			,			UpperCAmelCase__      :	Dict=[1_6, 2_4, 4_0]			,			UpperCAmelCase__      :	Optional[Any]=[1, 1, 2]			,			UpperCAmelCase__      :	List[Any]=[1, 1, 2]			,			UpperCAmelCase__      :	str=[1, 6, 6]			,			UpperCAmelCase__      :	List[Any]=True			,			UpperCAmelCase__      :	int=True			,			UpperCAmelCase__      :	Any=3_7			,			UpperCAmelCase__      :	Any="gelu"			,			UpperCAmelCase__      :	List[Any]=1_0			,			)   ->Dict:
        """simple docstring"""







        SCREAMING_SNAKE_CASE	: Optional[int]					       =				parent
        SCREAMING_SNAKE_CASE	: Any					       =				batch_size
        SCREAMING_SNAKE_CASE	: List[Any]					       =				image_size
        SCREAMING_SNAKE_CASE	: str					       =				num_channels
        SCREAMING_SNAKE_CASE	: int					       =				kernel_sizes
        SCREAMING_SNAKE_CASE	: Union[str, Any]					       =				in_channels
        SCREAMING_SNAKE_CASE	: List[str]					       =				out_channels
        SCREAMING_SNAKE_CASE	: int					       =				strides
        SCREAMING_SNAKE_CASE	: List[str]					       =				num_block_repeats
        SCREAMING_SNAKE_CASE	: Tuple					       =				expand_ratios
        SCREAMING_SNAKE_CASE	: Dict					       =				is_training
        SCREAMING_SNAKE_CASE	: Optional[int]					       =				hidden_act
        SCREAMING_SNAKE_CASE	: int					       =				num_labels
        SCREAMING_SNAKE_CASE	: Dict					       =				use_labels


  def 	_lowercase (     self      :	Any						)   ->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE	: Union[str, Any]					       =				floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]						)

        SCREAMING_SNAKE_CASE	: Dict					       =				None
        if self.use_labels:
              SCREAMING_SNAKE_CASE	: Optional[Any]					       =				ids_tensor([self.batch_size]			,			self.num_labels						)

        SCREAMING_SNAKE_CASE	: Dict					       =				self.get_config()
        return config, pixel_values, labels


  def 	_lowercase (     self      :	Tuple						)   ->Optional[int]:
        """simple docstring"""







        return EfficientNetConfig(
            num_channels=self.num_channels			,			kernel_sizes=self.kernel_sizes			,			in_channels=self.in_channels			,			out_channels=self.out_channels			,			strides=self.strides			,			num_block_repeats=self.num_block_repeats			,			expand_ratios=self.expand_ratios			,			hidden_act=self.hidden_act			,			num_labels=self.num_labels			,			)


  def 	_lowercase (     self      :	List[Any]			,			UpperCAmelCase__      :	str			,			UpperCAmelCase__      :	int			,			UpperCAmelCase__      :	Optional[Any]						)   ->Union[str, Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE	: List[Any]					       =				EfficientNetModel(config=UpperCAmelCase__						)
        model.to(UpperCAmelCase__						)
        model.eval()
        SCREAMING_SNAKE_CASE	: str					       =				model(UpperCAmelCase__						)
        # expected last hidden states: B, C, H // 4, W // 4
        self.parent.assertEqual(
            result.last_hidden_state.shape			,			(self.batch_size, config.hidden_dim, self.image_size // 4, self.image_size // 4)			,			)


  def 	_lowercase (     self      :	Optional[int]			,			UpperCAmelCase__      :	List[Any]			,			UpperCAmelCase__      :	Optional[int]			,			UpperCAmelCase__      :	Union[str, Any]						)   ->int:
        """simple docstring"""







        SCREAMING_SNAKE_CASE	: Union[str, Any]					       =				EfficientNetForImageClassification(UpperCAmelCase__						)
        model.to(UpperCAmelCase__						)
        model.eval()
        SCREAMING_SNAKE_CASE	: Any					       =				model(UpperCAmelCase__			,			labels=UpperCAmelCase__						)
        self.parent.assertEqual(result.logits.shape			,			(self.batch_size, self.num_labels)						)




  def 	_lowercase (     self      :	Dict						)   ->str:
        """simple docstring"""







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







@require_torch
class     a__    (				UpperCAmelCase		,     UpperCAmelCase		,     unittest.TestCase    ):







  """simple docstring"""




  UpperCAmelCase__ :	Tuple         =(EfficientNetModel, EfficientNetForImageClassification) if is_torch_available() else ()
  UpperCAmelCase__ :	Optional[int]         =(
      {"""feature-extraction""": EfficientNetModel, """image-classification""": EfficientNetForImageClassification}
      if is_torch_available()
      else {}
  )

  UpperCAmelCase__ :	List[str]         =False
  UpperCAmelCase__ :	Any         =False
  UpperCAmelCase__ :	Union[str, Any]         =False
  UpperCAmelCase__ :	Dict         =False
  UpperCAmelCase__ :	List[str]         =False


  def 	_lowercase (     self      :	Optional[Any]						)   ->Tuple:
        """simple docstring"""







        SCREAMING_SNAKE_CASE	: Optional[Any]					       =				EfficientNetModelTester(self						)
        SCREAMING_SNAKE_CASE	: Any					       =				ConfigTester(
            self			,			config_class=UpperCAmelCase__			,			has_text_modality=UpperCAmelCase__			,			hidden_size=3_7						)


  def 	_lowercase (     self      :	Optional[int]						)   ->List[Any]:
        """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 	_lowercase (     self      :	List[Any]						)   ->Optional[int]:
        """simple docstring"""







        return


  @unittest.skip(reason="""EfficientNet does not use inputs_embeds"""						)
  def 	_lowercase (     self      :	Dict						)   ->Tuple:
        """simple docstring"""







        pass


  @unittest.skip(reason="""EfficientNet does not support input and output embeddings"""						)
  def 	_lowercase (     self      :	Dict						)   ->Dict:
        """simple docstring"""







        pass


  @unittest.skip(reason="""EfficientNet does not use feedforward chunking"""						)
  def 	_lowercase (     self      :	Optional[Any]						)   ->Any:
        """simple docstring"""







        pass


  def 	_lowercase (     self      :	str						)   ->Optional[int]:
        """simple docstring"""







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

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

              SCREAMING_SNAKE_CASE	: Dict					       =				["""pixel_values"""]
              self.assertListEqual(arg_names[:1]			,			UpperCAmelCase__						)


  def 	_lowercase (     self      :	Optional[int]						)   ->Optional[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE	: Optional[int]					       =				self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*UpperCAmelCase__						)


  def 	_lowercase (     self      :	List[Any]						)   ->int:
        """simple docstring"""







        def check_hidden_states_output(UpperCAmelCase__      :	Optional[Any]			,			UpperCAmelCase__      :	Any			,			UpperCAmelCase__      :	Dict						):
              SCREAMING_SNAKE_CASE	: Optional[Any]					       =				model_class(UpperCAmelCase__						)
              model.to(UpperCAmelCase__						)
              model.eval()

              with torch.no_grad():
                    SCREAMING_SNAKE_CASE	: Any					       =				model(**self._prepare_for_class(UpperCAmelCase__			,			UpperCAmelCase__						)						)

              SCREAMING_SNAKE_CASE	: Optional[int]					       =				outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states
              SCREAMING_SNAKE_CASE	: Any					       =				sum(config.num_block_repeats						) * 4
              self.assertEqual(len(UpperCAmelCase__						)			,			UpperCAmelCase__						)

              # EfficientNet's feature maps are of shape (batch_size, num_channels, height, width)
              self.assertListEqual(
                  list(hidden_states[0].shape[-2:]						)			,			[self.model_tester.image_size // 2, self.model_tester.image_size // 2]			,			)

        SCREAMING_SNAKE_CASE       ,		SCREAMING_SNAKE_CASE	: Tuple					       =				self.model_tester.prepare_config_and_inputs_for_common()

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

              # check that output_hidden_states also work using config
              del inputs_dict["output_hidden_states"]
              SCREAMING_SNAKE_CASE	: Dict					       =				True

              check_hidden_states_output(UpperCAmelCase__			,			UpperCAmelCase__			,			UpperCAmelCase__						)


  def 	_lowercase (     self      :	Union[str, Any]						)   ->Optional[int]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE	: Tuple					       =				self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_image_classification(*UpperCAmelCase__						)


  @unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""						)
  def 	_lowercase (     self      :	List[str]						)   ->Optional[Any]:
        """simple docstring"""







        pass


  @slow
  def 	_lowercase (     self      :	Dict						)   ->List[Any]:
        """simple docstring"""







        for model_name in EFFICIENTNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
              SCREAMING_SNAKE_CASE	: str					       =				EfficientNetModel.from_pretrained(UpperCAmelCase__						)
              self.assertIsNotNone(UpperCAmelCase__						)




  @is_pipeline_test
  @require_vision
  @slow
  def 	_lowercase (     self      :	Any						)   ->Any:
        """simple docstring"""







        super().test_pipeline_image_classification()






def    __lowercase (       )       ->  List[Any]:
      SCREAMING_SNAKE_CASE	: List[str]					       =				Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""     )
      return image







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







  """simple docstring"""




  @cached_property
  def 	_lowercase (     self      :	List[str]						)   ->Tuple:
        """simple docstring"""







        return AutoImageProcessor.from_pretrained("""google/efficientnet-b7"""						) if is_vision_available() else None




  @slow
  def 	_lowercase (     self      :	List[Any]						)   ->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE	: Tuple					       =				EfficientNetForImageClassification.from_pretrained("""google/efficientnet-b7"""						).to(UpperCAmelCase__						)

        SCREAMING_SNAKE_CASE	: int					       =				self.default_image_processor
        SCREAMING_SNAKE_CASE	: List[Any]					       =				prepare_img()
        SCREAMING_SNAKE_CASE	: Optional[Any]					       =				image_processor(images=UpperCAmelCase__			,			return_tensors="""pt"""						).to(UpperCAmelCase__						)

        # forward pass
        with torch.no_grad():
              SCREAMING_SNAKE_CASE	: Optional[Any]					       =				model(**UpperCAmelCase__						)

        # verify the logits
        SCREAMING_SNAKE_CASE	: Union[str, Any]					       =				torch.Size((1, 1_0_0_0)						)
        self.assertEqual(outputs.logits.shape			,			UpperCAmelCase__						)

        SCREAMING_SNAKE_CASE	: Any					       =				torch.tensor([0.00_01, 0.00_02, 0.00_02]						).to(UpperCAmelCase__						)
        self.assertTrue(torch.allclose(outputs.logits[0, :3]			,			UpperCAmelCase__			,			atol=1e-4						)						)




