




"""simple docstring"""


import argparse
from collections import OrderedDict
from pathlib import Path

import torch

from transformers import (
    VisualBertConfig,
    VisualBertForMultipleChoice,
    VisualBertForPreTraining,
    VisualBertForQuestionAnswering,
    VisualBertForVisualReasoning,
)
from transformers.utils import logging


logging.set_verbosity_info()
A_						=				logging.get_logger(__name__)

A_						=				[
    ('''bert.bert''', '''visual_bert'''),
    ('''bert.cls''', '''cls'''),
    ('''bert.classifier''', '''cls'''),
    ('''token_type_embeddings_visual''', '''visual_token_type_embeddings'''),
    ('''position_embeddings_visual''', '''visual_position_embeddings'''),
    ('''projection''', '''visual_projection'''),
]

A_						=				[
    '''nlvr2_coco_pre_trained.th''',
    '''nlvr2_fine_tuned.th''',
    '''nlvr2_pre_trained.th''',
    '''vcr_coco_pre_train.th''',
    '''vcr_fine_tune.th''',
    '''vcr_pre_train.th''',
    '''vqa_coco_pre_trained.th''',
    '''vqa_fine_tuned.th''',
    '''vqa_pre_trained.th''',
]


def 					UpperCAmelCase__       (snake_case__							:					str    ):
  """simple docstring"""

  _snake_case							:		str								=  torch.load(snake_case__		,   map_location="""cpu"""    )
  return sd


def 					UpperCAmelCase__       (snake_case__							:					int		,   snake_case__							:					str		,   snake_case__							:					int=rename_keys_prefix    ):
  """simple docstring"""

  _snake_case							:		Optional[int]								=  OrderedDict()
  _snake_case							:		List[str]								=  torch.arange(config.max_position_embeddings    ).expand((1, -1)    )
  # detector_d = OrderedDict()
  for key in d:
    if "detector" in key:
      # detector_d[key.replace('detector.','')] = d[key]
      continue
    _snake_case							:		Any								=  key
    for name_pair in rename_keys_prefix:
      _snake_case							:		str								=  new_key.replace(name_pair[0]		,   name_pair[1]    )
    _snake_case							:		Union[str, Any]								=  d[key]
    if key == "bert.cls.predictions.decoder.weight":
      # Old bert code didn't have `decoder.bias`, but was added separately
      _snake_case							:		List[Any]								=  new_d["""cls.predictions.bias"""]
  return new_d


@torch.no_grad()
def 					UpperCAmelCase__       (snake_case__							:					Any		,   snake_case__							:					str    ):
  """simple docstring"""

  assert (
      checkpoint_path.split("""/"""    )[-1] in ACCEPTABLE_CHECKPOINTS
  ), F"The checkpoint provided must be in {ACCEPTABLE_CHECKPOINTS}."

  # Get Config
  if "pre" in checkpoint_path:
    _snake_case							:		Union[str, Any]								=  """pretraining"""
    if "vcr" in checkpoint_path:
      _snake_case							:		int								=  {"""visual_embedding_dim""": 5_12}
    elif "vqa_advanced" in checkpoint_path:
      _snake_case							:		Optional[int]								=  {"""visual_embedding_dim""": 20_48}
    elif "vqa" in checkpoint_path:
      _snake_case							:		Any								=  {"""visual_embedding_dim""": 20_48}
    elif "nlvr" in checkpoint_path:
      _snake_case							:		List[Any]								=  {"""visual_embedding_dim""": 10_24}
    else:
      raise NotImplementedError(F"No implementation found for `{checkpoint_path}`."    )
  else:
    if "vcr" in checkpoint_path:
      _snake_case							:		Dict								=  {"""visual_embedding_dim""": 5_12}
      _snake_case							:		Union[str, Any]								=  """multichoice"""
    elif "vqa_advanced" in checkpoint_path:
      _snake_case							:		Dict								=  {"""visual_embedding_dim""": 20_48}
      _snake_case							:		Optional[Any]								=  """vqa_advanced"""
    elif "vqa" in checkpoint_path:
      _snake_case							:		Optional[Any]								=  {"""visual_embedding_dim""": 20_48, """num_labels""": 31_29}
      _snake_case							:		List[str]								=  """vqa"""
    elif "nlvr" in checkpoint_path:
      _snake_case							:		List[Any]								=  {
          """visual_embedding_dim""": 10_24,
          """num_labels""": 2,
      }
      _snake_case							:		List[Any]								=  """nlvr"""

  _snake_case							:		Optional[int]								=  VisualBertConfig(**snake_case__    )

  # Load State Dict
  _snake_case							:		List[str]								=  load_state_dict(snake_case__    )

  _snake_case							:		Any								=  get_new_dict(snake_case__		,   snake_case__    )

  if model_type == "pretraining":
    _snake_case							:		Union[str, Any]								=  VisualBertForPreTraining(snake_case__    )
  elif model_type == "vqa":
    _snake_case							:		Any								=  VisualBertForQuestionAnswering(snake_case__    )
  elif model_type == "nlvr":
    _snake_case							:		Optional[Any]								=  VisualBertForVisualReasoning(snake_case__    )
  elif model_type == "multichoice":
    _snake_case							:		Optional[int]								=  VisualBertForMultipleChoice(snake_case__    )

  model.load_state_dict(snake_case__    )
  # Save Checkpoints
  Path(snake_case__    ).mkdir(exist_ok=snake_case__    )
  model.save_pretrained(snake_case__    )


if __name__ == "__main__":
     A_						=				argparse.ArgumentParser()
     # Required parameters
     parser.add_argument('''orig_checkpoint_path''', type=str, help='''A path to .th on local filesystem.''')
     parser.add_argument('''pytorch_dump_folder_path''', type=str, help='''Path to the output PyTorch model.''')
     A_						=				parser.parse_args()
     convert_visual_bert_checkpoint(args.orig_checkpoint_path, args.pytorch_dump_folder_path)








"""simple docstring"""


import os
import zipfile

import requests
from get_ci_error_statistics import download_artifact, get_artifacts_links


def 					UpperCAmelCase__       (snake_case__							:					Optional[int]		,   snake_case__							:					Any=7    ):
  """simple docstring"""

  _snake_case							:		Any								=  None
  if token is not None:
    _snake_case							:		Any								=  {"""Accept""": """application/vnd.github+json""", """Authorization""": F"Bearer {token}"}

  # The id of a workflow (not of a workflow run)
  _snake_case							:		List[str]								=  """636036"""

  _snake_case							:		Union[str, Any]								=  F"https://api.github.com/repos/huggingface/transformers/actions/workflows/{workflow_id}/runs"
  # On `main` branch + event being `schedule` + not returning PRs + only `num_runs` results
  url += F"?branch=main&event=schedule&exclude_pull_requests=true&per_page={num_runs}"

  _snake_case							:		str								=  requests.get(snake_case__		,   headers=snake_case__    ).json()

  return result["workflow_runs"]


def 					UpperCAmelCase__       (snake_case__							:					Optional[Any]    ):
  """simple docstring"""

  _snake_case							:		str								=  get_daily_ci_runs(snake_case__    )
  _snake_case							:		str								=  None
  for workflow_run in workflow_runs:
    if workflow_run["status"] == "completed":
      _snake_case							:		List[str]								=  workflow_run["""id"""]
      break

  return workflow_run_id


def 					UpperCAmelCase__       (snake_case__							:					str		,   snake_case__							:					Union[str, Any]		,   snake_case__							:					Optional[int]    ):
  """simple docstring"""

  _snake_case							:		Optional[Any]								=  get_last_daily_ci_runs(snake_case__    )
  if workflow_run_id is not None:
    _snake_case							:		Optional[Any]								=  get_artifacts_links(worflow_run_id=snake_case__		,   token=snake_case__    )
    for artifact_name in artifact_names:
      if artifact_name in artifacts_links:
        _snake_case							:		Optional[int]								=  artifacts_links[artifact_name]
        download_artifact(
            artifact_name=snake_case__		,   artifact_url=snake_case__		,   output_dir=snake_case__		,   token=snake_case__    )






def 					UpperCAmelCase__       (snake_case__							:					Union[str, Any]		,   snake_case__							:					List[str]		,   snake_case__							:					int    ):
  """simple docstring"""

  get_last_daily_ci_artifacts(snake_case__		,   snake_case__		,   snake_case__    )

  _snake_case							:		int								=  {}
  for artifact_name in artifact_names:
    _snake_case							:		int								=  os.path.join(snake_case__		,   F"{artifact_name}.zip"    )
    if os.path.isfile(snake_case__    ):
      _snake_case							:		Tuple								=  {}
      with zipfile.ZipFile(snake_case__    ) as z:
        for filename in z.namelist():
          if not os.path.isdir(snake_case__    ):
            # read the file
            with z.open(snake_case__    ) as f:
              _snake_case							:		Any								=  f.read().decode("""UTF-8"""    )

  return results


