




"""simple docstring"""




# Lint as: python3
import os
import re
import urllib.parse
from pathlib import Path
from typing import Callable, List, Optional, Union
from zipfile import ZipFile

from ..utils.file_utils import cached_path, hf_github_url
from ..utils.logging import get_logger
from ..utils.version import Version


a_        =       get_logger(__name__)

class   UpperCAmelCase_    :
		UpperCamelCase							='''dummy_data'''
		UpperCamelCase							='''datasets'''
		UpperCamelCase							=False
		def __init__(						self				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_ = None				,						UpperCamelCase_ = False				,						UpperCamelCase_ = True				,						UpperCamelCase_ = None				,						)						->       List[str]:
					__lowercase			:  Optional[int]								=       0
					__lowercase			:  Tuple								=       dataset_name
					__lowercase			:  List[str]								=       cache_dir
					__lowercase			:  int								=       use_local_dummy_data
					__lowercase			:  Dict								=       config
					# download_callbacks take a single url as input
					__lowercase			:  List[Callable]								=       download_callbacks or []
					# if False, it doesn't load existing files and it returns the paths of the dummy files relative
					# to the dummy_data zip file root
					__lowercase			:  Union[str, Any]								=       load_existing_dummy_data

					# TODO(PVP, QL) might need to make this more general
					__lowercase			:  List[Any]								=       str(lowercase__       )
					# to be downloaded
					__lowercase			:  str								=       None
					__lowercase			:  Optional[int]								=       None
		@property
		def 	_lowerCamelCase							(						self       )						->       Tuple:
					if self._dummy_file is None:
								__lowercase			:  int								=       self.download_dummy_data()
					return self._dummy_file
		@property
		def 	_lowerCamelCase							(						self       )						->       Dict:
					if self.config is not None:
								# structure is dummy / config_name / version_name
								return os.path.join('''dummy'''				,						self.config.name				,						self.version_name       )
					# structure is dummy / version_name
					return os.path.join('''dummy'''				,						self.version_name       )
		@property
		def 	_lowerCamelCase							(						self       )						->       Optional[Any]:
					return os.path.join(self.dummy_data_folder				,						'''dummy_data.zip'''       )
		def 	_lowerCamelCase							(						self       )						->       str:
					__lowercase			:  Optional[int]								=       (
					    self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
					)

					__lowercase			:  int								=       cached_path(
					    lowercase__				,						cache_dir=self.cache_dir				,						extract_compressed_file=lowercase__				,						force_extract=lowercase__       )

					return os.path.join(lowercase__				,						self.dummy_file_name       )
		@property
		def 	_lowerCamelCase							(						self       )						->       Union[str, Any]:
					return os.path.join(self.datasets_scripts_dir				,						self.dataset_name				,						self.dummy_zip_file       )
		@property
		def 	_lowerCamelCase							(						self       )						->       Dict:
					if self._bucket_url is None:
								__lowercase			:  str								=       hf_github_url(self.dataset_name				,						self.dummy_zip_file.replace(os.sep				,						'''/'''       )       )
					return self._bucket_url
		@property
		def 	_lowerCamelCase							(						self       )						->       Any:
					# return full path if its a dir
					if os.path.isdir(self.dummy_file       ):
								return self.dummy_file
					# else cut off path to file -> example `xsum`.
					return "/".join(self.dummy_file.replace(os.sep				,						'''/'''       ).split('''/'''       )[:-1]       )
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						*UpperCamelCase_       )						->       Dict:
					if self.load_existing_dummy_data:
								# dummy data is downloaded and tested
								__lowercase			:  str								=       self.dummy_file
					else:
								# dummy data cannot be downloaded and only the path to dummy file is returned
								__lowercase			:  Optional[int]								=       self.dummy_file_name

					# special case when data_url is a dict
					if isinstance(lowercase__				,						lowercase__       ):
								return self.create_dummy_data_dict(lowercase__				,						lowercase__       )
					elif isinstance(lowercase__				,						(list, tuple)       ):
								return self.create_dummy_data_list(lowercase__				,						lowercase__       )
					else:
								return self.create_dummy_data_single(lowercase__				,						lowercase__       )
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						*UpperCamelCase_       )						->       Optional[int]:
					return self.download_and_extract(lowercase__       )
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						UpperCamelCase_       )						->       List[str]:
					return self.download_and_extract(lowercase__       )
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						*UpperCamelCase_				,						**UpperCamelCase_       )						->       int:
					return path
		def 	_lowerCamelCase							(						self       )						->       str:
					return {}
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						UpperCamelCase_       )						->       List[str]:
					__lowercase			:  Dict								=       {}
					for key, single_urls in data_url.items():
								for download_callback in self.download_callbacks:
											if isinstance(lowercase__				,						lowercase__       ):
														for single_url in single_urls:
																	download_callback(lowercase__       )
											else:
														__lowercase			:  List[Any]								=       single_urls
														download_callback(lowercase__       )
            # we force the name of each key to be the last file / folder name of the url path
            # if the url has arguments, we need to encode them with urllib.parse.quote_plus
								if isinstance(lowercase__				,						lowercase__       ):
											__lowercase			:  Optional[int]								=       [os.path.join(lowercase__				,						urllib.parse.quote_plus(Path(lowercase__       ).name       )       ) for x in single_urls]
								else:
											__lowercase			:  List[Any]								=       single_urls
											__lowercase			:  List[Any]								=       os.path.join(lowercase__				,						urllib.parse.quote_plus(Path(lowercase__       ).name       )       )
								__lowercase			:  List[Any]								=       value

					# make sure that values are unique
					if all(isinstance(lowercase__				,						lowercase__       ) for i in dummy_data_dict.values()       ) and len(set(dummy_data_dict.values()       )       ) < len(
					    dummy_data_dict.values()       ):
								# append key to value to make its name unique
								__lowercase			:  Tuple								=       {key: value + key for key, value in dummy_data_dict.items()}

					return dummy_data_dict
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						UpperCamelCase_       )						->       Optional[int]:
					__lowercase			:  Tuple								=       []
					# trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
					__lowercase			:  Optional[int]								=       all(bool(re.findall('''[0-9]{3,}-of-[0-9]{3,}'''				,						lowercase__       )       ) for url in data_url       )
					__lowercase			:  Optional[int]								=       all(
					    url.startswith('''https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/pubmed'''       ) for url in data_url       )
					if data_url and (is_tf_records or is_pubmed_records):
								__lowercase			:  Optional[int]								=       [data_url[0]] * len(lowercase__       )
					for single_url in data_url:
								for download_callback in self.download_callbacks:
											download_callback(lowercase__       )
								# we force the name of each key to be the last file / folder name of the url path
								# if the url has arguments, we need to encode them with urllib.parse.quote_plus
								__lowercase			:  Union[str, Any]								=       os.path.join(lowercase__				,						urllib.parse.quote_plus(single_url.split('''/'''       )[-1]       )       )
								dummy_data_list.append(lowercase__       )
					return dummy_data_list
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						UpperCamelCase_       )						->       List[Any]:
					for download_callback in self.download_callbacks:
								download_callback(lowercase__       )
					# we force the name of each key to be the last file / folder name of the url path
					# if the url has arguments, we need to encode them with urllib.parse.quote_plus
					__lowercase			:  Union[str, Any]								=       os.path.join(lowercase__				,						urllib.parse.quote_plus(data_url.split('''/'''       )[-1]       )       )
					if os.path.exists(lowercase__       ) or not self.load_existing_dummy_data:
								return value
					else:
								# Backward compatibility, maybe deprecate at one point.
								# For many datasets with single url calls to dl_manager.download_and_extract,
								# the dummy_data.zip file is actually the zipped downloaded file
								# while now we expected the dummy_data.zip file to be a directory containing
								# the downloaded file.
								return path_to_dummy_data
		def 	_lowerCamelCase							(						self       )						->       Optional[Any]:
					pass
		def 	_lowerCamelCase							(						self       )						->       Optional[Any]:
					pass
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_       )						->       int:
					def _iter_archive_members(UpperCamelCase_       ):
								# this preserves the order of the members inside the ZIP archive
								__lowercase			:  Union[str, Any]								=       Path(self.dummy_file       ).parent
								__lowercase			:  Dict								=       path.relative_to(lowercase__       )
								with ZipFile(self.local_path_to_dummy_data       ) as zip_file:
											__lowercase			:  int								=       zip_file.namelist()
								for member in members:
											if member.startswith(relative_path.as_posix()       ):
														yield dummy_parent_path.joinpath(lowercase__       )

					__lowercase			:  Optional[Any]								=       Path(lowercase__       )
					__lowercase			:  List[Any]								=       _iter_archive_members(lowercase__       ) if self.use_local_dummy_data else path.rglob('''*'''       )
					for file_path in file_paths:
								if file_path.is_file() and not file_path.name.startswith(('''.''', '''__''')       ):
											yield file_path.relative_to(lowercase__       ).as_posix(), file_path.open('''rb'''       )
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_       )						->       Any:
					if not isinstance(lowercase__				,						lowercase__       ):
								__lowercase			:  Any								=       [paths]
					for path in paths:
								if os.path.isfile(lowercase__       ):
											if os.path.basename(lowercase__       ).startswith(('''.''', '''__''')       ):
														return
											yield path
								else:
											for dirpath, dirnames, filenames in os.walk(lowercase__       ):
														if os.path.basename(lowercase__       ).startswith(('''.''', '''__''')       ):
																	continue
														dirnames.sort()
														for filename in sorted(lowercase__       ):
																	if filename.startswith(('''.''', '''__''')       ):
																				continue
																	yield os.path.join(lowercase__				,						lowercase__       )







import copy
import inspect
import unittest

from transformers import AutoBackbone
from transformers.configuration_utils import PretrainedConfig
from transformers.testing_utils import require_timm, require_torch, torch_device
from transformers.utils.import_utils import is_torch_available

from ...test_backbone_common import BackboneTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor


if is_torch_available():
				import torch

				from transformers import TimmBackbone, TimmBackboneConfig

from ...test_pipeline_mixin import PipelineTesterMixin




class 							lowerCamelCase   :


	def __init__(    self				,     lowercase__				,     lowercase__=None				,     lowercase__=None				,     lowercase__=None				,     lowercase__="resnet50"				,     lowercase__=3				,     lowercase__=3_2				,     lowercase__=3				,     lowercase__=True				,     lowercase__=True				,     ):
							__UpperCAmelCase			:							List[str]	    =	parent
							__UpperCAmelCase			:							Dict	    =	out_indices if out_indices is not None else [4]
							__UpperCAmelCase			:							List[Any]	    =	stage_names
							__UpperCAmelCase			:							int	    =	out_features
							__UpperCAmelCase			:							Union[str, Any]	    =	backbone
							__UpperCAmelCase			:							Optional[Any]	    =	batch_size
							__UpperCAmelCase			:							Optional[Any]	    =	image_size
							__UpperCAmelCase			:							Optional[int]	    =	num_channels
							__UpperCAmelCase			:							List[Any]	    =	use_pretrained_backbone
							__UpperCAmelCase			:							Dict	    =	is_training


	def  A(    self):
							__UpperCAmelCase			:							Optional[Any]	    =	floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size])
							__UpperCAmelCase			:							Tuple	    =	self.get_config()

							return config, pixel_values


	def  A(    self):
							return TimmBackboneConfig(
							    image_size=self.image_size				,     num_channels=self.num_channels				,     out_features=self.out_features				,     out_indices=self.out_indices				,     stage_names=self.stage_names				,     use_pretrained_backbone=self.use_pretrained_backbone				,     backbone=self.backbone				,     )


	def  A(    self				,     lowercase__				,     lowercase__):
							__UpperCAmelCase			:							Tuple	    =	TimmBackbone(config=lowercase__)
							model.to(lowercase__)
							model.eval()
							with torch.no_grad():
													__UpperCAmelCase			:							Union[str, Any]	    =	model(lowercase__)
							self.parent.assertEqual(
							    result.feature_map[-1].shape				,     (self.batch_size, model.channels[-1], 1_4, 1_4)				,     )


	def  A(    self):
							__UpperCAmelCase			:							Dict	    =	self.prepare_config_and_inputs()
							__UpperCAmelCase    ,      __UpperCAmelCase			:							List[Any]	    =	config_and_inputs
							__UpperCAmelCase			:							List[Any]	    =	{'''pixel_values''': pixel_values}
							return config, inputs_dict






@require_torch
@require_timm
class 							lowerCamelCase   (  _UpperCamelCase							,						_UpperCamelCase							,						_UpperCamelCase							,						unittest.TestCase ):
	_lowerCAmelCase     :		Dict										=						(TimmBackbone,) if is_torch_available() else ()
	_lowerCAmelCase     :		str										=						{'''feature-extraction''': TimmBackbone} if is_torch_available() else {}
	_lowerCAmelCase     :		List[str]										=						False
	_lowerCAmelCase     :		str										=						False
	_lowerCAmelCase     :		List[Any]										=						False
	_lowerCAmelCase     :		List[str]										=						False


	def  A(    self):
							__UpperCAmelCase			:							List[Any]	    =	TimmBackboneModelTester(self)
							__UpperCAmelCase			:							int	    =	ConfigTester(self				,     config_class=lowercase__				,     has_text_modality=lowercase__)


	def  A(    self):
							self.config_tester.create_and_test_config_to_json_string()
							self.config_tester.create_and_test_config_to_json_file()
							self.config_tester.create_and_test_config_from_and_save_pretrained()
							self.config_tester.create_and_test_config_with_num_labels()
							self.config_tester.check_config_can_be_init_without_params()
							self.config_tester.check_config_arguments_init()


	def  A(    self):
							__UpperCAmelCase			:							int	    =	'''resnet18'''
							__UpperCAmelCase			:							List[str]	    =	'''microsoft/resnet-18'''

							__UpperCAmelCase			:							Any	    =	AutoBackbone.from_pretrained(lowercase__				,     use_timm_backbone=lowercase__)
							__UpperCAmelCase			:							Union[str, Any]	    =	AutoBackbone.from_pretrained(lowercase__)

							self.assertEqual(len(timm_model.out_features)				,     len(transformers_model.out_features))
							self.assertEqual(len(timm_model.stage_names)				,     len(transformers_model.stage_names))
							self.assertEqual(timm_model.channels				,     transformers_model.channels)
							# Out indices are set to the last layer by default. For timm models, we don't know
							# the number of layers in advance, so we set it to (-1,), whereas for transformers
							# models, we set it to [len(stage_names) - 1] (kept for backward compatibility).
							self.assertEqual(timm_model.out_indices				,     (-1,))
							self.assertEqual(transformers_model.out_indices				,     [len(timm_model.stage_names) - 1])

							__UpperCAmelCase			:							Union[str, Any]	    =	AutoBackbone.from_pretrained(lowercase__				,     use_timm_backbone=lowercase__				,     out_indices=[1, 2, 3])
							__UpperCAmelCase			:							Any	    =	AutoBackbone.from_pretrained(lowercase__				,     out_indices=[1, 2, 3])

							self.assertEqual(timm_model.out_indices				,     transformers_model.out_indices)
							self.assertEqual(len(timm_model.out_features)				,     len(transformers_model.out_features))
							self.assertEqual(timm_model.channels				,     transformers_model.channels)


	@unittest.skip('''TimmBackbone doesn\'t support feed forward chunking''')
	def  A(    self):
							pass


	@unittest.skip('''TimmBackbone doesn\'t have num_hidden_layers attribute''')
	def  A(    self):
							pass


	@unittest.skip('''TimmBackbone initialization is managed on the timm side''')
	def  A(    self):
							pass


	@unittest.skip('''TimmBackbone models doesn\'t have inputs_embeds''')
	def  A(    self):
							pass


	@unittest.skip('''TimmBackbone models doesn\'t have inputs_embeds''')
	def  A(    self):
							pass


	@unittest.skip('''TimmBackbone model cannot be created without specifying a backbone checkpoint''')
	def  A(    self):
							pass


	@unittest.skip('''Only checkpoints on timm can be loaded into TimmBackbone''')
	def  A(    self):
							pass


	@unittest.skip('''model weights aren\'t tied in TimmBackbone.''')
	def  A(    self):
							pass


	@unittest.skip('''model weights aren\'t tied in TimmBackbone.''')
	def  A(    self):
							pass


	@unittest.skip('''Only checkpoints on timm can be loaded into TimmBackbone''')
	def  A(    self):
							pass


	@unittest.skip('''Only checkpoints on timm can be loaded into TimmBackbone''')
	def  A(    self):
							pass


	@unittest.skip('''TimmBackbone doesn\'t have hidden size info in its configuration.''')
	def  A(    self):
							pass


	@unittest.skip('''TimmBackbone doesn\'t support output_attentions.''')
	def  A(    self):
							pass


	@unittest.skip('''Safetensors is not supported by timm.''')
	def  A(    self):
							pass


	@unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.''')
	def  A(    self):
							pass


	def  A(    self):
							__UpperCAmelCase    ,      __UpperCAmelCase			:							Optional[int]	    =	self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
													__UpperCAmelCase			:							Any	    =	model_class(lowercase__)
													__UpperCAmelCase			:							Optional[int]	    =	inspect.signature(model.forward)
													# signature.parameters is an OrderedDict => so arg_names order is deterministic
													__UpperCAmelCase			:							Optional[int]	    =	[*signature.parameters.keys()]

													__UpperCAmelCase			:							Dict	    =	['''pixel_values''']
													self.assertListEqual(arg_names[:1]				,     lowercase__)


	def  A(    self):
							__UpperCAmelCase    ,      __UpperCAmelCase			:							str	    =	self.model_tester.prepare_config_and_inputs_for_common()
							__UpperCAmelCase			:							Tuple	    =	True
							__UpperCAmelCase			:							Optional[int]	    =	self.has_attentions

							# no need to test all models as different heads yield the same functionality
							__UpperCAmelCase			:							Optional[Any]	    =	self.all_model_classes[0]
							__UpperCAmelCase			:							Optional[int]	    =	model_class(lowercase__)
							model.to(lowercase__)

							__UpperCAmelCase			:							Tuple	    =	self._prepare_for_class(lowercase__				,     lowercase__)
							__UpperCAmelCase			:							Optional[int]	    =	model(**lowercase__)
							__UpperCAmelCase			:							List[str]	    =	outputs[0][-1]

							# Encoder-/Decoder-only models
							__UpperCAmelCase			:							Dict	    =	outputs.hidden_states[0]
							hidden_states.retain_grad()

							if self.has_attentions:
													__UpperCAmelCase			:							Optional[int]	    =	outputs.attentions[0]
													attentions.retain_grad()

							output.flatten()[0].backward(retain_graph=lowercase__)

							self.assertIsNotNone(hidden_states.grad)

							if self.has_attentions:
													self.assertIsNotNone(attentions.grad)


	def  A(    self):
							__UpperCAmelCase    ,      __UpperCAmelCase			:							Dict	    =	self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
													__UpperCAmelCase			:							Optional[Any]	    =	model_class(lowercase__)
													model.to(lowercase__)
													model.eval()
													__UpperCAmelCase			:							Union[str, Any]	    =	model(**lowercase__)

													self.assertEqual(len(result.feature_maps)				,     len(config.out_indices))
													self.assertEqual(len(model.channels)				,     len(config.out_indices))

													# Check output of last stage is taken if out_features=None, out_indices=None
													__UpperCAmelCase			:							List[str]	    =	copy.deepcopy(lowercase__)
													__UpperCAmelCase			:							str	    =	None
													__UpperCAmelCase			:							Optional[int]	    =	model_class(lowercase__)
													model.to(lowercase__)
													model.eval()
													__UpperCAmelCase			:							Optional[Any]	    =	model(**lowercase__)

													self.assertEqual(len(result.feature_maps)				,     1)
													self.assertEqual(len(model.channels)				,     1)

													# Check backbone can be initialized with fresh weights
													__UpperCAmelCase			:							Tuple	    =	copy.deepcopy(lowercase__)
													__UpperCAmelCase			:							Optional[int]	    =	False
													__UpperCAmelCase			:							Optional[int]	    =	model_class(lowercase__)
													model.to(lowercase__)
													model.eval()
													__UpperCAmelCase			:							List[Any]	    =	model(**lowercase__)

