


'''simple docstring'''




import sys
from pathlib import Path


UpperCamelCase__ :			Dict												=				Path(__file__).resolve().parents[3] / 'src'
sys.path.insert(1, str(git_repo_path))

import dataclasses  # noqa
import io  # noqa
import itertools  # noqa
import json  # noqa
import os  # noqa
import unittest  # noqa
from copy import deepcopy  # noqa

from parameterized import parameterized  # noqa
from transformers import TrainingArguments, is_torch_available  # noqa
from transformers.deepspeed import is_deepspeed_available  # noqa
from transformers.file_utils import WEIGHTS_NAME  # noqa
from transformers.testing_utils import (  # noqa
    CaptureLogger,
    ExtendSysPath,
    TestCasePlus,
    execute_subprocess_async,
    get_gpu_count,
    mockenv_context,
    require_deepspeed,
    require_torch_gpu,
    require_torch_multi_gpu,
    slow,
)
from transformers.trainer_utils import set_seed  # noqa


set_seed(42)

UpperCamelCase__ :			Optional[int]												=				{'base': 'patrickvonplaten/wav2vec2_tiny_random', 'robust': 'patrickvonplaten/wav2vec2_tiny_random_robust'}

UpperCamelCase__ :			Union[str, Any]												=				'zero2'
UpperCamelCase__ :			Union[str, Any]												=				'zero3'
UpperCamelCase__ :			str												=				[ZEROa, ZEROa]


def        lowerCAmelCase_     (			_lowerCamelCase:     Tuple ,   _lowerCamelCase:     Dict ,   _lowerCamelCase:     List[Any]      ):
		__SCREAMING_SNAKE_CASE :	Any					 =	parameterized.to_safe_name("""_""".join(str(lowerCAmelCase_      ) for x in param.args      )      )
		return F"{func.__name__}_{param_based_name}"


# Cartesian-product of zero stages with models to test
UpperCamelCase__ :			List[str]												=				list(itertools.product(stages, models.keys()))






@slow
@require_deepspeed
@require_torch_gpu
class 		_UpperCamelCase						(							a__					):


			'''simple docstring'''

			@parameterized.expand(lowercase__							,  name_func=lowercase__		)
			def UpperCamelCase__      (							self  :				Optional[Any]							,  lowerCAmelCase__  :				Tuple							,  lowerCAmelCase__  :				List[Any]		):





					"""simple docstring"""






					self.run_and_check(
					    stage=lowercase__							,  model=lowercase__							,  distributed=lowercase__							,  fpaa=lowercase__							,  )

			@require_torch_multi_gpu
			@parameterized.expand(lowercase__							,  name_func=lowercase__		)
			def UpperCamelCase__      (							self  :				Union[str, Any]							,  lowerCAmelCase__  :				Dict							,  lowerCAmelCase__  :				Union[str, Any]		):





					"""simple docstring"""






					self.run_and_check(
					    stage=lowercase__							,  model=lowercase__							,  distributed=lowercase__							,  fpaa=lowercase__							,  )

			@parameterized.expand(lowercase__							,  name_func=lowercase__		)
			def UpperCamelCase__      (							self  :				int							,  lowerCAmelCase__  :				Dict							,  lowerCAmelCase__  :				List[Any]		):





					"""simple docstring"""






					self.run_and_check(
					    stage=lowercase__							,  model=lowercase__							,  distributed=lowercase__							,  fpaa=lowercase__							,  )

			@require_torch_multi_gpu
			@parameterized.expand(lowercase__							,  name_func=lowercase__		)
			def UpperCamelCase__      (							self  :				str							,  lowerCAmelCase__  :				Dict							,  lowerCAmelCase__  :				Union[str, Any]		):





					"""simple docstring"""






					self.run_and_check(
					    stage=lowercase__							,  model=lowercase__							,  distributed=lowercase__							,  fpaa=lowercase__							,  )

			def UpperCamelCase__      (							self  :				Optional[int]							,  lowerCAmelCase__  :				Dict		):





					"""simple docstring"""






					pass

			def UpperCamelCase__      (							self  :				Dict							,  lowerCAmelCase__  :				List[Any]							,  lowerCAmelCase__  :				Optional[Any]							,  lowerCAmelCase__  :				Optional[int] = 1_0							,  lowerCAmelCase__  :				Tuple = True							,  lowerCAmelCase__  :				List[str] = True							,  lowerCAmelCase__  :				Dict = True							,  ):





					"""simple docstring"""






					__SCREAMING_SNAKE_CASE :	Dict					 =	models[model]

					__SCREAMING_SNAKE_CASE :	int					 =	self.run_trainer(
					    stage=lowercase__							,  model_name=lowercase__							,  eval_steps=lowercase__							,  num_train_epochs=1							,  distributed=lowercase__							,  fpaa=lowercase__							,  )

					self.do_checks(lowercase__		)

					return output_dir

			def UpperCamelCase__      (							self  :				Dict							,  lowerCAmelCase__  :				List[str]							,  lowerCAmelCase__  :				List[str]							,  lowerCAmelCase__  :				Union[str, Any] = 1_0							,  lowerCAmelCase__  :				Optional[int] = 1							,  lowerCAmelCase__  :				Dict = True							,  lowerCAmelCase__  :				List[str] = True							,  ):





					"""simple docstring"""






					__SCREAMING_SNAKE_CASE :	Any					 =	self.get_auto_remove_tmp_dir("""./xxx"""							,  after=lowercase__		)
					__SCREAMING_SNAKE_CASE :	str					 =	F"\n            --model_name_or_path {model_name}\n            --dataset_name hf-internal-testing/librispeech_asr_dummy\n            --dataset_config_name clean\n            --train_split_name validation\n            --validation_split_name validation\n            --output_dir {output_dir}\n            --num_train_epochs {str(lowercase__		)}\n            --per_device_train_batch_size 2\n            --per_device_eval_batch_size 2\n            --evaluation_strategy steps\n            --learning_rate 5e-4\n            --warmup_steps 8\n            --orthography timit\n            --preprocessing_num_workers 1\n            --group_by_length\n            --freeze_feature_extractor\n            --report_to none\n            --save_steps 0\n            --eval_steps {eval_steps}\n            --report_to none\n        ".split()

					if fpaa:
							args.extend(["""--fp16"""]		)

					# currently ds_config_wav2vec2_zero.json requires "zero_optimization.find_unused_parameters": true,
					# hence the separate config files
					__SCREAMING_SNAKE_CASE :	Dict					 =	F"--deepspeed {self.test_file_dir_str}/ds_config_wav2vec2_{stage}.json".split()
					__SCREAMING_SNAKE_CASE :	Any					 =	[F"{self.examples_dir_str}/research_projects/wav2vec2/run_asr.py"]
					__SCREAMING_SNAKE_CASE :	Tuple					 =	self.get_launcher(lowercase__		)

					__SCREAMING_SNAKE_CASE :	List[Any]					 =	launcher + script + args + ds_args
					# keep for quick debug
					# print(" ".join([f"\nPYTHONPATH={self.src_dir_str}"] +cmd)); die
					execute_subprocess_async(lowercase__							,  env=self.get_env()		)

					return output_dir



			def UpperCamelCase__      (							self  :				Any							,  lowerCAmelCase__  :				Union[str, Any]=False		):





					"""simple docstring"""






					__SCREAMING_SNAKE_CASE :	Tuple					 =	min(2							,  get_gpu_count()		) if distributed else 1
					return F"deepspeed --num_nodes 1 --num_gpus {num_gpus}".split()




'''simple docstring'''




import argparse
import json
from pathlib import Path

import requests
import timm
import torch
from huggingface_hub import hf_hub_download
from PIL import Image

from transformers import AutoImageProcessor, SwinvaConfig, SwinvaForImageClassification


def        lowerCAmelCase_     (			_lowerCamelCase:     Tuple      ):
		__SCREAMING_SNAKE_CASE :	List[Any]					 =	SwinvaConfig()
		__SCREAMING_SNAKE_CASE :	List[Any]					 =	swinva_name.split("""_"""      )

		__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	name_split[1]
		if "to" in name_split[3]:
				__SCREAMING_SNAKE_CASE :	Dict					 =	int(name_split[3][-3:]      )
		else:
				__SCREAMING_SNAKE_CASE :	str					 =	int(name_split[3]      )
		if "to" in name_split[2]:
				__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	int(name_split[2][-2:]      )
		else:
				__SCREAMING_SNAKE_CASE :	Optional[int]					 =	int(name_split[2][6:]      )

		if model_size == "tiny":
				__SCREAMING_SNAKE_CASE :	Dict					 =	96
				__SCREAMING_SNAKE_CASE :	List[str]					 =	(2, 2, 6, 2)
				__SCREAMING_SNAKE_CASE :	List[Any]					 =	(3, 6, 12, 24)
		elif model_size == "small":
				__SCREAMING_SNAKE_CASE :	List[str]					 =	96
				__SCREAMING_SNAKE_CASE :	int					 =	(2, 2, 18, 2)
				__SCREAMING_SNAKE_CASE :	int					 =	(3, 6, 12, 24)
		elif model_size == "base":
				__SCREAMING_SNAKE_CASE :	int					 =	1_28
				__SCREAMING_SNAKE_CASE :	str					 =	(2, 2, 18, 2)
				__SCREAMING_SNAKE_CASE :	Optional[int]					 =	(4, 8, 16, 32)
		else:
				__SCREAMING_SNAKE_CASE :	List[str]					 =	1_92
				__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	(2, 2, 18, 2)
				__SCREAMING_SNAKE_CASE :	Dict					 =	(6, 12, 24, 48)

		if "to" in swinva_name:
				__SCREAMING_SNAKE_CASE :	int					 =	(12, 12, 12, 6)

		if ("22k" in swinva_name) and ("to" not in swinva_name):
				__SCREAMING_SNAKE_CASE :	int					 =	2_18_41
				__SCREAMING_SNAKE_CASE :	str					 =	"""huggingface/label-files"""
				__SCREAMING_SNAKE_CASE :	List[str]					 =	"""imagenet-22k-id2label.json"""
				__SCREAMING_SNAKE_CASE :	List[str]					 =	json.load(open(hf_hub_download(_lowerCamelCase ,   _lowerCamelCase ,   repo_type="""dataset"""      ) ,   """r"""      )      )
				__SCREAMING_SNAKE_CASE :	List[Any]					 =	{int(_lowerCamelCase      ): v for k, v in idalabel.items()}
				__SCREAMING_SNAKE_CASE :	Optional[int]					 =	idalabel
				__SCREAMING_SNAKE_CASE :	str					 =	{v: k for k, v in idalabel.items()}

		else:
				__SCREAMING_SNAKE_CASE :	str					 =	10_00
				__SCREAMING_SNAKE_CASE :	Optional[int]					 =	"""huggingface/label-files"""
				__SCREAMING_SNAKE_CASE :	Any					 =	"""imagenet-1k-id2label.json"""
				__SCREAMING_SNAKE_CASE :	Optional[int]					 =	json.load(open(hf_hub_download(_lowerCamelCase ,   _lowerCamelCase ,   repo_type="""dataset"""      ) ,   """r"""      )      )
				__SCREAMING_SNAKE_CASE :	int					 =	{int(_lowerCamelCase      ): v for k, v in idalabel.items()}
				__SCREAMING_SNAKE_CASE :	Optional[int]					 =	idalabel
				__SCREAMING_SNAKE_CASE :	Dict					 =	{v: k for k, v in idalabel.items()}

		__SCREAMING_SNAKE_CASE :	Any					 =	img_size
		__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	num_classes
		__SCREAMING_SNAKE_CASE :	int					 =	embed_dim
		__SCREAMING_SNAKE_CASE :	Dict					 =	depths
		__SCREAMING_SNAKE_CASE :	str					 =	num_heads
		__SCREAMING_SNAKE_CASE :	int					 =	window_size

		return config


def        lowerCAmelCase_     (			_lowerCamelCase:     int      ):
		if "patch_embed.proj" in name:
				__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	name.replace("""patch_embed.proj""" ,   """embeddings.patch_embeddings.projection"""      )
		if "patch_embed.norm" in name:
				__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	name.replace("""patch_embed.norm""" ,   """embeddings.norm"""      )
		if "layers" in name:
				__SCREAMING_SNAKE_CASE :	Optional[int]					 =	"""encoder.""" + name
		if "attn.proj" in name:
				__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	name.replace("""attn.proj""" ,   """attention.output.dense"""      )
		if "attn" in name:
				__SCREAMING_SNAKE_CASE :	Any					 =	name.replace("""attn""" ,   """attention.self"""      )
		if "norm1" in name:
				__SCREAMING_SNAKE_CASE :	Optional[int]					 =	name.replace("""norm1""" ,   """layernorm_before"""      )
		if "norm2" in name:
				__SCREAMING_SNAKE_CASE :	Dict					 =	name.replace("""norm2""" ,   """layernorm_after"""      )
		if "mlp.fc1" in name:
				__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	name.replace("""mlp.fc1""" ,   """intermediate.dense"""      )
		if "mlp.fc2" in name:
				__SCREAMING_SNAKE_CASE :	List[Any]					 =	name.replace("""mlp.fc2""" ,   """output.dense"""      )
		if "q_bias" in name:
				__SCREAMING_SNAKE_CASE :	Tuple					 =	name.replace("""q_bias""" ,   """query.bias"""      )
		if "k_bias" in name:
				__SCREAMING_SNAKE_CASE :	Optional[int]					 =	name.replace("""k_bias""" ,   """key.bias"""      )
		if "v_bias" in name:
				__SCREAMING_SNAKE_CASE :	List[str]					 =	name.replace("""v_bias""" ,   """value.bias"""      )
		if "cpb_mlp" in name:
				__SCREAMING_SNAKE_CASE :	str					 =	name.replace("""cpb_mlp""" ,   """continuous_position_bias_mlp"""      )
		if name == "norm.weight":
				__SCREAMING_SNAKE_CASE :	Tuple					 =	"""layernorm.weight"""
		if name == "norm.bias":
				__SCREAMING_SNAKE_CASE :	Optional[int]					 =	"""layernorm.bias"""

		if "head" in name:
				__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	name.replace("""head""" ,   """classifier"""      )
		else:
				__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	"""swinv2.""" + name

		return name


def        lowerCAmelCase_     (			_lowerCamelCase:     int ,   _lowerCamelCase:     Optional[Any]      ):
		for key in orig_state_dict.copy().keys():
				__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	orig_state_dict.pop(_lowerCamelCase      )

				if "mask" in key:
						continue
				elif "qkv" in key:
						__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	key.split("""."""      )
						__SCREAMING_SNAKE_CASE :	List[str]					 =	int(key_split[1]      )
						__SCREAMING_SNAKE_CASE :	Dict					 =	int(key_split[3]      )
						__SCREAMING_SNAKE_CASE :	str					 =	model.swinva.encoder.layers[layer_num].blocks[block_num].attention.self.all_head_size

						if "weight" in key:
								__SCREAMING_SNAKE_CASE :	Optional[int]					 =	val[:dim, :]
								__SCREAMING_SNAKE_CASE :	str					 =	val[dim : dim * 2, :]
								__SCREAMING_SNAKE_CASE :	Dict					 =	val[-dim:, :]
						else:
								__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	val[:dim]
								__SCREAMING_SNAKE_CASE :	int					 =	val[
								    dim : dim * 2
								]
								__SCREAMING_SNAKE_CASE :	int					 =	val[-dim:]
				else:
						__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	val

		return orig_state_dict


def        lowerCAmelCase_     (			_lowerCamelCase:     Tuple ,   _lowerCamelCase:     int      ):
		__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	timm.create_model(_lowerCamelCase ,   pretrained=_lowerCamelCase      )
		timm_model.eval()

		__SCREAMING_SNAKE_CASE :	int					 =	get_swinva_config(_lowerCamelCase      )
		__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	SwinvaForImageClassification(_lowerCamelCase      )
		model.eval()

		__SCREAMING_SNAKE_CASE :	Optional[int]					 =	convert_state_dict(timm_model.state_dict() ,   _lowerCamelCase      )
		model.load_state_dict(_lowerCamelCase      )

		__SCREAMING_SNAKE_CASE :	int					 =	"""http://images.cocodataset.org/val2017/000000039769.jpg"""

		__SCREAMING_SNAKE_CASE :	List[Any]					 =	AutoImageProcessor.from_pretrained("""microsoft/{}""".format(swinva_name.replace("""_""" ,   """-"""      )      )      )
		__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	Image.open(requests.get(_lowerCamelCase ,   stream=_lowerCamelCase      ).raw      )
		__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	image_processor(images=_lowerCamelCase ,   return_tensors="""pt"""      )

		__SCREAMING_SNAKE_CASE :	int					 =	timm_model(inputs["""pixel_values"""]      )
		__SCREAMING_SNAKE_CASE :	Dict					 =	model(**_lowerCamelCase      ).logits

		assert torch.allclose(_lowerCamelCase ,   _lowerCamelCase ,   atol=1E-3      )

		print(F"Saving model {swinva_name} to {pytorch_dump_folder_path}"      )
		model.save_pretrained(_lowerCamelCase      )

		print(F"Saving image processor to {pytorch_dump_folder_path}"      )
		image_processor.save_pretrained(_lowerCamelCase      )

		model.push_to_hub(
		    repo_path_or_name=Path(_lowerCamelCase ,   _lowerCamelCase      ) ,   organization="""nandwalritik""" ,   commit_message="""Add model""" ,   )


if __name__ == "__main__":
						UpperCamelCase__ :			Union[str, Any]												=				argparse.ArgumentParser()
						# Required parameters
						parser.add_argument(
						    '''--swinv2_name''',
						    default='''swinv2_tiny_patch4_window8_256''',
						    type=str,
						    help='''Name of the Swinv2 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.'''
						)

						UpperCamelCase__ :			Optional[int]												=				parser.parse_args()
						convert_swinva_checkpoint(args.swinva_name, args.pytorch_dump_folder_path)