





"""simple docstring"""




import argparse
import logging
import os
import re

import tensorflow as tf

from transformers import (
    AutoConfig,
    AutoTokenizer,
    DataCollatorForLanguageModeling,
    PushToHubCallback,
    TFAutoModelForMaskedLM,
    create_optimizer,
)


a		:					Union[str, Any]	      =     logging.getLogger(__name__)

a		:					int	      =     tf.data.AUTOTUNE






def  _SCREAMING_SNAKE_CASE						(      )					->List[str]:






	'''simple docstring'''






	a :						Dict      =							argparse.ArgumentParser(description="Train a masked language model on TPU."   )
	parser.add_argument(
	    "--pretrained_model_config"       ,			type=_lowercase       ,			default="roberta-base"       ,			help="The model config to use. Note that we don't copy the model's weights, only the config!"       ,			)
	parser.add_argument(
	    "--tokenizer"       ,			type=_lowercase       ,			default="unigram-tokenizer-wikitext"       ,			help="The name of the tokenizer to load. We use the pretrained tokenizer to initialize the model's vocab size."       ,			)

	parser.add_argument(
	    "--per_replica_batch_size"       ,			type=_lowercase       ,			default=8       ,			help="Batch size per TPU core."       ,			)

	parser.add_argument(
	    "--no_tpu"       ,			action="store_true"       ,			help="If set, run on CPU and don't try to initialize a TPU. Useful for debugging on non-TPU instances."       ,			)

	parser.add_argument(
	    "--tpu_name"       ,			type=_lowercase       ,			help="Name of TPU resource to initialize. Should be blank on Colab, and 'local' on TPU VMs."       ,			default="local"       ,			)

	parser.add_argument(
	    "--tpu_zone"       ,			type=_lowercase       ,			help="Google cloud zone that TPU resource is located in. Only used for non-Colab TPU nodes."       ,			)

	parser.add_argument(
	    "--gcp_project"       ,			type=_lowercase       ,			help="Google cloud project name. Only used for non-Colab TPU nodes."   )

	parser.add_argument(
	    "--bfloat16"       ,			action="store_true"       ,			help="Use mixed-precision bfloat16 for training. This is the recommended lower-precision format for TPU."       ,			)

	parser.add_argument(
	    "--train_dataset"       ,			type=_lowercase       ,			help="Path to training dataset to load. If the path begins with `gs://`"
	    " then the dataset will be loaded from a Google Cloud Storage bucket."       ,			)

	parser.add_argument(
	    "--shuffle_buffer_size"       ,			type=_lowercase       ,			default=2**18       ,			help="Size of the shuffle buffer (in samples)"       ,			)

	parser.add_argument(
	    "--eval_dataset"       ,			type=_lowercase       ,			help="Path to evaluation dataset to load. If the path begins with `gs://`"
	    " then the dataset will be loaded from a Google Cloud Storage bucket."       ,			)

	parser.add_argument(
	    "--num_epochs"       ,			type=_lowercase       ,			default=1       ,			help="Number of epochs to train for."       ,			)

	parser.add_argument(
	    "--learning_rate"       ,			type=_lowercase       ,			default=1E-4       ,			help="Learning rate to use for training."       ,			)

	parser.add_argument(
	    "--weight_decay_rate"       ,			type=_lowercase       ,			default=1E-3       ,			help="Weight decay rate to use for training."       ,			)

	parser.add_argument(
	    "--max_length"       ,			type=_lowercase       ,			default=512       ,			help="Maximum length of tokenized sequences. Should match the setting used in prepare_tfrecord_shards.py"       ,			)

	parser.add_argument(
	    "--mlm_probability"       ,			type=_lowercase       ,			default=0.15       ,			help="Fraction of tokens to mask during training."       ,			)

	parser.add_argument("--output_dir"       ,			type=_lowercase       ,			required=_lowercase       ,			help="Path to save model checkpoints to."   )
	parser.add_argument("--hub_model_id"       ,			type=_lowercase       ,			help="Model ID to upload to on the Hugging Face Hub."   )

	a :						str      =							parser.parse_args()
	return args






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






	'''simple docstring'''






	try:
		if args.tpu_name:
			a :						Any      =							tf.distribute.cluster_resolver.TPUClusterResolver(
			    args.tpu_name       ,			zone=args.tpu_zone       ,			project=args.gcp_project   )
		else:
			a :						str      =							tf.distribute.cluster_resolver.TPUClusterResolver()
	except ValueError:
		raise RuntimeError(
		    "Couldn't connect to TPU! Most likely you need to specify --tpu_name, --tpu_zone, or "
		    "--gcp_project. When running on a TPU VM, use --tpu_name local."   )

	tf.config.experimental_connect_to_cluster(_lowercase   )
	tf.tpu.experimental.initialize_tpu_system(_lowercase   )

	return tpu






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       Optional[int]   )					->List[Any]:






	'''simple docstring'''






	a :						Any      =							0
	for file in file_list:
		a :						List[str]      =							file.split("/"   )[-1]
		a :						List[Any]      =							re.search(R"-\d+-(\d+)\.tfrecord"       ,			_lowercase   ).group(1   )
		a :						Dict      =							int(_lowercase   )
		num_samples += sample_count

	return num_samples






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       str       ,			_lowercase		:       Any       ,			_lowercase		:       Dict       ,			_lowercase		:       Tuple       ,			_lowercase		:       Union[str, Any]       ,			_lowercase		:       Dict=None   )					->Dict:






	'''simple docstring'''






	a :						List[str]      =							count_samples(_lowercase   )
	a :						Union[str, Any]      =							tf.data.Dataset.from_tensor_slices(_lowercase   )
	if shuffle:
		a :						Optional[int]      =							dataset.shuffle(len(_lowercase   )   )
	a :						List[Any]      =							tf.data.TFRecordDataset(_lowercase       ,			num_parallel_reads=_lowercase   )
	# TF can't infer the total sample count because it doesn't read all the records yet, so we assert it here
	a :						Dict      =							dataset.apply(tf.data.experimental.assert_cardinality(_lowercase   )   )
	a :						Optional[Any]      =							dataset.map(_lowercase       ,			num_parallel_calls=_lowercase   )
	if shuffle:
		assert shuffle_buffer_size is not None
		a :						Union[str, Any]      =							dataset.shuffle(args.shuffle_buffer_size   )
	a :						Union[str, Any]      =							dataset.batch(_lowercase       ,			drop_remainder=_lowercase   )
	a :						Dict      =							dataset.map(_lowercase       ,			num_parallel_calls=_lowercase   )
	a :						List[Any]      =							dataset.prefetch(_lowercase   )
	return dataset






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       List[Any]   )					->Optional[Any]:






	'''simple docstring'''






	if not args.no_tpu:
		a :						str      =							initialize_tpu(_lowercase   )
		a :						Optional[Any]      =							tf.distribute.TPUStrategy(_lowercase   )
	else:
		a :						Optional[Any]      =							tf.distribute.OneDeviceStrategy(device="/gpu:0"   )

	if args.bfloataa:
		tf.keras.mixed_precision.set_global_policy("mixed_bfloat16"   )

	a :						Tuple      =							AutoTokenizer.from_pretrained(args.tokenizer   )
	a :						Tuple      =							AutoConfig.from_pretrained(args.pretrained_model_config   )
	a :						int      =							tokenizer.vocab_size

	a :						Optional[int]      =							tf.io.gfile.glob(os.path.join(args.train_dataset       ,			"*.tfrecord"   )   )
	if not training_records:
		raise ValueError(F"""No .tfrecord files found in {args.train_dataset}."""   )
	a :						Tuple      =							tf.io.gfile.glob(os.path.join(args.eval_dataset       ,			"*.tfrecord"   )   )
	if not eval_records:
		raise ValueError(F"""No .tfrecord files found in {args.eval_dataset}."""   )

	a :						Any      =							count_samples(_lowercase   )

	a :						List[str]      =							num_train_samples // (args.per_replica_batch_size * strategy.num_replicas_in_sync)
	a :						List[Any]      =							steps_per_epoch * args.num_epochs

	with strategy.scope():
		a :						int      =							TFAutoModelForMaskedLM.from_config(_lowercase   )
		model(model.dummy_inputs   )  # Pass some dummy inputs through the model to ensure all the weights are built
		a,			a :						Optional[Any]      =							create_optimizer(
		    num_train_steps=_lowercase       ,			num_warmup_steps=total_train_steps // 20       ,			init_lr=args.learning_rate       ,			weight_decay_rate=args.weight_decay_rate       ,			)

		# Transformers models compute the right loss for their task by default when labels are passed, and will
		# use this for training unless you specify your own loss function in compile().
		model.compile(optimizer=_lowercase       ,			metrics=["accuracy"]   )

	def decode_fn(_lowercase		:       str   ):
		a :						Optional[Any]      =							{
		    "input_ids": tf.io.FixedLenFeature(dtype=tf.intaa       ,			shape=(args.max_length,)   ),
		    "attention_mask": tf.io.FixedLenFeature(dtype=tf.intaa       ,			shape=(args.max_length,)   ),
		}
		return tf.io.parse_single_example(_lowercase       ,			_lowercase   )

	# Many of the data collators in Transformers are TF-compilable when return_tensors == "tf", so we can
	# use their methods in our data pipeline.
	a :						Optional[Any]      =							DataCollatorForLanguageModeling(
	    tokenizer=_lowercase       ,			mlm_probability=args.mlm_probability       ,			mlm=_lowercase       ,			return_tensors="tf"   )

	def mask_with_collator(_lowercase		:       int   ):
		# TF really needs an isin() function
		a :						Dict      =							(
		    ~tf.cast(batch["attention_mask"]       ,			tf.bool   )
		    | (batch["input_ids"] == tokenizer.cls_token_id)
		    | (batch["input_ids"] == tokenizer.sep_token_id)
		)
		a,			a :						Optional[Any]      =							data_collator.tf_mask_tokens(
		    batch["input_ids"]       ,			vocab_size=len(_lowercase   )       ,			mask_token_id=tokenizer.mask_token_id       ,			special_tokens_mask=_lowercase       ,			)
		return batch

	a :						Union[str, Any]      =							args.per_replica_batch_size * strategy.num_replicas_in_sync

	a :						str      =							prepare_dataset(
	    _lowercase       ,			decode_fn=_lowercase       ,			mask_fn=_lowercase       ,			batch_size=_lowercase       ,			shuffle=_lowercase       ,			shuffle_buffer_size=args.shuffle_buffer_size       ,			)

	a :						str      =							prepare_dataset(
	    _lowercase       ,			decode_fn=_lowercase       ,			mask_fn=_lowercase       ,			batch_size=_lowercase       ,			shuffle=_lowercase       ,			)

	a :						Tuple      =							[]
	if args.hub_model_id:
		callbacks.append(
		    PushToHubCallback(output_dir=args.output_dir       ,			hub_model_id=args.hub_model_id       ,			tokenizer=_lowercase   )   )

	model.fit(
	    _lowercase       ,			validation_data=_lowercase       ,			epochs=args.num_epochs       ,			callbacks=_lowercase       ,			)

	model.save_pretrained(args.output_dir   )


if __name__ == "__main__":
		a		:					List[str]	      =     parse_args()
		main(args)









"""simple docstring"""




import argparse
import json

import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image
from torchvision import transforms

from transformers import BitImageProcessor, FocalNetConfig, FocalNetForImageClassification
from transformers.image_utils import IMAGENET_DEFAULT_MEAN, IMAGENET_DEFAULT_STD, PILImageResampling






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       List[Any]   )					->Dict:






	'''simple docstring'''






	a :						Any      =							[2, 2, 6, 2] if "tiny" in model_name else [2, 2, 18, 2]
	a :						str      =							True if "large" in model_name or "huge" in model_name else False
	a :						Optional[Any]      =							True if "large" in model_name or "huge" in model_name else False
	a :						Dict      =							True if "large" in model_name or "huge" in model_name else False

	if "large" in model_name or "xlarge" in model_name or "huge" in model_name:
		if "fl3" in model_name:
			a :						Union[str, Any]      =							[3, 3, 3, 3]
			a :						List[str]      =							[5, 5, 5, 5]
		elif "fl4" in model_name:
			a :						Any      =							[4, 4, 4, 4]
			a :						Optional[Any]      =							[3, 3, 3, 3]

	if "tiny" in model_name or "small" in model_name or "base" in model_name:
		a :						Dict      =							[3, 3, 3, 3]
		if "lrf" in model_name:
			a :						Optional[int]      =							[3, 3, 3, 3]
		else:
			a :						Tuple      =							[2, 2, 2, 2]

	if "tiny" in model_name:
		a :						List[str]      =							96
	elif "small" in model_name:
		a :						Union[str, Any]      =							96
	elif "base" in model_name:
		a :						Dict      =							128
	elif "large" in model_name:
		a :						Union[str, Any]      =							192
	elif "xlarge" in model_name:
		a :						Tuple      =							256
	elif "huge" in model_name:
		a :						List[str]      =							352

	# set label information
	a :						List[Any]      =							"huggingface/label-files"
	if "large" in model_name or "huge" in model_name:
		a :						Optional[int]      =							"imagenet-22k-id2label.json"
	else:
		a :						List[str]      =							"imagenet-1k-id2label.json"

	a :						Optional[int]      =							json.load(open(hf_hub_download(_lowercase       ,			_lowercase       ,			repo_type="dataset"   )       ,			"r"   )   )
	a :						str      =							{int(_lowercase   ): v for k, v in idalabel.items()}
	a :						List[str]      =							{v: k for k, v in idalabel.items()}

	a :						Dict      =							FocalNetConfig(
	    embed_dim=_lowercase       ,			depths=_lowercase       ,			focal_levels=_lowercase       ,			focal_windows=_lowercase       ,			use_conv_embed=_lowercase       ,			idalabel=_lowercase       ,			labelaid=_lowercase       ,			use_post_layernorm=_lowercase       ,			use_layerscale=_lowercase       ,			)

	return config






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       List[str]   )					->List[Any]:






	'''simple docstring'''






	if "patch_embed.proj" in name:
		a :						Any      =							name.replace("patch_embed.proj"       ,			"embeddings.patch_embeddings.projection"   )
	if "patch_embed.norm" in name:
		a :						List[str]      =							name.replace("patch_embed.norm"       ,			"embeddings.norm"   )
	if "layers" in name:
		a :						List[Any]      =							"encoder." + name
	if "encoder.layers" in name:
		a :						int      =							name.replace("encoder.layers"       ,			"encoder.stages"   )
	if "downsample.proj" in name:
		a :						Any      =							name.replace("downsample.proj"       ,			"downsample.projection"   )
	if "blocks" in name:
		a :						str      =							name.replace("blocks"       ,			"layers"   )
	if "modulation.f.weight" in name or "modulation.f.bias" in name:
		a :						Union[str, Any]      =							name.replace("modulation.f"       ,			"modulation.projection_in"   )
	if "modulation.h.weight" in name or "modulation.h.bias" in name:
		a :						Dict      =							name.replace("modulation.h"       ,			"modulation.projection_context"   )
	if "modulation.proj.weight" in name or "modulation.proj.bias" in name:
		a :						Any      =							name.replace("modulation.proj"       ,			"modulation.projection_out"   )

	if name == "norm.weight":
		a :						str      =							"layernorm.weight"
	if name == "norm.bias":
		a :						Optional[Any]      =							"layernorm.bias"

	if "head" in name:
		a :						Tuple      =							name.replace("head"       ,			"classifier"   )
	else:
		a :						int      =							"focalnet." + name

	return name






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       Tuple       ,			_lowercase		:       Optional[Any]       ,			_lowercase		:       Tuple=False   )					->str:






	'''simple docstring'''






	a :						List[Any]      =							{
	    "focalnet-tiny": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_tiny_srf.pth",
	    "focalnet-tiny-lrf": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_tiny_lrf.pth",
	    "focalnet-small": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_small_srf.pth",
	    "focalnet-small-lrf": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_small_lrf.pth",
	    "focalnet-base": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_base_srf.pth",
	    "focalnet-base-lrf": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_base_lrf.pth",
	    "focalnet-large-lrf-fl3": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_large_lrf_384.pth",
	    "focalnet-large-lrf-fl4": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_large_lrf_384_fl4.pth",
	    "focalnet-xlarge-lrf-fl3": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_xlarge_lrf_384.pth",
	    "focalnet-xlarge-lrf-fl4": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_xlarge_lrf_384_fl4.pth",
	}
	# fmt: on

	a :						str      =							model_name_to_url[model_name]
	print("Checkpoint URL: "       ,			_lowercase   )
	a :						Any      =							torch.hub.load_state_dict_from_url(_lowercase       ,			map_location="cpu"   )["model"]

	# rename keys
	for key in state_dict.copy().keys():
		a :						Any      =							state_dict.pop(_lowercase   )
		a :						Any      =							val

	a :						Any      =							get_focalnet_config(_lowercase   )
	a :						Optional[int]      =							FocalNetForImageClassification(_lowercase   )
	model.eval()

	# load state dict
	model.load_state_dict(_lowercase   )

	# verify conversion
	a :						Union[str, Any]      =							"http://images.cocodataset.org/val2017/000000039769.jpg"

	a :						Optional[int]      =							BitImageProcessor(
	    do_resize=_lowercase       ,			size={"shortest_edge": 256}       ,			resample=PILImageResampling.BILINEAR       ,			do_center_crop=_lowercase       ,			crop_size=224       ,			do_normalize=_lowercase       ,			image_mean=_lowercase       ,			image_std=_lowercase       ,			)
	a :						int      =							Image.open(requests.get(_lowercase       ,			stream=_lowercase   ).raw   )
	a :						Dict      =							processor(images=_lowercase       ,			return_tensors="pt"   )

	a :						Optional[int]      =							transforms.Compose(
	    [
	        transforms.Resize(256   ),
	        transforms.CenterCrop(224   ),
	        transforms.ToTensor(),
	        transforms.Normalize(mean=[0.485, 0.456, 0.406]       ,			std=[0.229, 0.224, 0.225]   ),
	    ]   )

	a :						str      =							image_transforms(_lowercase   ).unsqueeze(0   )

	# verify pixel_values
	assert torch.allclose(inputs.pixel_values       ,			_lowercase       ,			atol=1E-4   )

	a :						Dict      =							model(**_lowercase   )

	a :						List[str]      =							outputs.logits.argmax(-1   ).item()
	print("Predicted class:"       ,			model.config.idalabel[predicted_class_idx]   )

	print("First values of logits:"       ,			outputs.logits[0, :3]   )

	if model_name == "focalnet-tiny":
		a :						Union[str, Any]      =							torch.tensor([0.2166, -0.4368, 0.2191]   )
	elif model_name == "focalnet-tiny-lrf":
		a :						Union[str, Any]      =							torch.tensor([1.1669, 0.0125, -0.1695]   )
	elif model_name == "focalnet-small":
		a :						Dict      =							torch.tensor([0.4917, -0.0430, 0.1341]   )
	elif model_name == "focalnet-small-lrf":
		a :						Dict      =							torch.tensor([-0.2588, -0.5342, -0.2331]   )
	elif model_name == "focalnet-base":
		a :						Any      =							torch.tensor([-0.1655, -0.4090, -0.1730]   )
	elif model_name == "focalnet-base-lrf":
		a :						str      =							torch.tensor([0.5306, -0.0483, -0.3928]   )
	assert torch.allclose(outputs.logits[0, :3]       ,			_lowercase       ,			atol=1E-4   )
	print("Looks ok!"   )

	if pytorch_dump_folder_path is not None:
		print(F"""Saving model and processor of {model_name} to {pytorch_dump_folder_path}"""   )
		model.save_pretrained(_lowercase   )
		processor.save_pretrained(_lowercase   )

	if push_to_hub:
		print(F"""Pushing model and processor of {model_name} to the hub..."""   )
		model.push_to_hub(F"""{model_name}"""   )
		processor.push_to_hub(F"""{model_name}"""   )


if __name__ == "__main__":
		a		:					str	      =     argparse.ArgumentParser()
		# Required parameters
		parser.add_argument(
		    '''--model_name''',
		    default='''focalnet-tiny''',
		    type=str,
		    help='''Name of the FocalNet 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 and processor to the hub.''',
		)

		a		:					Tuple	      =     parser.parse_args()
		convert_focalnet_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)


