


"""simple docstring"""



import logging
import os
from dataclasses import dataclass
from typing import List, Optional, Union

import tqdm
from filelock import FileLock

from transformers import (
    BartTokenizer,
    BartTokenizerFast,
    DataProcessor,
    PreTrainedTokenizer,
    RobertaTokenizer,
    RobertaTokenizerFast,
    XLMRobertaTokenizer,
    is_tf_available,
    is_torch_available,
)


__snake_case       =       logging.getLogger(__name__)







@dataclass(frozen=a__      )
class _lowerCAmelCase							:
   __UpperCAmelCase					:			Optional[int]				    =    42
   __UpperCAmelCase					:			int				    =    42
   __UpperCAmelCase					:			Optional[int]				    =    None
   __UpperCAmelCase					:			Union[str, Any]				    =    None
   __UpperCAmelCase					:			Dict				    =    None







@dataclass(frozen=a__      )
class _lowerCAmelCase							:
   __UpperCAmelCase					:			Any				    =    42
   __UpperCAmelCase					:			int				    =    None
   __UpperCAmelCase					:			Tuple				    =    None
   __UpperCAmelCase					:			Tuple				    =    None
   __UpperCAmelCase					:			Optional[Any]				    =    None


if is_torch_available():
  import torch
  from torch.utils.data import Dataset







  class _lowerCAmelCase							( a__      ):
     __UpperCAmelCase					:			List[Any]				    =    42







     def __init__(      self		,      UpperCamelCase__		,      UpperCamelCase__		,      UpperCamelCase__		,      UpperCamelCase__ = None		,      UpperCamelCase__=False		,      UpperCamelCase__ = False		,      )      ->						Dict:

         '''simple docstring'''




         snake_case       : Any    =       hans_processors[task]()

         snake_case       : Optional[int]    =       os.path.join(
             __a		,      "cached_{}_{}_{}_{}".format(
                 "dev" if evaluate else "train"		,      tokenizer.__class__.__name__		,      str(__a		)		,      __a		,      )		,      )
         snake_case       : List[str]    =       processor.get_labels()
         if tokenizer.__class__ in (
             RobertaTokenizer,
             RobertaTokenizerFast,
             XLMRobertaTokenizer,
             BartTokenizer,
             BartTokenizerFast,
         ):
             # HACK(label indices are swapped in RoBERTa pretrained model)
             snake_case       : Tuple    =       label_list[2], label_list[1]
         snake_case       : List[Any]    =       label_list

         # Make sure only the first process in distributed training processes the dataset,
         # and the others will use the cache.
         snake_case       : List[Any]    =       cached_features_file + """.lock"""
         with FileLock(__a		):
             if os.path.exists(__a		) and not overwrite_cache:
                 logger.info(F'Loading features from cached file {cached_features_file}'		)
                 snake_case       : List[str]    =       torch.load(__a		)
             else:
                 logger.info(F'Creating features from dataset file at {data_dir}'		)

                 snake_case       : List[str]    =       (
                     processor.get_dev_examples(__a		) if evaluate else processor.get_train_examples(__a		)
                 )

                 logger.info("Training examples: %s"		,      len(__a		)		)
                 snake_case       : str    =       hans_convert_examples_to_features(__a		,      __a		,      __a		,      __a		)
                 logger.info("Saving features into cached file %s"		,      __a		)
                 torch.save(self.features		,      __a		)







     def __len__(      self		)      ->						Tuple:

         '''simple docstring'''




         return len(self.features		)







     def __getitem__(      self		,      UpperCamelCase__		)      ->						InputFeatures:

         '''simple docstring'''




         return self.features[i]







     def 					lowerCamelCase						(      self		)      ->						Tuple:

         '''simple docstring'''




         return self.label_list


if is_tf_available():
  import tensorflow as tf







  class _lowerCAmelCase							:
     __UpperCAmelCase					:			str				    =    42







     def __init__(      self		,      UpperCamelCase__		,      UpperCamelCase__		,      UpperCamelCase__		,      UpperCamelCase__ = 128		,      UpperCamelCase__=False		,      UpperCamelCase__ = False		,      )      ->						int:

         '''simple docstring'''




         snake_case       : Tuple    =       hans_processors[task]()
         snake_case       : str    =       processor.get_labels()
         if tokenizer.__class__ in (
             RobertaTokenizer,
             RobertaTokenizerFast,
             XLMRobertaTokenizer,
             BartTokenizer,
             BartTokenizerFast,
         ):
             # HACK(label indices are swapped in RoBERTa pretrained model)
             snake_case       : Any    =       label_list[2], label_list[1]
         snake_case       : Optional[int]    =       label_list

         snake_case       : Any    =       processor.get_dev_examples(__a		) if evaluate else processor.get_train_examples(__a		)
         snake_case       : Dict    =       hans_convert_examples_to_features(__a		,      __a		,      __a		,      __a		)

         def gen():
             for ex_index, ex in tqdm.tqdm(enumerate(self.features		)		,      desc="convert examples to features"		):
                 if ex_index % 1_0000 == 0:
                     logger.info("Writing example %d of %d" % (ex_index, len(__a		))		)

                 yield (
                     {
                         "example_id": 0,
                         "input_ids": ex.input_ids,
                         "attention_mask": ex.attention_mask,
                         "token_type_ids": ex.token_type_ids,
                     },
                     ex.label,
                 )

         snake_case       : Union[str, Any]    =       tf.data.Dataset.from_generator(
             __a		,      (
                 {
                     "example_id": tf.intaa,
                     "input_ids": tf.intaa,
                     "attention_mask": tf.intaa,
                     "token_type_ids": tf.intaa,
                 },
                 tf.intaa,
             )		,      (
                 {
                     "example_id": tf.TensorShape([]		),
                     "input_ids": tf.TensorShape([None, None]		),
                     "attention_mask": tf.TensorShape([None, None]		),
                     "token_type_ids": tf.TensorShape([None, None]		),
                 },
                 tf.TensorShape([]		),
             )		,      )







     def 					lowerCamelCase						(      self		)      ->						Tuple:

         '''simple docstring'''




         return self.dataset







     def __len__(      self		)      ->						Dict:

         '''simple docstring'''




         return len(self.features		)







     def __getitem__(      self		,      UpperCamelCase__		)      ->						InputFeatures:

         '''simple docstring'''




         return self.features[i]







     def 					lowerCamelCase						(      self		)      ->						Tuple:

         '''simple docstring'''




         return self.label_list







class _lowerCAmelCase							( a__      ):







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		)      ->						Dict:

       '''simple docstring'''




       return self._create_examples(self._read_tsv(os.path.join(__a		,      "heuristics_train_set.txt"		)		)		,      "train"		)







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		)      ->						Union[str, Any]:

       '''simple docstring'''




       return self._create_examples(self._read_tsv(os.path.join(__a		,      "heuristics_evaluation_set.txt"		)		)		,      "dev"		)







   def 					lowerCamelCase						(      self		)      ->						str:

       '''simple docstring'''




       return ["contradiction", "entailment", "neutral"]







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		,      UpperCamelCase__		)      ->						List[Any]:

       '''simple docstring'''




       snake_case       : Optional[int]    =       []
       for i, line in enumerate(__a		):
           if i == 0:
               continue
           snake_case       : int    =       """%s-%s""" % (set_type, line[0])
           snake_case       : int    =       line[5]
           snake_case       : Tuple    =       line[6]
           snake_case       : Any    =       line[7][2:] if line[7].startswith("ex"		) else line[7]
           snake_case       : Optional[Any]    =       line[0]
           examples.append(InputExample(guid=__a		,      text_a=__a		,      text_b=__a		,      label=__a		,      pairID=__a		)		)
       return examples







def     __lowerCAmelCase		(		lowercase    :    List[InputExample]		,				lowercase    :    List[str]		,				lowercase    :    int		,				lowercase    :    PreTrainedTokenizer		,				)      ->		Optional[int]:


    """simple docstring"""





    snake_case       : List[str]    =       {label: i for i, label in enumerate(A__					)}

    snake_case       : Optional[int]    =       []
    for ex_index, example in tqdm.tqdm(enumerate(A__					)		,				desc="convert examples to features"					):
        if ex_index % 1_0000 == 0:
            logger.info("Writing example %d" % (ex_index)					)

        snake_case       : Union[str, Any]    =       tokenizer(
            example.text_a		,				example.text_b		,				add_special_tokens=A__		,				max_length=A__		,				padding="max_length"		,				truncation=A__		,				return_overflowing_tokens=A__		,				)

        snake_case       : Dict    =       label_map[example.label] if example.label in label_map else 0

        snake_case       : Dict    =       int(example.pairID					)

        features.append(InputFeatures(**A__		,				label=A__		,				pairID=A__					)					)

    for i, example in enumerate(examples[:5]					):
        logger.info("*** Example ***"					)
        logger.info(F'guid: {example}'					)
        logger.info(F'features: {features[i]}'					)

    return features


__snake_case       =       {
    'hans': 3,
}


__snake_case       =       {
    'hans': HansProcessor,
}









import argparse
import json
from pathlib import Path

import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image
from timm import create_model
from timm.data import resolve_data_config
from timm.data.transforms_factory import create_transform

from transformers import BitConfig, BitForImageClassification, BitImageProcessor
from transformers.image_utils import PILImageResampling
from transformers.utils import logging


logging.set_verbosity_info()
snake_case__				:	Dict					       =					logging.get_logger(__name__)


def    __lowerCamelCase   (      A__						: Optional[Any]					)     ->    List[str]:
   lowerCamelCase_						:   int		= """huggingface/label-files"""
   lowerCamelCase_						:   Dict		= """imagenet-1k-id2label.json"""
   lowerCamelCase_						:   Optional[Any]		= json.load(open(hf_hub_download(A__						,						A__						,						repo_type="""dataset"""					)						,						"""r"""					)					)
   lowerCamelCase_						:   str		= {int(A__					): v for k, v in idalabel.items()}
   lowerCamelCase_						:   List[Any]		= {v: k for k, v in idalabel.items()}

   lowerCamelCase_						:   str		= """std_conv""" if """bit""" in model_name else False

   # note that when using BiT as backbone for ViT-hybrid checkpoints,
   # one needs to additionally set config.layer_type = "bottleneck", config.stem_type = "same",
   # config.conv_layer = "std_conv_same"
   lowerCamelCase_						:   Optional[Any]		= BitConfig(
       conv_layer=A__						,						num_labels=1000						,						idalabel=A__						,						labelaid=A__						,						)

   return config


def    __lowerCamelCase   (      A__						: str					)     ->    Any:
   if "stem.conv" in name:
      lowerCamelCase_						:   Any		= name.replace("""stem.conv"""						,						"""bit.embedder.convolution"""					)
   if "blocks" in name:
      lowerCamelCase_						:   Dict		= name.replace("""blocks"""						,						"""layers"""					)
   if "head.fc" in name:
      lowerCamelCase_						:   Optional[Any]		= name.replace("""head.fc"""						,						"""classifier.1"""					)
   if name.startswith("""norm"""					):
      lowerCamelCase_						:   int		= """bit.""" + name
   if "bit" not in name and "classifier" not in name:
      lowerCamelCase_						:   str		= """bit.encoder.""" + name

   return name


def    __lowerCamelCase   (      )     ->    List[Any]:
   lowerCamelCase_						:   List[Any]		= """http://images.cocodataset.org/val2017/000000039769.jpg"""
   lowerCamelCase_						:   Optional[Any]		= Image.open(requests.get(A__						,						stream=A__					).raw					)
   return im


@torch.no_grad()
def    __lowerCamelCase   (      A__						: List[Any]						,						A__						: List[str]						,						A__						: Tuple=False					)     ->    List[str]:
   lowerCamelCase_						:   Optional[Any]		= get_config(A__					)

   # load original model from timm
   lowerCamelCase_						:   Optional[Any]		= create_model(A__						,						pretrained=A__					)
   timm_model.eval()

   # load state_dict of original model
   lowerCamelCase_						:   Optional[int]		= timm_model.state_dict()
   for key in state_dict.copy().keys():
      lowerCamelCase_						:   int		= state_dict.pop(A__					)
      lowerCamelCase_						:   Union[str, Any]		= val.squeeze() if """head""" in key else val

   # load HuggingFace model
   lowerCamelCase_						:   Tuple		= BitForImageClassification(A__					)
   model.eval()
   model.load_state_dict(A__					)

   # create image processor
   lowerCamelCase_						:   Optional[Any]		= create_transform(**resolve_data_config({}						,						model=A__					)					)
   lowerCamelCase_						:   List[Any]		= transform.transforms

   lowerCamelCase_						:   List[Any]		= {
       """bilinear""": PILImageResampling.BILINEAR,
       """bicubic""": PILImageResampling.BICUBIC,
       """nearest""": PILImageResampling.NEAREST,
   }

   lowerCamelCase_						:   List[str]		= BitImageProcessor(
       do_resize=A__						,						size={"""shortest_edge""": timm_transforms[0].size}						,						resample=pillow_resamplings[timm_transforms[0].interpolation.value]						,						do_center_crop=A__						,						crop_size={"""height""": timm_transforms[1].size[0], """width""": timm_transforms[1].size[1]}						,						do_normalize=A__						,						image_mean=timm_transforms[-1].mean.tolist()						,						image_std=timm_transforms[-1].std.tolist()						,						)

   lowerCamelCase_						:   int		= prepare_img()
   lowerCamelCase_						:   int		= transform(A__					).unsqueeze(0					)
   lowerCamelCase_						:   List[str]		= processor(A__						,						return_tensors="""pt"""					).pixel_values

   # verify pixel values
   assert torch.allclose(A__						,						A__					)

   # verify logits
   with torch.no_grad():
      lowerCamelCase_						:   str		= model(A__					)
      lowerCamelCase_						:   int		= outputs.logits

   print("""Logits:"""						,						logits[0, :3]					)
   print("""Predicted class:"""						,						model.config.idalabel[logits.argmax(-1					).item()]					)
   lowerCamelCase_						:   List[Any]		= timm_model(A__					)
   assert timm_logits.shape == outputs.logits.shape
   assert torch.allclose(A__						,						outputs.logits						,						atol=1e-3					)
   print("""Looks ok!"""					)

   if pytorch_dump_folder_path is not None:
      Path(A__					).mkdir(exist_ok=A__					)
      print(f'''Saving model {model_name} and processor to {pytorch_dump_folder_path}'''					)
      model.save_pretrained(A__					)
      processor.save_pretrained(A__					)

   if push_to_hub:
      print(f'''Pushing model {model_name} and processor to the hub'''					)
      model.push_to_hub(f'''ybelkada/{model_name}'''					)
      processor.push_to_hub(f'''ybelkada/{model_name}'''					)


if __name__ == "__main__":
      snake_case__				:	Dict					       =					argparse.ArgumentParser()
      # Required parameters
      parser.add_argument(
          '--model_name',
          default='resnetv2_50x1_bitm',
          type=str,
          help='Name of the BiT timm model you\'d like to convert.',
      )
      parser.add_argument(
          '--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model directory.'
      )
      parser.add_argument(
          '--push_to_hub',
          action='store_true',
          help='Whether to push the model to the hub.',
      )

      snake_case__				:	int					       =					parser.parse_args()
      convert_bit_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)





