


'''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_		(    __A				:    Union[str, Any]       )			->	Dict:

       """simple docstring"""

       a_      :	Tuple							=						[2, 2, 6, 2] if 'tiny' in model_name else [2, 2, 18, 2]
       a_      :	List[str]							=						True if 'large' in model_name or 'huge' in model_name else False
       a_      :	str							=						True if 'large' in model_name or 'huge' in model_name else False
       a_      :	Tuple							=						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_      :	List[str]							=						[3, 3, 3, 3]
                     a_      :	Tuple							=						[5, 5, 5, 5]
              elif "fl4" in model_name:
                     a_      :	List[str]							=						[4, 4, 4, 4]
                     a_      :	List[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_      :	Union[str, Any]							=						[3, 3, 3, 3]
              else:
                     a_      :	Any							=						[2, 2, 2, 2]

       if "tiny" in model_name:
              a_      :	Tuple							=						96
       elif "small" in model_name:
              a_      :	Optional[int]							=						96
       elif "base" in model_name:
              a_      :	str							=						1_28
       elif "large" in model_name:
              a_      :	str							=						1_92
       elif "xlarge" in model_name:
              a_      :	int							=						2_56
       elif "huge" in model_name:
              a_      :	Optional[int]							=						3_52

       # set label information
       a_      :	str							=						'huggingface/label-files'
       if "large" in model_name or "huge" in model_name:
              a_      :	Tuple							=						'imagenet-22k-id2label.json'
       else:
              a_      :	int							=						'imagenet-1k-id2label.json'

       a_      :	Optional[int]							=						json.load(open(hf_hub_download(__A						,					__A						,					repo_type='dataset'       )						,					'r'       )       )
       a_      :	Optional[int]							=						{int(__A       ): v for k, v in idalabel.items()}
       a_      :	Union[str, Any]							=						{v: k for k, v in idalabel.items()}

       a_      :	Optional[Any]							=						FocalNetConfig(
           embed_dim=__A						,					depths=__A						,					focal_levels=__A						,					focal_windows=__A						,					use_conv_embed=__A						,					idalabel=__A						,					labelaid=__A						,					use_post_layernorm=__A						,					use_layerscale=__A						,					)

       return config







def  SCREAMING_SNAKE_CASE_		(    __A				:    List[Any]       )			->	Optional[int]:

       """simple docstring"""

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

       if name == "norm.weight":
              a_      :	Dict							=						'layernorm.weight'
       if name == "norm.bias":
              a_      :	Optional[Any]							=						'layernorm.bias'

       if "head" in name:
              a_      :	Dict							=						name.replace('head'						,					'classifier'       )
       else:
              a_      :	int							=						'focalnet.' + name

       return name







def  SCREAMING_SNAKE_CASE_		(    __A				:    str						,					__A				:    Union[str, Any]						,					__A				:    int=False       )			->	List[Any]:

       """simple docstring"""

       a_      :	Tuple							=						{
           '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: '						,					__A       )
       a_      :	List[Any]							=						torch.hub.load_state_dict_from_url(__A						,					map_location='cpu'       )['model']

       # rename keys
       for key in state_dict.copy().keys():
              a_      :	List[str]							=						state_dict.pop(__A       )
              a_      :	Optional[Any]							=						val

       a_      :	Optional[int]							=						get_focalnet_config(__A       )
       a_      :	Any							=						FocalNetForImageClassification(__A       )
       model.eval()

       # load state dict
       model.load_state_dict(__A       )

       # verify conversion
       a_      :	List[Any]							=						'http://images.cocodataset.org/val2017/000000039769.jpg'

       a_      :	Optional[int]							=						BitImageProcessor(
           do_resize=__A						,					size={'shortest_edge': 2_56}						,					resample=PILImageResampling.BILINEAR						,					do_center_crop=__A						,					crop_size=2_24						,					do_normalize=__A						,					image_mean=__A						,					image_std=__A						,					)
       a_      :	Tuple							=						Image.open(requests.get(__A						,					stream=__A       ).raw       )
       a_      :	Optional[int]							=						processor(images=__A						,					return_tensors='pt'       )

       a_      :	int							=						transforms.Compose(
           [
               transforms.Resize(2_56       ),
               transforms.CenterCrop(2_24       ),
               transforms.ToTensor(),
               transforms.Normalize(mean=[0.485, 0.456, 0.406]						,					std=[0.229, 0.224, 0.225]       ),
           ]       )

       a_      :	str							=						image_transforms(__A       ).unsqueeze(0       )

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

       a_      :	List[Any]							=						model(**__A       )

       a_      :	List[Any]							=						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_      :	Optional[int]							=						torch.tensor([0.2166, -0.4368, 0.2191]       )
       elif model_name == "focalnet-tiny-lrf":
              a_      :	int							=						torch.tensor([1.1669, 0.0125, -0.1695]       )
       elif model_name == "focalnet-small":
              a_      :	Any							=						torch.tensor([0.4917, -0.0430, 0.1341]       )
       elif model_name == "focalnet-small-lrf":
              a_      :	Any							=						torch.tensor([-0.2588, -0.5342, -0.2331]       )
       elif model_name == "focalnet-base":
              a_      :	Union[str, Any]							=						torch.tensor([-0.1655, -0.4090, -0.1730]       )
       elif model_name == "focalnet-base-lrf":
              a_      :	int							=						torch.tensor([0.5306, -0.0483, -0.3928]       )
       assert torch.allclose(outputs.logits[0, :3]						,					__A						,					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(__A       )
              processor.save_pretrained(__A       )

       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__":
   UpperCAmelCase_   :				Tuple						      = 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.',
   )

   UpperCAmelCase_   :				Optional[int]						      = parser.parse_args()
   convert_focalnet_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)






import pyarrow.parquet as pq
import pytest

from datasets import Audio, Dataset, DatasetDict, Features, NamedSplit, Sequence, Value, config
from datasets.features.image import Image
from datasets.io.parquet import ParquetDatasetReader, ParquetDatasetWriter, get_writer_batch_size

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases







def  SCREAMING_SNAKE_CASE_		(    __A				:    Union[str, Any]						,					__A				:    Tuple       )			->	Any:

       """simple docstring"""

       assert isinstance(__A						,					__A       )
       assert dataset.num_rows == 4
       assert dataset.num_columns == 3
       assert dataset.column_names == ["col_1", "col_2", "col_3"]
       for feature, expected_dtype in expected_features.items():
              assert dataset.features[feature].dtype == expected_dtype







@pytest.mark.parametrize('keep_in_memory'						,					[False, True]       )
def  SCREAMING_SNAKE_CASE_		(    __A				:    List[Any]						,					__A				:    Any						,					__A				:    Tuple       )			->	Tuple:

       """simple docstring"""

       a_      :	Dict							=						tmp_path / 'cache'
       a_      :	int							=						{'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'}
       with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
              a_      :	Union[str, Any]							=						ParquetDatasetReader(__A						,					cache_dir=__A						,					keep_in_memory=__A       ).read()
       _check_parquet_dataset(__A						,					__A       )







@pytest.mark.parametrize(
    'features'						,					[
        None,
        {'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'},
        {'col_1': 'string', 'col_2': 'string', 'col_3': 'string'},
        {'col_1': 'int32', 'col_2': 'int32', 'col_3': 'int32'},
        {'col_1': 'float32', 'col_2': 'float32', 'col_3': 'float32'},
    ]						,					)
def  SCREAMING_SNAKE_CASE_		(    __A				:    str						,					__A				:    Dict						,					__A				:    Union[str, Any]       )			->	Dict:

       """simple docstring"""

       a_      :	Tuple							=						tmp_path / 'cache'
       a_      :	str							=						{'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'}
       a_      :	List[str]							=						features.copy() if features else default_expected_features
       a_      :	int							=						(
           Features({feature: Value(__A       ) for feature, dtype in features.items()}       ) if features is not None else None
       )
       a_      :	List[str]							=						ParquetDatasetReader(__A						,					features=__A						,					cache_dir=__A       ).read()
       _check_parquet_dataset(__A						,					__A       )







@pytest.mark.parametrize('split'						,					[None, NamedSplit('train'       ), 'train', 'test']       )
def  SCREAMING_SNAKE_CASE_		(    __A				:    List[str]						,					__A				:    List[str]						,					__A				:    int       )			->	List[str]:

       """simple docstring"""

       a_      :	int							=						tmp_path / 'cache'
       a_      :	str							=						{'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'}
       a_      :	List[str]							=						ParquetDatasetReader(__A						,					cache_dir=__A						,					split=__A       ).read()
       _check_parquet_dataset(__A						,					__A       )
       assert dataset.split == split if split else "train"







@pytest.mark.parametrize('path_type'						,					[str, list]       )
def  SCREAMING_SNAKE_CASE_		(    __A				:    Tuple						,					__A				:    str						,					__A				:    Optional[int]       )			->	Any:

       """simple docstring"""

       if issubclass(__A						,					__A       ):
              a_      :	Tuple							=						parquet_path
       elif issubclass(__A						,					__A       ):
              a_      :	str							=						[parquet_path]
       a_      :	int							=						tmp_path / 'cache'
       a_      :	int							=						{'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'}
       a_      :	str							=						ParquetDatasetReader(__A						,					cache_dir=__A       ).read()
       _check_parquet_dataset(__A						,					__A       )







def  SCREAMING_SNAKE_CASE_		(    __A				:    Any						,					__A				:    Dict						,					__A				:    Optional[Any]=("train",)       )			->	Optional[int]:

       """simple docstring"""

       assert isinstance(__A						,					__A       )
       for split in splits:
              a_      :	str							=						dataset_dict[split]
              assert dataset.num_rows == 4
              assert dataset.num_columns == 3
              assert dataset.column_names == ["col_1", "col_2", "col_3"]
              for feature, expected_dtype in expected_features.items():
                     assert dataset.features[feature].dtype == expected_dtype







@pytest.mark.parametrize('keep_in_memory'						,					[False, True]       )
def  SCREAMING_SNAKE_CASE_		(    __A				:    Optional[Any]						,					__A				:    str						,					__A				:    str       )			->	Union[str, Any]:

       """simple docstring"""

       a_      :	Union[str, Any]							=						tmp_path / 'cache'
       a_      :	Any							=						{'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'}
       with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
              a_      :	Tuple							=						ParquetDatasetReader(
                  {'train': parquet_path}						,					cache_dir=__A						,					keep_in_memory=__A       ).read()
       _check_parquet_datasetdict(__A						,					__A       )







@pytest.mark.parametrize(
    'features'						,					[
        None,
        {'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'},
        {'col_1': 'string', 'col_2': 'string', 'col_3': 'string'},
        {'col_1': 'int32', 'col_2': 'int32', 'col_3': 'int32'},
        {'col_1': 'float32', 'col_2': 'float32', 'col_3': 'float32'},
    ]						,					)
def  SCREAMING_SNAKE_CASE_		(    __A				:    Any						,					__A				:    Optional[int]						,					__A				:    Tuple       )			->	List[Any]:

       """simple docstring"""

       a_      :	Optional[Any]							=						tmp_path / 'cache'
       a_      :	List[str]							=						{'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'}
       a_      :	Optional[int]							=						features.copy() if features else default_expected_features
       a_      :	Tuple							=						(
           Features({feature: Value(__A       ) for feature, dtype in features.items()}       ) if features is not None else None
       )
       a_      :	Optional[Any]							=						ParquetDatasetReader({'train': parquet_path}						,					features=__A						,					cache_dir=__A       ).read()
       _check_parquet_datasetdict(__A						,					__A       )







@pytest.mark.parametrize('split'						,					[None, NamedSplit('train'       ), 'train', 'test']       )
def  SCREAMING_SNAKE_CASE_		(    __A				:    int						,					__A				:    List[Any]						,					__A				:    Optional[Any]       )			->	Any:

       """simple docstring"""

       if split:
              a_      :	Any							=						{split: parquet_path}
       else:
              a_      :	Dict							=						'train'
              a_      :	int							=						{'train': parquet_path, 'test': parquet_path}
       a_      :	int							=						tmp_path / 'cache'
       a_      :	List[Any]							=						{'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'}
       a_      :	Tuple							=						ParquetDatasetReader(__A						,					cache_dir=__A       ).read()
       _check_parquet_datasetdict(__A						,					__A						,					splits=list(path.keys()       )       )
       assert all(dataset[split].split == split for split in path.keys()       )







def  SCREAMING_SNAKE_CASE_		(    __A				:    Any						,					__A				:    Optional[int]       )			->	List[Any]:

       """simple docstring"""

       a_      :	List[str]							=						ParquetDatasetWriter(__A						,					tmp_path / 'foo.parquet'       )
       assert writer.write() > 0
       a_      :	List[str]							=						pq.ParquetFile(tmp_path / 'foo.parquet'       )
       a_      :	Dict							=						pf.read()
       assert dataset.data.table == output_table







def  SCREAMING_SNAKE_CASE_		(    __A				:    Optional[int]						,					__A				:    Any       )			->	Optional[int]:

       """simple docstring"""

       a_      :	str							=						str(shared_datadir / 'test_image_rgb.jpg'       )
       a_      :	List[Any]							=						{'image': [image_path]}
       a_      :	int							=						Features({'image': Image()}       )
       a_      :	List[Any]							=						Dataset.from_dict(__A						,					features=__A       )
       a_      :	str							=						ParquetDatasetWriter(__A						,					tmp_path / 'foo.parquet'       )
       assert writer.write() > 0

       a_      :	str							=						Dataset.from_parquet(str(tmp_path / 'foo.parquet'       )       )
       assert dataset.features == reloaded_dataset.features

       a_      :	str							=						ParquetDatasetReader(str(tmp_path / 'foo.parquet'       )						,					streaming=__A       ).read()
       assert dataset.features == reloaded_iterable_dataset.features







@pytest.mark.parametrize(
    'feature, expected'						,					[
        (Features({'foo': Value('int32'       )}       ), None),
        (Features({'image': Image(), 'foo': Value('int32'       )}       ), config.PARQUET_ROW_GROUP_SIZE_FOR_IMAGE_DATASETS),
        (Features({'nested': Sequence(Audio()       )}       ), config.PARQUET_ROW_GROUP_SIZE_FOR_AUDIO_DATASETS),
    ]						,					)
def  SCREAMING_SNAKE_CASE_		(    __A				:    int						,					__A				:    List[str]       )			->	List[str]:

       """simple docstring"""

       assert get_writer_batch_size(__A       ) == expected


