


'''simple docstring'''






import argparse
import json
from pathlib import Path

import requests
import torch
from huggingface_hub import cached_download, hf_hub_url
from PIL import Image

from transformers import DeformableDetrConfig, DeformableDetrForObjectDetection, DeformableDetrImageProcessor
from transformers.utils import logging


logging.set_verbosity_info()
a_       :     Optional[int] 							=     logging.get_logger(__name__)






def 						_A					(lowerCAmelCase__	:Dict		) ->	List[str]:




     '''simple docstring'''



     if "backbone.0.body" in orig_key:
          _a         =				orig_key.replace('backbone.0.body'					,    'backbone.conv_encoder.model'		)
     if "transformer" in orig_key:
          _a         =				orig_key.replace('transformer.'					,    ''		)
     if "norm1" in orig_key:
          if "encoder" in orig_key:
               _a         =				orig_key.replace('norm1'					,    'self_attn_layer_norm'		)
          else:
               _a         =				orig_key.replace('norm1'					,    'encoder_attn_layer_norm'		)
     if "norm2" in orig_key:
          if "encoder" in orig_key:
               _a         =				orig_key.replace('norm2'					,    'final_layer_norm'		)
          else:
               _a         =				orig_key.replace('norm2'					,    'self_attn_layer_norm'		)
     if "norm3" in orig_key:
          _a         =				orig_key.replace('norm3'					,    'final_layer_norm'		)
     if "linear1" in orig_key:
          _a         =				orig_key.replace('linear1'					,    'fc1'		)
     if "linear2" in orig_key:
          _a         =				orig_key.replace('linear2'					,    'fc2'		)
     if "query_embed" in orig_key:
          _a         =				orig_key.replace('query_embed'					,    'query_position_embeddings'		)
     if "cross_attn" in orig_key:
          _a         =				orig_key.replace('cross_attn'					,    'encoder_attn'		)

     return orig_key






def 						_A					(lowerCAmelCase__	:Any		) ->	Optional[int]:




     '''simple docstring'''



     for i in range(6		):
          # read in weights + bias of input projection layer of self-attention
          _a         =				state_dict.pop(f'decoder.layers.{i}.self_attn.in_proj_weight'		)
          _a         =				state_dict.pop(f'decoder.layers.{i}.self_attn.in_proj_bias'		)
          # next, add query, keys and values (in that order) to the state dict
          _a         =				in_proj_weight[:2_56, :]
          _a         =				in_proj_bias[:2_56]
          _a         =				in_proj_weight[2_56:5_12, :]
          _a         =				in_proj_bias[2_56:5_12]
          _a         =				in_proj_weight[-2_56:, :]
          _a         =				in_proj_bias[-2_56:]






def 						_A					() ->	Union[str, Any]:




     '''simple docstring'''



     _a         =				'http://images.cocodataset.org/val2017/000000039769.jpg'
     _a         =				Image.open(requests.get(lowerCAmelCase__					,    stream=lowerCAmelCase__		).raw		)

     return im






@torch.no_grad()
def 						_A					(lowerCAmelCase__	:Any					,    lowerCAmelCase__	:Optional[int]					,    lowerCAmelCase__	:Optional[int]					,    lowerCAmelCase__	:Union[str, Any]					,    lowerCAmelCase__	:Tuple					,    lowerCAmelCase__	:Tuple					,    lowerCAmelCase__	:str					,    ) ->	str:




     '''simple docstring'''



     _a         =				DeformableDetrConfig()
     # set config attributes
     if single_scale:
          _a         =				1
     _a         =				dilation
     _a         =				with_box_refine
     _a         =				two_stage
     # set labels
     _a         =				91
     _a         =				'huggingface/label-files'
     _a         =				'coco-detection-id2label.json'
     _a         =				json.load(open(cached_download(hf_hub_url(lowerCAmelCase__					,    lowerCAmelCase__					,    repo_type='dataset'		)		)					,    'r'		)		)
     _a         =				{int(lowerCAmelCase__		): v for k, v in idalabel.items()}
     _a         =				idalabel
     _a         =				{v: k for k, v in idalabel.items()}

     # load image processor
     _a         =				DeformableDetrImageProcessor(format='coco_detection'		)

     # prepare image
     _a         =				prepare_img()
     _a         =				image_processor(images=lowerCAmelCase__					,    return_tensors='pt'		)
     _a         =				encoding['pixel_values']

     logger.info('Converting model...'		)

     # load original state dict
     _a         =				torch.load(lowerCAmelCase__					,    map_location='cpu'		)['model']
     # rename keys
     for key in state_dict.copy().keys():
          _a         =				state_dict.pop(lowerCAmelCase__		)
          _a         =				val
     # query, key and value matrices need special treatment
     read_in_q_k_v(lowerCAmelCase__		)
     # important: we need to prepend a prefix to each of the base model keys as the head models use different attributes for them
     _a         =				'model.'
     for key in state_dict.copy().keys():
          if not key.startswith('class_embed'		) and not key.startswith('bbox_embed'		):
               _a         =				state_dict.pop(lowerCAmelCase__		)
               _a         =				val
    # finally, create HuggingFace model and load state dict
     _a         =				DeformableDetrForObjectDetection(lowerCAmelCase__		)
     model.load_state_dict(lowerCAmelCase__		)
     model.eval()

     _a         =				'cuda' if torch.cuda.is_available() else 'cpu'
     model.to(lowerCAmelCase__		)
     # verify our conversion
     _a         =				model(pixel_values.to(lowerCAmelCase__		)		)

     _a         =				torch.tensor(
         [[-9.6_6_4_5, -4.3_4_4_9, -5.8_7_0_5], [-9.7_0_3_5, -3.8_5_0_4, -5.0_7_2_4], [-1_0.5_6_3_4, -5.3_3_7_9, -7.5_1_1_6]]		)
     _a         =				torch.tensor([[0.8_6_9_3, 0.2_2_8_9, 0.2_4_9_2], [0.3_1_5_0, 0.5_4_8_9, 0.5_8_4_5], [0.5_5_6_3, 0.7_5_8_0, 0.8_5_1_8]]		)

     if single_scale:
          _a         =				torch.tensor(
              [[-9.9_0_5_1, -4.2_5_4_1, -6.4_8_5_2], [-9.6_9_4_7, -4.0_8_5_4, -6.8_0_3_3], [-1_0.0_6_6_5, -5.8_4_7_0, -7.7_0_0_3]]		)
          _a         =				torch.tensor([[0.7_2_9_2, 0.4_9_9_1, 0.5_5_3_2], [0.7_9_5_9, 0.2_4_2_6, 0.4_2_3_6], [0.7_5_8_2, 0.3_5_1_8, 0.4_4_5_1]]		)

     if single_scale and dilation:
          _a         =				torch.tensor(
              [[-8.9_6_5_2, -4.1_0_7_4, -5.6_6_3_5], [-9.0_5_9_6, -4.9_4_4_7, -6.6_0_7_5], [-1_0.1_1_7_8, -4.5_2_7_5, -6.2_6_7_1]]		)
          _a         =				torch.tensor([[0.7_6_6_5, 0.4_1_3_0, 0.4_7_6_9], [0.8_3_6_4, 0.1_8_4_1, 0.3_3_9_1], [0.6_2_6_1, 0.3_8_9_5, 0.7_9_7_8]]		)

     if with_box_refine:
          _a         =				torch.tensor(
              [[-8.8_8_9_5, -5.4_1_8_7, -6.8_1_5_3], [-8.4_7_0_6, -6.1_6_6_8, -7.6_1_8_4], [-9.0_0_4_2, -5.5_3_5_9, -6.9_1_4_1]]		)
          _a         =				torch.tensor([[0.7_8_2_8, 0.2_2_0_8, 0.4_3_2_3], [0.0_8_9_2, 0.5_9_9_6, 0.1_3_1_9], [0.5_5_2_4, 0.6_3_8_9, 0.8_9_1_4]]		)

     if with_box_refine and two_stage:
          _a         =				torch.tensor(
              [[-6.7_1_0_8, -4.3_2_1_3, -6.3_7_7_7], [-8.9_0_1_4, -6.1_7_9_9, -6.7_2_4_0], [-6.9_3_1_5, -4.4_7_3_5, -6.2_2_9_8]]		)
          _a         =				torch.tensor([[0.2_5_8_3, 0.5_4_9_9, 0.4_6_8_3], [0.7_6_5_2, 0.9_0_6_8, 0.4_8_8_2], [0.5_4_9_0, 0.2_7_6_3, 0.0_5_6_4]]		)

     print('Logits:'					,    outputs.logits[0, :3, :3]		)

     assert torch.allclose(outputs.logits[0, :3, :3]					,    expected_logits.to(lowerCAmelCase__		)					,    atol=1E-4		)
     assert torch.allclose(outputs.pred_boxes[0, :3, :3]					,    expected_boxes.to(lowerCAmelCase__		)					,    atol=1E-4		)

     print('Everything ok!'		)

     # Save model and image processor
     logger.info(f'Saving PyTorch model and image processor to {pytorch_dump_folder_path}...'		)
     Path(lowerCAmelCase__		).mkdir(exist_ok=lowerCAmelCase__		)
     model.save_pretrained(lowerCAmelCase__		)
     image_processor.save_pretrained(lowerCAmelCase__		)

     # Push to hub
     if push_to_hub:
          _a         =				'deformable-detr'
          model_name += "-single-scale" if single_scale else ""
          model_name += "-dc5" if dilation else ""
          model_name += "-with-box-refine" if with_box_refine else ""
          model_name += "-two-stage" if two_stage else ""
          print('Pushing model to hub...'		)
          model.push_to_hub(repo_path_or_name=lowerCAmelCase__					,    organization='nielsr'					,    commit_message='Add model'		)


if __name__ == "__main__":
 a_       :     Optional[int] 							=     argparse.ArgumentParser()

 parser.add_argument(
     "--checkpoint_path",
     type=str,
     default="/home/niels/checkpoints/deformable_detr/r50_deformable_detr-checkpoint.pth",
     help="Path to Pytorch checkpoint (.pth file) you'd like to convert.",
 )
 parser.add_argument("--single_scale", action="store_true", help="Whether to set config.num_features_levels = 1.")
 parser.add_argument("--dilation", action="store_true", help="Whether to set config.dilation=True.")
 parser.add_argument("--with_box_refine", action="store_true", help="Whether to set config.with_box_refine=True.")
 parser.add_argument("--two_stage", action="store_true", help="Whether to set config.two_stage=True.")
 parser.add_argument(
     "--pytorch_dump_folder_path",
     default=None,
     type=str,
     required=True,
     help="Path to the folder to output PyTorch model.",
 )
 parser.add_argument(
     "--push_to_hub", action="store_true", help="Whether or not to push the converted model to the 🤗 hub."
 )
 a_       :     Union[str, Any] 							=     parser.parse_args()
 convert_deformable_detr_checkpoint(
     args.checkpoint_path,
     args.single_scale,
     args.dilation,
     args.with_box_refine,
     args.two_stage,
     args.pytorch_dump_folder_path,
     args.push_to_hub,
 )





'''simple docstring'''






import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


a_       :     Tuple 							=     logging.get_logger(__name__)

a_       :     Optional[Any] 							=     {
    "vocab_file": "vocab.json",
    "merges_file": "merges.txt",
}

a_       :     str 							=     {
    "vocab_file": {"openai-gpt": "https://huggingface.co/openai-gpt/resolve/main/vocab.json"},
    "merges_file": {"openai-gpt": "https://huggingface.co/openai-gpt/resolve/main/merges.txt"},
}

a_       :     Union[str, Any] 							=     {
    "openai-gpt": 5_1_2,
}






def 						_A					(lowerCAmelCase__	:Any		) ->	int:




     '''simple docstring'''



     _a         =				text.strip()
     if not text:
          return []
     _a         =				text.split()
     return tokens




class 			a  (  _SCREAMING_SNAKE_CASE							):
 def __init__(			self   ,					__magic_name__=True   ,					__magic_name__=None   ,					__magic_name__=True   ,					__magic_name__=None   ,					__magic_name__=True   ,					)       ->				Optional[Any]:
      if never_split is None:
           _a         =				[]
      _a         =				do_lower_case
      _a         =				set(__magic_name__    )
      _a         =				tokenize_chinese_chars
      _a         =				strip_accents
      _a         =				do_split_on_punc
 def 				__UpperCAmelCase		(			self   ,					__magic_name__   ,					__magic_name__=None    )       ->				List[str]:
      _a         =				self.never_split.union(set(__magic_name__    )    ) if never_split else self.never_split
      _a         =				self._clean_text(__magic_name__    )

      # This was added on November 1st, 2018 for the multilingual and Chinese
      # models. This is also applied to the English models now, but it doesn't
      # matter since the English models were not trained on any Chinese data
      # and generally don't have any Chinese data in them (there are Chinese
      # characters in the vocabulary because Wikipedia does have some Chinese
      # words in the English Wikipedia.).
      if self.tokenize_chinese_chars:
           _a         =				self._tokenize_chinese_chars(__magic_name__    )
      # prevents treating the same character with different unicode codepoints as different characters
      _a         =				unicodedata.normalize('NFC'   ,					__magic_name__    )
      _a         =				whitespace_tokenize(__magic_name__    )
      _a         =				[]
      for token in orig_tokens:
           if token not in never_split:
                if self.do_lower_case:
                     _a         =				token.lower()
                     if self.strip_accents is not False:
                          _a         =				self._run_strip_accents(__magic_name__    )
                elif self.strip_accents:
                     _a         =				self._run_strip_accents(__magic_name__    )
           split_tokens.extend(self._run_split_on_punc(__magic_name__   ,					__magic_name__    )    )

      _a         =				whitespace_tokenize(' '.join(__magic_name__    )    )
      return output_tokens
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				List[str]:
      _a         =				unicodedata.normalize('NFD'   ,					__magic_name__    )
      _a         =				[]
      for char in text:
           _a         =				unicodedata.category(__magic_name__    )
           if cat == "Mn":
                continue
           output.append(__magic_name__    )
      return "".join(__magic_name__    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__   ,					__magic_name__=None    )       ->				Dict:
      if not self.do_split_on_punc or (never_split is not None and text in never_split):
           return [text]
      _a         =				list(__magic_name__    )
      _a         =				0
      _a         =				True
      _a         =				[]
      while i < len(__magic_name__    ):
           _a         =				chars[i]
           if _is_punctuation(__magic_name__    ):
                output.append([char]    )
                _a         =				True
           else:
                if start_new_word:
                     output.append([]    )
                _a         =				False
                output[-1].append(__magic_name__    )
           i += 1

      return ["".join(__magic_name__    ) for x in output]
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Tuple:
      _a         =				[]
      for char in text:
           _a         =				ord(__magic_name__    )
           if self._is_chinese_char(__magic_name__    ):
                output.append(' '    )
                output.append(__magic_name__    )
                output.append(' '    )
           else:
                output.append(__magic_name__    )
      return "".join(__magic_name__    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Union[str, Any]:
      # This defines a "chinese character" as anything in the CJK Unicode block:
      #   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
      #
      # Note that the CJK Unicode block is NOT all Japanese and Korean characters,
      # despite its name. The modern Korean Hangul alphabet is a different block,
      # as is Japanese Hiragana and Katakana. Those alphabets are used to write
      # space-separated words, so they are not treated specially and handled
      # like the all of the other languages.
      if (
          (cp >= 0x4E00 and cp <= 0x9FFF)
          or (cp >= 0x3400 and cp <= 0x4DBF)  #
          or (cp >= 0x2_0000 and cp <= 0x2_A6DF)  #
          or (cp >= 0x2_A700 and cp <= 0x2_B73F)  #
          or (cp >= 0x2_B740 and cp <= 0x2_B81F)  #
          or (cp >= 0x2_B820 and cp <= 0x2_CEAF)  #
          or (cp >= 0xF900 and cp <= 0xFAFF)
          or (cp >= 0x2_F800 and cp <= 0x2_FA1F)  #
      ):  #
           return True

      return False




 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      _a         =				[]
      for char in text:
           _a         =				ord(__magic_name__    )
           if cp == 0 or cp == 0xFFFD or _is_control(__magic_name__    ):
                continue
           if _is_whitespace(__magic_name__    ):
                output.append(' '    )
           else:
                output.append(__magic_name__    )
      return "".join(__magic_name__    )






def 						_A					(lowerCAmelCase__	:str		) ->	Tuple:




     '''simple docstring'''



     _a         =				set()
     _a         =				word[0]
     for char in word[1:]:
          pairs.add((prev_char, char)		)
          _a         =				char
     return pairs






def 						_A					(lowerCAmelCase__	:Tuple		) ->	Union[str, Any]:




     '''simple docstring'''



     _a         =				text.replace('—'					,    '-'		)
     _a         =				text.replace('–'					,    '-'		)
     _a         =				text.replace('―'					,    '-'		)
     _a         =				text.replace('…'					,    '...'		)
     _a         =				text.replace('´'					,    '\''		)
     _a         =				re.sub(r'(-+|~+|!+|"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)'					,    r' \1 '					,    lowerCAmelCase__		)
     _a         =				re.sub(r'\s*\n\s*'					,    ' \n '					,    lowerCAmelCase__		)
     _a         =				re.sub(r'[^\S\n]+'					,    ' '					,    lowerCAmelCase__		)
     return text.strip()






class 			a  (  _SCREAMING_SNAKE_CASE							):
 _lowerCAmelCase			       =						VOCAB_FILES_NAMES
 _lowerCAmelCase			       =						PRETRAINED_VOCAB_FILES_MAP
 _lowerCAmelCase			       =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _lowerCAmelCase			       =						["""input_ids""", """attention_mask"""]
 def __init__(			self   ,					__magic_name__   ,					__magic_name__   ,					__magic_name__="<unk>"   ,					**__magic_name__    )       ->				Any:
      super().__init__(unk_token=__magic_name__   ,					**__magic_name__    )

      try:
           import ftfy
           from spacy.lang.en import English

           _a         =				English()
           _a         =				_nlp.tokenizer
           _a         =				ftfy.fix_text
      except ImportError:
           logger.warning('ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.'    )
           _a         =				BasicTokenizer(do_lower_case=__magic_name__    )
           _a         =				None

      with open(__magic_name__   ,					encoding='utf-8'    ) as vocab_handle:
           _a         =				json.load(__magic_name__    )
      _a         =				{v: k for k, v in self.encoder.items()}
      with open(__magic_name__   ,					encoding='utf-8'    ) as merges_handle:
           _a         =				merges_handle.read().split('\n'    )[1:-1]
      _a         =				[tuple(merge.split()    ) for merge in merges]
      _a         =				dict(zip(__magic_name__   ,					range(len(__magic_name__    )    )    )    )
      _a         =				{}
 @property
 def 				__UpperCAmelCase		(			self    )       ->				Optional[int]:
      return True
 @property
 def 				__UpperCAmelCase		(			self    )       ->				int:
      return len(self.encoder    )
 def 				__UpperCAmelCase		(			self    )       ->				List[str]:
      return dict(self.encoder   ,					**self.added_tokens_encoder    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      _a         =				tuple(token[:-1]    ) + (token[-1] + '</w>',)
      if token in self.cache:
           return self.cache[token]
      _a         =				get_pairs(__magic_name__    )

      if not pairs:
           return token + "</w>"

      while True:
           _a         =				min(__magic_name__   ,					key=lambda __magic_name__    : self.bpe_ranks.get(__magic_name__   ,					float('inf'    )    )    )
           if bigram not in self.bpe_ranks:
                break
           _a   , _a         =				bigram
           _a         =				[]
           _a         =				0
           while i < len(__magic_name__    ):
                try:
                     _a         =				word.index(__magic_name__   ,					__magic_name__    )
                except ValueError:
                     new_word.extend(word[i:]    )
                     break
                else:
                     new_word.extend(word[i:j]    )
                     _a         =				j

                if word[i] == first and i < len(__magic_name__    ) - 1 and word[i + 1] == second:
                     new_word.append(first + second    )
                     i += 2
                else:
                     new_word.append(word[i]    )
                     i += 1
           _a         =				tuple(__magic_name__    )
           _a         =				new_word
           if len(__magic_name__    ) == 1:
                break
           else:
                _a         =				get_pairs(__magic_name__    )
      _a         =				' '.join(__magic_name__    )
      if word == "\n  </w>":
           _a         =				'\n</w>'
      _a         =				word
      return word
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Union[str, Any]:
      _a         =				[]
      if self.fix_text is None:
           # Using BERT's BasicTokenizer
           _a         =				self.nlp.tokenize(__magic_name__    )
           for token in text:
                split_tokens.extend(list(self.bpe(__magic_name__    ).split(' '    )    )    )
      else:
           # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
           _a         =				self.nlp(text_standardize(self.fix_text(__magic_name__    )    )    )
           for token in text:
                split_tokens.extend(list(self.bpe(token.text.lower()    ).split(' '    )    )    )
      return split_tokens
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      return self.encoder.get(__magic_name__   ,					self.encoder.get(self.unk_token    )    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      return self.decoder.get(__magic_name__   ,					self.unk_token    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Dict:
      _a         =				''.join(__magic_name__    ).replace('</w>'   ,					' '    ).strip()
      return out_string




 def 				__UpperCAmelCase		(			self   ,					__magic_name__   ,					__magic_name__ = None    )       ->				Tuple[str]:
      if not os.path.isdir(__magic_name__    ):
           logger.error(f'Vocabulary path ({save_directory}) should be a directory'    )
           return
      _a         =				os.path.join(
          __magic_name__   ,					(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']    )
      _a         =				os.path.join(
          __magic_name__   ,					(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']    )

      with open(__magic_name__   ,					'w'   ,					encoding='utf-8'    ) as f:
           f.write(json.dumps(self.encoder   ,					indent=2   ,					sort_keys=__magic_name__   ,					ensure_ascii=__magic_name__    ) + '\n'    )

      _a         =				0
      with open(__magic_name__   ,					'w'   ,					encoding='utf-8'    ) as writer:
           writer.write('#version: 0.2\n'    )
           for bpe_tokens, token_index in sorted(self.bpe_ranks.items()   ,					key=lambda __magic_name__    : kv[1]    ):
                if index != token_index:
                     logger.warning(
                         f'Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'
                         ' Please check that the tokenizer is not corrupted!'    )
                     _a         =				token_index
                writer.write(' '.join(__magic_name__    ) + '\n'    )
                index += 1

      return vocab_file, merge_file

