
'''simple docstring'''





import inspect
import unittest

from transformers import ResNetConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_backbone_common import BackboneTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
						import torch
						from torch import nn

						from transformers import ResNetBackbone, ResNetForImageClassification, ResNetModel
						from transformers.models.resnet.modeling_resnet import RESNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
						from PIL import Image

						from transformers import AutoImageProcessor

class     a :




				def __init__(				self						:					Union[str, Any],     SCREAMING_SNAKE_CASE_						:					List[Any],     SCREAMING_SNAKE_CASE_						:					Optional[int]=3,     SCREAMING_SNAKE_CASE_						:					Optional[int]=32,     SCREAMING_SNAKE_CASE_						:					Union[str, Any]=3,     SCREAMING_SNAKE_CASE_						:					Optional[int]=10,     SCREAMING_SNAKE_CASE_						:					int=[10, 20, 30, 40],     SCREAMING_SNAKE_CASE_						:					Optional[Any]=[1, 1, 2, 1],     SCREAMING_SNAKE_CASE_						:					int=True,     SCREAMING_SNAKE_CASE_						:					Dict=True,     SCREAMING_SNAKE_CASE_						:					List[str]="relu",     SCREAMING_SNAKE_CASE_						:					Union[str, Any]=3,     SCREAMING_SNAKE_CASE_						:					Union[str, Any]=None,     SCREAMING_SNAKE_CASE_						:					int=["stage2", "stage3", "stage4"],     SCREAMING_SNAKE_CASE_						:					str=[2, 3, 4],     ):
										snake_case							:  List[str]			=     parent
										snake_case							:  List[Any]			=     batch_size
										snake_case							:  Dict			=     image_size
										snake_case							:  Tuple			=     num_channels
										snake_case							:  List[str]			=     embeddings_size
										snake_case							:  Optional[int]			=     hidden_sizes
										snake_case							:  List[Any]			=     depths
										snake_case							:  Union[str, Any]			=     is_training
										snake_case							:  Tuple			=     use_labels
										snake_case							:  Dict			=     hidden_act
										snake_case							:  Union[str, Any]			=     num_labels
										snake_case							:  Dict			=     scope
										snake_case							:  Any			=     len(SCREAMING_SNAKE_CASE_   )
										snake_case							:  Union[str, Any]			=     out_features
										snake_case							:  Optional[Any]			=     out_indices




				def 						__snake_case				(				self						:					Any   ):
										snake_case							:  List[Any]			=     floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]   )

										snake_case							:  Tuple			=     None
										if self.use_labels:
																snake_case							:  int			=     ids_tensor([self.batch_size],     self.num_labels   )

										snake_case							:  Dict			=     self.get_config()

										return config, pixel_values, labels




				def 						__snake_case				(				self						:					Tuple   ):
										return ResNetConfig(
										    num_channels=self.num_channels,     embeddings_size=self.embeddings_size,     hidden_sizes=self.hidden_sizes,     depths=self.depths,     hidden_act=self.hidden_act,     num_labels=self.num_labels,     out_features=self.out_features,     out_indices=self.out_indices,     )




				def 						__snake_case				(				self						:					Union[str, Any],     SCREAMING_SNAKE_CASE_						:					Dict,     SCREAMING_SNAKE_CASE_						:					Union[str, Any],     SCREAMING_SNAKE_CASE_						:					int   ):
										snake_case							:  Optional[Any]			=     ResNetModel(config=SCREAMING_SNAKE_CASE_   )
										model.to(SCREAMING_SNAKE_CASE_   )
										model.eval()
										snake_case							:  Dict			=     model(SCREAMING_SNAKE_CASE_   )
										# expected last hidden states: B, C, H // 32, W // 32
										self.parent.assertEqual(
										    result.last_hidden_state.shape,     (self.batch_size, self.hidden_sizes[-1], self.image_size // 32, self.image_size // 32),     )




				def 						__snake_case				(				self						:					List[str],     SCREAMING_SNAKE_CASE_						:					Any,     SCREAMING_SNAKE_CASE_						:					Optional[int],     SCREAMING_SNAKE_CASE_						:					Any   ):
										snake_case							:  List[Any]			=     self.num_labels
										snake_case							:  List[Any]			=     ResNetForImageClassification(SCREAMING_SNAKE_CASE_   )
										model.to(SCREAMING_SNAKE_CASE_   )
										model.eval()
										snake_case							:  str			=     model(SCREAMING_SNAKE_CASE_,     labels=SCREAMING_SNAKE_CASE_   )
										self.parent.assertEqual(result.logits.shape,     (self.batch_size, self.num_labels)   )




				def 						__snake_case				(				self						:					List[Any],     SCREAMING_SNAKE_CASE_						:					Any,     SCREAMING_SNAKE_CASE_						:					List[str],     SCREAMING_SNAKE_CASE_						:					str   ):
										snake_case							:  Any			=     ResNetBackbone(config=SCREAMING_SNAKE_CASE_   )
										model.to(SCREAMING_SNAKE_CASE_   )
										model.eval()
										snake_case							:  Any			=     model(SCREAMING_SNAKE_CASE_   )

										# verify feature maps
										self.parent.assertEqual(len(result.feature_maps   ),     len(config.out_features   )   )
										self.parent.assertListEqual(list(result.feature_maps[0].shape   ),     [self.batch_size, self.hidden_sizes[1], 4, 4]   )

										# verify channels
										self.parent.assertEqual(len(model.channels   ),     len(config.out_features   )   )
										self.parent.assertListEqual(model.channels,     config.hidden_sizes[1:]   )

										# verify backbone works with out_features=None
										snake_case							:  int			=     None
										snake_case							:  List[str]			=     ResNetBackbone(config=SCREAMING_SNAKE_CASE_   )
										model.to(SCREAMING_SNAKE_CASE_   )
										model.eval()
										snake_case							:  List[str]			=     model(SCREAMING_SNAKE_CASE_   )

										# verify feature maps
										self.parent.assertEqual(len(result.feature_maps   ),     1   )
										self.parent.assertListEqual(list(result.feature_maps[0].shape   ),     [self.batch_size, self.hidden_sizes[-1], 1, 1]   )

										# verify channels
										self.parent.assertEqual(len(model.channels   ),     1   )
										self.parent.assertListEqual(model.channels,     [config.hidden_sizes[-1]]   )




				def 						__snake_case				(				self						:					Optional[int]   ):
										snake_case							:  str			=     self.prepare_config_and_inputs()
										snake_case,       snake_case,       snake_case							:  Optional[int]			=     config_and_inputs
										snake_case							:  List[str]			=     {'''pixel_values''': pixel_values}
										return config, inputs_dict




@require_torch
class     a (					__magic_name__					,__magic_name__					,unittest.TestCase     ):
				_snake_case		     =			(
				    (
				        ResNetModel,
				        ResNetForImageClassification,
				        ResNetBackbone,
				    )
				    if is_torch_available()
				    else ()
				)
				_snake_case		     =			(
				    {'''feature-extraction''': ResNetModel, '''image-classification''': ResNetForImageClassification}
				    if is_torch_available()
				    else {}
				)

				_snake_case		     =			True
				_snake_case		     =			False
				_snake_case		     =			False
				_snake_case		     =			False
				_snake_case		     =			False




				def 						__snake_case				(				self						:					Any   ):
										snake_case							:  Any			=     ResNetModelTester(self   )
										snake_case							:  Dict			=     ConfigTester(self,     config_class=SCREAMING_SNAKE_CASE_,     has_text_modality=SCREAMING_SNAKE_CASE_   )




				def 						__snake_case				(				self						:					str   ):
										self.create_and_test_config_common_properties()
										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 						__snake_case				(				self						:					Union[str, Any]   ):
										return




				@unittest.skip(reason='''ResNet does not use inputs_embeds'''   )
				def 						__snake_case				(				self						:					Union[str, Any]   ):
										pass




				@unittest.skip(reason='''ResNet does not support input and output embeddings'''   )
				def 						__snake_case				(				self						:					Dict   ):
										pass




				def 						__snake_case				(				self						:					Dict   ):
										snake_case,       snake_case							:  List[str]			=     self.model_tester.prepare_config_and_inputs_for_common()

										for model_class in self.all_model_classes:
																snake_case							:  Union[str, Any]			=     model_class(SCREAMING_SNAKE_CASE_   )
																snake_case							:  Dict			=     inspect.signature(model.forward   )
																# signature.parameters is an OrderedDict => so arg_names order is deterministic
																snake_case							:  Dict			=     [*signature.parameters.keys()]

																snake_case							:  Optional[Any]			=     ['''pixel_values''']
																self.assertListEqual(arg_names[:1],     SCREAMING_SNAKE_CASE_   )




				def 						__snake_case				(				self						:					Dict   ):
										snake_case							:  Union[str, Any]			=     self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_model(*SCREAMING_SNAKE_CASE_   )




				def 						__snake_case				(				self						:					Any   ):
										snake_case							:  Optional[Any]			=     self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_backbone(*SCREAMING_SNAKE_CASE_   )




				def 						__snake_case				(				self						:					List[Any]   ):
										snake_case,       snake_case							:  int			=     self.model_tester.prepare_config_and_inputs_for_common()

										for model_class in self.all_model_classes:
																snake_case							:  List[Any]			=     model_class(config=SCREAMING_SNAKE_CASE_   )
																for name, module in model.named_modules():
																						if isinstance(SCREAMING_SNAKE_CASE_,     (nn.BatchNormad, nn.GroupNorm)   ):
																												self.assertTrue(
																												    torch.all(module.weight == 1   ),     msg=F"""Parameter {name} of model {model_class} seems not properly initialized""",     )
																												self.assertTrue(
																												    torch.all(module.bias == 0   ),     msg=F"""Parameter {name} of model {model_class} seems not properly initialized""",     )




				def 						__snake_case				(				self						:					Tuple   ):
										def check_hidden_states_output(SCREAMING_SNAKE_CASE_						:					Optional[Any],     SCREAMING_SNAKE_CASE_						:					str,     SCREAMING_SNAKE_CASE_						:					int   ):
																snake_case							:  Optional[int]			=     model_class(SCREAMING_SNAKE_CASE_   )
																model.to(SCREAMING_SNAKE_CASE_   )
																model.eval()

																with torch.no_grad():
																						snake_case							:  Optional[int]			=     model(**self._prepare_for_class(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_   )   )

																snake_case							:  Optional[Any]			=     outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

																snake_case							:  Union[str, Any]			=     self.model_tester.num_stages
																self.assertEqual(len(SCREAMING_SNAKE_CASE_   ),     expected_num_stages + 1   )

																# ResNet's feature maps are of shape (batch_size, num_channels, height, width)
																self.assertListEqual(
																    list(hidden_states[0].shape[-2:]   ),     [self.model_tester.image_size // 4, self.model_tester.image_size // 4],     )

										snake_case,       snake_case							:  int			=     self.model_tester.prepare_config_and_inputs_for_common()
										snake_case							:  List[str]			=     ['''basic''', '''bottleneck''']
										for model_class in self.all_model_classes:
																for layer_type in layers_type:
																						snake_case							:  Tuple			=     layer_type
																						snake_case							:  Union[str, Any]			=     True
																						check_hidden_states_output(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_   )

																						# check that output_hidden_states also work using config
																						del inputs_dict["output_hidden_states"]
																						snake_case							:  Optional[Any]			=     True

																						check_hidden_states_output(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_   )




				@unittest.skip(reason='''ResNet does not use feedforward chunking'''   )
				def 						__snake_case				(				self						:					str   ):
										pass




				def 						__snake_case				(				self						:					str   ):
										snake_case							:  List[str]			=     self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_for_image_classification(*SCREAMING_SNAKE_CASE_   )




				@slow
				def 						__snake_case				(				self						:					Any   ):
										for model_name in RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																snake_case							:  str			=     ResNetModel.from_pretrained(SCREAMING_SNAKE_CASE_   )
																self.assertIsNotNone(SCREAMING_SNAKE_CASE_   )




def   A			(							):
						snake_case							:  Union[str, Any]			=     Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''							)
						return image

@require_torch
@require_vision
class     a (					unittest.TestCase     ):




				@cached_property
				def 						__snake_case				(				self						:					Any   ):
										return (
										    AutoImageProcessor.from_pretrained(RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]   )
										    if is_vision_available()
										    else None
										)




				@slow
				def 						__snake_case				(				self						:					Optional[Any]   ):
										snake_case							:  Union[str, Any]			=     ResNetForImageClassification.from_pretrained(RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]   ).to(SCREAMING_SNAKE_CASE_   )

										snake_case							:  List[str]			=     self.default_image_processor
										snake_case							:  Any			=     prepare_img()
										snake_case							:  str			=     image_processor(images=SCREAMING_SNAKE_CASE_,     return_tensors='''pt'''   ).to(SCREAMING_SNAKE_CASE_   )

										# forward pass
										with torch.no_grad():
																snake_case							:  Optional[int]			=     model(**SCREAMING_SNAKE_CASE_   )

										# verify the logits
										snake_case							:  Dict			=     torch.Size((1, 10_00)   )
										self.assertEqual(outputs.logits.shape,     SCREAMING_SNAKE_CASE_   )

										snake_case							:  List[str]			=     torch.tensor([-11.1069, -9.7877, -8.3777]   ).to(SCREAMING_SNAKE_CASE_   )

										self.assertTrue(torch.allclose(outputs.logits[0, :3],     SCREAMING_SNAKE_CASE_,     atol=1e-4   )   )




@require_torch
class     a (					__magic_name__					,unittest.TestCase     ):
				_snake_case		     =			(ResNetBackbone,) if is_torch_available() else ()
				_snake_case		     =			False
				_snake_case		     =			ResNetConfig




				def 						__snake_case				(				self						:					Union[str, Any]   ):
										snake_case							:  List[str]			=     ResNetModelTester(self   )









"""simple docstring"""



import importlib
import inspect
from functools import wraps
from typing import TYPE_CHECKING, Optional, Union

from .download.streaming_download_manager import (
    xbasename,
    xdirname,
    xet_parse,
    xexists,
    xgetsize,
    xglob,
    xgzip_open,
    xisdir,
    xisfile,
    xjoin,
    xlistdir,
    xnumpy_load,
    xopen,
    xpandas_read_csv,
    xpandas_read_excel,
    xPath,
    xrelpath,
    xsio_loadmat,
    xsplit,
    xsplitext,
    xwalk,
    xxml_dom_minidom_parse,
)
from .utils.logging import get_logger
from .utils.patching import patch_submodule
from .utils.py_utils import get_imports


A_     =				get_logger(__name__)


if TYPE_CHECKING:
						from .builder import DatasetBuilder




def   _lowerCAmelCase		(				UpperCAmelCase__							:       Optional[int],    UpperCAmelCase__							:       Optional[Union[str, bool]] = None   )		->Optional[Any]:
						A__    :	int			=	importlib.import_module(UpperCAmelCase__   )

						# TODO(QL): always update the module to add subsequent new authentication
						if hasattr(UpperCAmelCase__,    """_patched_for_streaming"""   ) and module._patched_for_streaming:
												return

						def wrap_auth(UpperCAmelCase__							:       List[str]   ):
												@wraps(UpperCAmelCase__   )
												def wrapper(*UpperCAmelCase__							:       Dict,    **UpperCAmelCase__							:       Dict   ):
																		return function(*UpperCAmelCase__,    use_auth_token=UpperCAmelCase__,    **UpperCAmelCase__   )

												A__    :	List[Any]			=	"""wrap_auth"""
												return wrapper

						# open files in a streaming fashion
						patch_submodule(UpperCAmelCase__,    """open""",    wrap_auth(UpperCAmelCase__   )   ).start()
						patch_submodule(UpperCAmelCase__,    """os.listdir""",    wrap_auth(UpperCAmelCase__   )   ).start()
						patch_submodule(UpperCAmelCase__,    """os.walk""",    wrap_auth(UpperCAmelCase__   )   ).start()
						patch_submodule(UpperCAmelCase__,    """glob.glob""",    wrap_auth(UpperCAmelCase__   )   ).start()
						# allow to navigate in remote zip files
						patch_submodule(UpperCAmelCase__,    """os.path.join""",    UpperCAmelCase__   ).start()
						patch_submodule(UpperCAmelCase__,    """os.path.dirname""",    UpperCAmelCase__   ).start()
						patch_submodule(UpperCAmelCase__,    """os.path.basename""",    UpperCAmelCase__   ).start()
						patch_submodule(UpperCAmelCase__,    """os.path.relpath""",    UpperCAmelCase__   ).start()
						patch_submodule(UpperCAmelCase__,    """os.path.split""",    UpperCAmelCase__   ).start()
						patch_submodule(UpperCAmelCase__,    """os.path.splitext""",    UpperCAmelCase__   ).start()
						# allow checks on paths
						patch_submodule(UpperCAmelCase__,    """os.path.exists""",    wrap_auth(UpperCAmelCase__   )   ).start()
						patch_submodule(UpperCAmelCase__,    """os.path.isdir""",    wrap_auth(UpperCAmelCase__   )   ).start()
						patch_submodule(UpperCAmelCase__,    """os.path.isfile""",    wrap_auth(UpperCAmelCase__   )   ).start()
						patch_submodule(UpperCAmelCase__,    """os.path.getsize""",    wrap_auth(UpperCAmelCase__   )   ).start()
						patch_submodule(UpperCAmelCase__,    """pathlib.Path""",    UpperCAmelCase__   ).start()
						# file readers
						patch_submodule(UpperCAmelCase__,    """gzip.open""",    wrap_auth(UpperCAmelCase__   )   ).start()
						patch_submodule(UpperCAmelCase__,    """numpy.load""",    wrap_auth(UpperCAmelCase__   )   ).start()
						patch_submodule(UpperCAmelCase__,    """pandas.read_csv""",    wrap_auth(UpperCAmelCase__   ),    attrs=["""__version__"""]   ).start()
						patch_submodule(UpperCAmelCase__,    """pandas.read_excel""",    wrap_auth(UpperCAmelCase__   ),    attrs=["""__version__"""]   ).start()
						patch_submodule(UpperCAmelCase__,    """scipy.io.loadmat""",    wrap_auth(UpperCAmelCase__   ),    attrs=["""__version__"""]   ).start()
						patch_submodule(UpperCAmelCase__,    """xml.etree.ElementTree.parse""",    wrap_auth(UpperCAmelCase__   )   ).start()
						patch_submodule(UpperCAmelCase__,    """xml.dom.minidom.parse""",    wrap_auth(UpperCAmelCase__   )   ).start()
						A__    :	Any			=	True




def   _lowerCAmelCase		(				UpperCAmelCase__							:       "DatasetBuilder"   )		->Tuple:
						extend_module_for_streaming(builder.__module__,    use_auth_token=builder.use_auth_token   )
						# if needed, we also have to extend additional internal imports (like wmt14 -> wmt_utils)
						if not builder.__module__.startswith("""datasets."""   ):  # check that it's not a packaged builder like csv
												for imports in get_imports(inspect.getfile(builder.__class__   )   ):
																		if imports[0] == "internal":
																								A__    :	Any			=	imports[1]
																								A__    :	Any			=	""".""".join(builder.__module__.split("""."""   )[:-1] + [internal_import_name]   )
																								extend_module_for_streaming(UpperCAmelCase__,    use_auth_token=builder.use_auth_token   )

    # builders can inherit from other builders that might use streaming functionality
    # (for example, ImageFolder and AudioFolder inherit from FolderBuilder which implements examples generation)
    # but these parents builders are not patched automatically as they are not instantiated, so we patch them here
						from .builder import DatasetBuilder

						A__    :	Any			=	[
						    cls.__module__
						    for cls in type(UpperCAmelCase__   ).__mro__[1:]  # make sure it's not the same module we've already patched
						    if issubclass(cls,    UpperCAmelCase__   ) and cls.__module__ != DatasetBuilder.__module__
						]  # check it's not a standard builder from datasets.builder
						for module in parent_builder_modules:
												extend_module_for_streaming(UpperCAmelCase__,    use_auth_token=builder.use_auth_token   )






