


"""simple docstring"""


import os
import sys
import warnings
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen
from ..table import array_cast
from ..utils.file_utils import is_local_path
from ..utils.py_utils import first_non_null_value, no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
	import PIL.Image

	from .features import FeatureType


A     :   Optional[List[str]]			    =			None
A     :   Any			    =			"<" if sys.byteorder == "little" else ">"
# Origin: https://github.com/python-pillow/Pillow/blob/698951e19e19972aeed56df686868f1329981c12/src/PIL/Image.py#L3126 minus "|i1" which values are not preserved correctly when saving and loading an image
A     :   int			    =			[
    np.dtype("|b1"),
    np.dtype("|u1"),
    np.dtype("<u2"),
    np.dtype(">u2"),
    np.dtype("<i2"),
    np.dtype(">i2"),
    np.dtype("<u4"),
    np.dtype(">u4"),
    np.dtype("<i4"),
    np.dtype(">i4"),
    np.dtype("<f4"),
    np.dtype(">f4"),
    np.dtype("<f8"),
    np.dtype(">f8"),
]
@dataclass
class   _UpperCamelCase  :
							'''simple docstring'''







							__UpperCAmelCase						:    bool				      =True
							__UpperCAmelCase						:    Optional[str]				      =None
							# Automatically constructed
							__UpperCAmelCase						:    ClassVar[str]				      ="PIL.Image.Image"
							__UpperCAmelCase						:    ClassVar[Any]				      =pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}		)
							__UpperCAmelCase						:    str				      =field(default="""Image"""       ,init=lowerCAmelCase__       ,repr=lowerCAmelCase__		)

							def __call__(     self	):
										return self.pa_type

							def 			snake_case				(     self    ,   __a	):
										if config.PIL_AVAILABLE:
													import PIL.Image
										else:
													raise ImportError("To support encoding images, please install 'Pillow'."	)

										if isinstance(__a    ,   __a	):
													__lowerCAmelCase				    =			np.array(__a	)

										if isinstance(__a    ,   __a	):
													return {"path": value, "bytes": None}
										elif isinstance(__a    ,   __a	):
													return {"path": None, "bytes": value}
										elif isinstance(__a    ,   np.ndarray	):
													# convert the image array to PNG/TIFF bytes
													return encode_np_array(__a	)
										elif isinstance(__a    ,   PIL.Image.Image	):
													# convert the PIL image to bytes (default format is PNG/TIFF)
													return encode_pil_image(__a	)
										elif value.get("path"	) is not None and os.path.isfile(value["path"]	):
													# we set "bytes": None to not duplicate the data if they're already available locally
													return {"bytes": None, "path": value.get("path"	)}
										elif value.get("bytes"	) is not None or value.get("path"	) is not None:
													# store the image bytes, and path is used to infer the image format using the file extension
													return {"bytes": value.get("bytes"	), "path": value.get("path"	)}
										else:
													raise ValueError(
													    f"An image sample should have one of 'path' or 'bytes' but they are missing or None in {value}."	)

							def 			snake_case				(     self    ,   __a    ,   __a=None	):
										if not self.decode:
													raise RuntimeError("Decoding is disabled for this feature. Please use Image(decode=True) instead."	)

										if config.PIL_AVAILABLE:
													import PIL.Image
										else:
													raise ImportError("To support decoding images, please install 'Pillow'."	)

										if token_per_repo_id is None:
													__lowerCAmelCase				    =			{}

										__lowerCAmelCase    ,	__lowerCAmelCase				    =			value["path"], value["bytes"]
										if bytes_ is None:
													if path is None:
																raise ValueError(f"An image should have one of 'path' or 'bytes' but both are None in {value}."	)
													else:
																if is_local_path(__a	):
																			__lowerCAmelCase				    =			PIL.Image.open(__a	)
																else:
																			__lowerCAmelCase				    =			path.split("::"	)[-1]
																			try:
																						__lowerCAmelCase				    =			string_to_dict(__a    ,   config.HUB_DATASETS_URL	)["repo_id"]
																						__lowerCAmelCase				    =			token_per_repo_id.get(__a	)
																			except ValueError:
																						__lowerCAmelCase				    =			None
																			with xopen(__a    ,   "rb"    ,   use_auth_token=__a	) as f:
																						__lowerCAmelCase				    =			BytesIO(f.read()	)
																			__lowerCAmelCase				    =			PIL.Image.open(bytes_	)
										else:
													__lowerCAmelCase				    =			PIL.Image.open(BytesIO(bytes_	)	)
										image.load()  # to avoid "Too many open files" errors
										return image

							def 			snake_case				(     self	):
										from .features import Value

										return (
										    self
										    if self.decode
										    else {
										        "bytes": Value("binary"	),
										        "path": Value("string"	),
										    }
										)

							def 			snake_case				(     self    ,   __a	):
										if pa.types.is_string(storage.type	):
													__lowerCAmelCase				    =			pa.array([None] * len(__a	)    ,   type=pa.binary()	)
													__lowerCAmelCase				    =			pa.StructArray.from_arrays([bytes_array, storage]    ,   ["bytes", "path"]    ,   mask=storage.is_null()	)
										elif pa.types.is_binary(storage.type	):
													__lowerCAmelCase				    =			pa.array([None] * len(__a	)    ,   type=pa.string()	)
													__lowerCAmelCase				    =			pa.StructArray.from_arrays([storage, path_array]    ,   ["bytes", "path"]    ,   mask=storage.is_null()	)
										elif pa.types.is_struct(storage.type	):
													if storage.type.get_field_index("bytes"	) >= 0:
																__lowerCAmelCase				    =			storage.field("bytes"	)
													else:
																__lowerCAmelCase				    =			pa.array([None] * len(__a	)    ,   type=pa.binary()	)
													if storage.type.get_field_index("path"	) >= 0:
																__lowerCAmelCase				    =			storage.field("path"	)
													else:
																__lowerCAmelCase				    =			pa.array([None] * len(__a	)    ,   type=pa.string()	)
													__lowerCAmelCase				    =			pa.StructArray.from_arrays([bytes_array, path_array]    ,   ["bytes", "path"]    ,   mask=storage.is_null()	)
										elif pa.types.is_list(storage.type	):
													__lowerCAmelCase				    =			pa.array(
													    [encode_np_array(np.array(__a	)	)["bytes"] if arr is not None else None for arr in storage.to_pylist()]    ,   type=pa.binary()    ,   )
													__lowerCAmelCase				    =			pa.array([None] * len(__a	)    ,   type=pa.string()	)
													__lowerCAmelCase				    =			pa.StructArray.from_arrays(
													    [bytes_array, path_array]    ,   ["bytes", "path"]    ,   mask=bytes_array.is_null()	)
										return array_cast(__a    ,   self.pa_type	)



							def 			snake_case				(     self    ,   __a	):

										@no_op_if_value_is_null
										def path_to_bytes(__a	):
													with xopen(__a    ,   "rb"	) as f:
																__lowerCAmelCase				    =			f.read()
													return bytes_

										__lowerCAmelCase				    =			pa.array(
										    [
										        (path_to_bytes(x["path"]	) if x["bytes"] is None else x["bytes"]) if x is not None else None
										        for x in storage.to_pylist()
										    ]    ,   type=pa.binary()    ,   )
										__lowerCAmelCase				    =			pa.array(
										    [os.path.basename(__a	) if path is not None else None for path in storage.field("path"	).to_pylist()]    ,   type=pa.string()    ,   )
										__lowerCAmelCase				    =			pa.StructArray.from_arrays([bytes_array, path_array]    ,   ["bytes", "path"]    ,   mask=bytes_array.is_null()	)
										return array_cast(__a    ,   self.pa_type	)


def _lowerCamelCase	(		):

			'''simple docstring'''




			if config.PIL_AVAILABLE:
						import PIL.Image
			else:
						raise ImportError("To support encoding images, please install 'Pillow'."      )

			global _IMAGE_COMPRESSION_FORMATS
			if _IMAGE_COMPRESSION_FORMATS is None:
						PIL.Image.init()
						__lowerCAmelCase				    =			list(set(PIL.Image.OPEN.keys()      ) & set(PIL.Image.SAVE.keys()      )      )
			return _IMAGE_COMPRESSION_FORMATS


def _lowerCamelCase	(		_UpperCamelCase      ):

			'''simple docstring'''




			__lowerCAmelCase				    =			BytesIO()
			if image.format in list_image_compression_formats():
						__lowerCAmelCase				    =			image.format
			else:
						__lowerCAmelCase				    =			"PNG" if image.mode in ["1", "L", "LA", "RGB", "RGBA"] else "TIFF"
			image.save(_UpperCamelCase    , format=_UpperCamelCase      )
			return buffer.getvalue()


def _lowerCamelCase	(		_UpperCamelCase      ):

			'''simple docstring'''




			if hasattr(_UpperCamelCase    , "filename"      ) and image.filename != "":
						return {"path": image.filename, "bytes": None}
			else:
						return {"path": None, "bytes": image_to_bytes(_UpperCamelCase      )}


def _lowerCamelCase	(		_UpperCamelCase      ):

			'''simple docstring'''




			if config.PIL_AVAILABLE:
						import PIL.Image
			else:
						raise ImportError("To support encoding images, please install 'Pillow'."      )

			__lowerCAmelCase				    =			array.dtype
			__lowerCAmelCase				    =			dtype.byteorder if dtype.byteorder != "=" else _NATIVE_BYTEORDER
			__lowerCAmelCase				    =			dtype.kind
			__lowerCAmelCase				    =			dtype.itemsize

			__lowerCAmelCase				    =			None

			# Multi-channel array case (only np.dtype("|u1") is allowed)
			if array.shape[2:]:
						__lowerCAmelCase				    =			np.dtype("|u1"      )
						if dtype_kind not in ["u", "i"]:
									raise TypeError(
									    f"Unsupported array dtype {dtype} for image encoding. Only {dest_dtype} is supported for multi-channel arrays."      )
						if dtype is not dest_dtype:
									warnings.warn(f"Downcasting array dtype {dtype} to {dest_dtype} to be compatible with 'Pillow'"      )
    # Exact match
			elif dtype in _VALID_IMAGE_ARRAY_DTPYES:
						__lowerCAmelCase				    =			dtype
			else:  # Downcast the type within the kind (np.can_cast(from_type, to_type, casting="same_kind") doesn't behave as expected, so do it manually)
						while dtype_itemsize >= 1:
									__lowerCAmelCase				    =			dtype_byteorder + dtype_kind + str(_UpperCamelCase      )
									__lowerCAmelCase				    =			np.dtype(_UpperCamelCase      )
									if dest_dtype in _VALID_IMAGE_ARRAY_DTPYES:
												warnings.warn(f"Downcasting array dtype {dtype} to {dest_dtype} to be compatible with 'Pillow'"      )
												break
									else:
												dtype_itemsize //= 2

			if dest_dtype is None:
						raise TypeError(
						    f"Cannot convert dtype {dtype} to a valid image dtype. Valid image dtypes: {_VALID_IMAGE_ARRAY_DTPYES}"      )
			__lowerCAmelCase				    =			PIL.Image.fromarray(array.astype(_UpperCamelCase      )      )
			return {"path": None, "bytes": image_to_bytes(_UpperCamelCase      )}


def _lowerCamelCase	(		_UpperCamelCase      ):

			'''simple docstring'''




			if config.PIL_AVAILABLE:
						import PIL.Image
			else:
						raise ImportError("To support encoding images, please install 'Pillow'."      )

			if objs:
						__lowerCAmelCase    ,	__lowerCAmelCase				    =			first_non_null_value(_UpperCamelCase      )
						if isinstance(_UpperCamelCase    , _UpperCamelCase      ):
									return [{"path": obj, "bytes": None} if obj is not None else None for obj in objs]
						if isinstance(_UpperCamelCase    , np.ndarray      ):
									__lowerCAmelCase				    =			no_op_if_value_is_null(_UpperCamelCase      )
									return [obj_to_image_dict_func(_UpperCamelCase      ) for obj in objs]
						elif isinstance(_UpperCamelCase    , PIL.Image.Image      ):
									__lowerCAmelCase				    =			no_op_if_value_is_null(_UpperCamelCase      )
									return [obj_to_image_dict_func(_UpperCamelCase      ) for obj in objs]
						else:
									return objs
			else:
						return objs








"""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     :   Tuple			    =			logging.get_logger(__name__)

A     :   Tuple			    =			[
    ("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     :   Optional[Any]			    =			[
    "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 _lowerCamelCase	(		_UpperCamelCase      ):

			'''simple docstring'''




			__lowerCAmelCase				    =			torch.load(_UpperCamelCase    , map_location="cpu"      )
			return sd


def _lowerCamelCase	(		_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase=rename_keys_prefix      ):

			'''simple docstring'''




			__lowerCAmelCase				    =			OrderedDict()
			__lowerCAmelCase				    =			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
						__lowerCAmelCase				    =			key
						for name_pair in rename_keys_prefix:
									__lowerCAmelCase				    =			new_key.replace(name_pair[0]    , name_pair[1]      )
						__lowerCAmelCase				    =			d[key]
						if key == "bert.cls.predictions.decoder.weight":
									# Old bert code didn't have `decoder.bias`, but was added separately
									__lowerCAmelCase				    =			new_d["cls.predictions.bias"]
			return new_d


@torch.no_grad()
def _lowerCamelCase	(		_UpperCamelCase    , _UpperCamelCase      ):

			'''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:
						__lowerCAmelCase				    =			"pretraining"
						if "vcr" in checkpoint_path:
									__lowerCAmelCase				    =			{"visual_embedding_dim": 512}
						elif "vqa_advanced" in checkpoint_path:
									__lowerCAmelCase				    =			{"visual_embedding_dim": 2048}
						elif "vqa" in checkpoint_path:
									__lowerCAmelCase				    =			{"visual_embedding_dim": 2048}
						elif "nlvr" in checkpoint_path:
									__lowerCAmelCase				    =			{"visual_embedding_dim": 1024}
						else:
									raise NotImplementedError(f"No implementation found for `{checkpoint_path}`."      )
			else:
						if "vcr" in checkpoint_path:
									__lowerCAmelCase				    =			{"visual_embedding_dim": 512}
									__lowerCAmelCase				    =			"multichoice"
						elif "vqa_advanced" in checkpoint_path:
									__lowerCAmelCase				    =			{"visual_embedding_dim": 2048}
									__lowerCAmelCase				    =			"vqa_advanced"
						elif "vqa" in checkpoint_path:
									__lowerCAmelCase				    =			{"visual_embedding_dim": 2048, "num_labels": 3129}
									__lowerCAmelCase				    =			"vqa"
						elif "nlvr" in checkpoint_path:
									__lowerCAmelCase				    =			{
									    "visual_embedding_dim": 1024,
									    "num_labels": 2,
									}
									__lowerCAmelCase				    =			"nlvr"

			__lowerCAmelCase				    =			VisualBertConfig(**_UpperCamelCase      )

			# Load State Dict
			__lowerCAmelCase				    =			load_state_dict(_UpperCamelCase      )

			__lowerCAmelCase				    =			get_new_dict(_UpperCamelCase    , _UpperCamelCase      )

			if model_type == "pretraining":
						__lowerCAmelCase				    =			VisualBertForPreTraining(_UpperCamelCase      )
			elif model_type == "vqa":
						__lowerCAmelCase				    =			VisualBertForQuestionAnswering(_UpperCamelCase      )
			elif model_type == "nlvr":
						__lowerCAmelCase				    =			VisualBertForVisualReasoning(_UpperCamelCase      )
			elif model_type == "multichoice":
						__lowerCAmelCase				    =			VisualBertForMultipleChoice(_UpperCamelCase      )

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


if __name__ == "__main__":
	A     :   Union[str, Any]			    =			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     :   Optional[int]			    =			parser.parse_args()
	convert_visual_bert_checkpoint(args.orig_checkpoint_path, args.pytorch_dump_folder_path)




