





from typing import Tuple, Union

from ...modeling_outputs import BackboneOutput
from ...modeling_utils import PreTrainedModel
from ...utils import is_timm_available, is_torch_available, requires_backends
from ...utils.backbone_utils import BackboneMixin
from .configuration_timm_backbone import TimmBackboneConfig


if is_timm_available():
      import timm


if is_torch_available():
      from torch import Tensor

class 		UpperCamelCase__			(				lowerCAmelCase_							,				lowerCAmelCase_       ):



     '''simple docstring'''

     __snake_case   :     Any		     =      "pixel_values"
     __snake_case   :     Union[str, Any]		     =      False
     __snake_case   :     Dict		     =      TimmBackboneConfig






     def __init__(							self						:   Tuple						,lowerCamelCase__						:   str						,**lowerCamelCase__						:   int	)					->		Any:




           '''simple docstring'''


           requires_backends(self						,"""timm"""	)
           super().__init__(lowerCamelCase__	)
           SCREAMING_SNAKE_CASE		=						config

           if config.backbone is None:
                 raise ValueError("""backbone is not set in the config. Please set it to a timm model name."""	)

           if config.backbone not in timm.list_models():
                 raise ValueError(F"""backbone {config.backbone} is not supported by timm."""	)

           if hasattr(lowerCamelCase__						,"""out_features"""	) and config.out_features is not None:
                 raise ValueError("""out_features is not supported by TimmBackbone. Please use out_indices instead."""	)

           SCREAMING_SNAKE_CASE		=						getattr(lowerCamelCase__						,"""use_pretrained_backbone"""						,lowerCamelCase__	)
           if pretrained is None:
                 raise ValueError("""use_pretrained_backbone is not set in the config. Please set it to True or False."""	)

           # We just take the final layer by default. This matches the default for the transformers models.
           SCREAMING_SNAKE_CASE		=						config.out_indices if getattr(lowerCamelCase__						,"""out_indices"""						,lowerCamelCase__	) is not None else (-1,)

           SCREAMING_SNAKE_CASE		=						timm.create_model(
               config.backbone						,pretrained=lowerCamelCase__						,features_only=config.features_only						,in_chans=config.num_channels						,out_indices=lowerCamelCase__						,**lowerCamelCase__						,)
           # These are used to control the output of the model when called. If output_hidden_states is True, then
           # return_layers is modified to include all layers.
           SCREAMING_SNAKE_CASE		=						self._backbone.return_layers
           SCREAMING_SNAKE_CASE		=						{layer["""module"""]: str(lowerCamelCase__	) for i, layer in enumerate(self._backbone.feature_info.info	)}
           super()._init_backbone(lowerCamelCase__	)






     @classmethod
     def     SCREAMING_SNAKE_CASE__  (							cls						:   str						,lowerCamelCase__						:   Dict						,*lowerCamelCase__						:   Dict						,**lowerCamelCase__						:   Any	)					->		str:




           '''simple docstring'''


           requires_backends(cls						,["""vision""", """timm"""]	)
           from ...models.timm_backbone import TimmBackboneConfig

           SCREAMING_SNAKE_CASE		=						kwargs.pop("""config"""						,TimmBackboneConfig()	)

           SCREAMING_SNAKE_CASE		=						kwargs.pop("""use_timm_backbone"""						,lowerCamelCase__	)
           if not use_timm:
                 raise ValueError("""use_timm_backbone must be True for timm backbones"""	)

           SCREAMING_SNAKE_CASE		=						kwargs.pop("""num_channels"""						,config.num_channels	)
           SCREAMING_SNAKE_CASE		=						kwargs.pop("""features_only"""						,config.features_only	)
           SCREAMING_SNAKE_CASE		=						kwargs.pop("""use_pretrained_backbone"""						,config.use_pretrained_backbone	)
           SCREAMING_SNAKE_CASE		=						kwargs.pop("""out_indices"""						,config.out_indices	)
           SCREAMING_SNAKE_CASE		=						TimmBackboneConfig(
               backbone=lowerCamelCase__						,num_channels=lowerCamelCase__						,features_only=lowerCamelCase__						,use_pretrained_backbone=lowerCamelCase__						,out_indices=lowerCamelCase__						,)
           return super()._from_config(lowerCamelCase__						,**lowerCamelCase__	)






     def     SCREAMING_SNAKE_CASE__  (							self						:   Optional[int]						,lowerCamelCase__						:   int	)					->		Dict:




           '''simple docstring'''


           pass






     def     SCREAMING_SNAKE_CASE__  (							self						:   List[str]						,lowerCamelCase__						:   int						,lowerCamelCase__						:   List[Any]=None						,lowerCamelCase__						:   Optional[int]=None						,lowerCamelCase__						:   List[str]=None						,**lowerCamelCase__						:   Union[str, Any]	)					->		Union[BackboneOutput, Tuple[Tensor, ...]]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						return_dict if return_dict is not None else self.config.use_return_dict
           SCREAMING_SNAKE_CASE		=						(
               output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
           )
           SCREAMING_SNAKE_CASE		=						output_attentions if output_attentions is not None else self.config.output_attentions

           if output_attentions:
                 raise ValueError("""Cannot output attentions for timm backbones at the moment"""	)

           if output_hidden_states:
                 # We modify the return layers to include all the stages of the backbone
                 SCREAMING_SNAKE_CASE		=						self._all_layers
                 SCREAMING_SNAKE_CASE		=						self._backbone(lowerCamelCase__						,**lowerCamelCase__	)
                 SCREAMING_SNAKE_CASE		=						self._return_layers
                 SCREAMING_SNAKE_CASE		=						tuple(hidden_states[i] for i in self.out_indices	)
           else:
                 SCREAMING_SNAKE_CASE		=						self._backbone(lowerCamelCase__						,**lowerCamelCase__	)
                 SCREAMING_SNAKE_CASE		=						None

           SCREAMING_SNAKE_CASE		=						tuple(lowerCamelCase__	)
           SCREAMING_SNAKE_CASE		=						tuple(lowerCamelCase__	) if hidden_states is not None else None

           if not return_dict:
                 SCREAMING_SNAKE_CASE		=						(feature_maps,)
                 if output_hidden_states:
                       SCREAMING_SNAKE_CASE		=						output + (hidden_states,)
                 return output

           return BackboneOutput(feature_maps=lowerCamelCase__						,hidden_states=lowerCamelCase__						,attentions=lowerCamelCase__	)









import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


SCREAMING_SNAKE_CASE_						       =	"""\
    Text data.
    Second line of data."""

SCREAMING_SNAKE_CASE_						       =	"""file"""



@pytest.fixture(scope="""session"""     )
def        __lowercase       (				_SCREAMING_SNAKE_CASE     )			->       Union[str, Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""     ) / (FILE_PATH + """.zstd""")
      SCREAMING_SNAKE_CASE		=						bytes(_SCREAMING_SNAKE_CASE			,						"""utf-8"""     )
      with zstd.open(_SCREAMING_SNAKE_CASE			,						"""wb"""     ) as f:
            f.write(_SCREAMING_SNAKE_CASE     )
      return path



@pytest.fixture
def        __lowercase       (				_SCREAMING_SNAKE_CASE     )			->       List[Any]:


      '''simple docstring'''


      with open(os.path.join(tmpfs.local_root_dir			,						_SCREAMING_SNAKE_CASE     )			,						"""w"""     ) as f:
            f.write(_SCREAMING_SNAKE_CASE     )
      return FILE_PATH



@pytest.mark.parametrize("""compression_format"""			,						["""gzip""", """xz""", """zstd"""]     )
def        __lowercase       (				_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     )			->       Tuple:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						{"""gzip""": gz_file, """xz""": xz_file, """zstd""": zstd_path}
      SCREAMING_SNAKE_CASE		=						input_paths[compression_format]
      SCREAMING_SNAKE_CASE		=						tmp_path / """cache"""
      SCREAMING_SNAKE_CASE		=						DownloadConfig(cache_dir=_SCREAMING_SNAKE_CASE			,						extract_compressed_file=_SCREAMING_SNAKE_CASE     )
      SCREAMING_SNAKE_CASE		=						cached_path(_SCREAMING_SNAKE_CASE			,						download_config=_SCREAMING_SNAKE_CASE     )
      with open(_SCREAMING_SNAKE_CASE     ) as f:
            SCREAMING_SNAKE_CASE		=						f.read()
      with open(_SCREAMING_SNAKE_CASE     ) as f:
            SCREAMING_SNAKE_CASE		=						f.read()
      assert extracted_file_content == expected_file_content



@pytest.mark.parametrize("""default_extracted"""			,						[True, False]     )
@pytest.mark.parametrize("""default_cache_dir"""			,						[True, False]     )
def        __lowercase       (				_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     )			->       List[str]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						"""custom_cache"""
      SCREAMING_SNAKE_CASE		=						"""custom_extracted_dir"""
      SCREAMING_SNAKE_CASE		=						tmp_path / """custom_extracted_path"""
      if default_extracted:
            SCREAMING_SNAKE_CASE		=						("""downloads""" if default_cache_dir else custom_cache_dir, """extracted""")
      else:
            monkeypatch.setattr("""datasets.config.EXTRACTED_DATASETS_DIR"""			,						_SCREAMING_SNAKE_CASE     )
            monkeypatch.setattr("""datasets.config.EXTRACTED_DATASETS_PATH"""			,						str(_SCREAMING_SNAKE_CASE     )     )
            SCREAMING_SNAKE_CASE		=						custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

      SCREAMING_SNAKE_CASE		=						xz_file
      SCREAMING_SNAKE_CASE		=						(
          DownloadConfig(extract_compressed_file=_SCREAMING_SNAKE_CASE     )
          if default_cache_dir
          else DownloadConfig(cache_dir=tmp_path / custom_cache_dir			,						extract_compressed_file=_SCREAMING_SNAKE_CASE     )
      )
      SCREAMING_SNAKE_CASE		=						cached_path(_SCREAMING_SNAKE_CASE			,						download_config=_SCREAMING_SNAKE_CASE     )
      assert Path(_SCREAMING_SNAKE_CASE     ).parent.parts[-2:] == expected



def        __lowercase       (				_SCREAMING_SNAKE_CASE     )			->       int:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(Path(_SCREAMING_SNAKE_CASE     ).resolve()     )
      assert cached_path(_SCREAMING_SNAKE_CASE     ) == text_file
      # relative path
      SCREAMING_SNAKE_CASE		=						str(Path(_SCREAMING_SNAKE_CASE     ).resolve().relative_to(Path(os.getcwd()     )     )     )
      assert cached_path(_SCREAMING_SNAKE_CASE     ) == text_file



def        __lowercase       (				_SCREAMING_SNAKE_CASE     )			->       Dict:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path.resolve() / """__missing_file__.txt"""     )
      with pytest.raises(_SCREAMING_SNAKE_CASE     ):
            cached_path(_SCREAMING_SNAKE_CASE     )
      # relative path
      SCREAMING_SNAKE_CASE		=						"""./__missing_file__.txt"""
      with pytest.raises(_SCREAMING_SNAKE_CASE     ):
            cached_path(_SCREAMING_SNAKE_CASE     )



def        __lowercase       (				_SCREAMING_SNAKE_CASE     )			->       str:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						get_from_cache(F"""tmp://{tmpfs_file}"""     )
      with open(_SCREAMING_SNAKE_CASE     ) as f:
            SCREAMING_SNAKE_CASE		=						f.read()
      assert output_file_content == FILE_CONTENT



@patch("""datasets.config.HF_DATASETS_OFFLINE"""			,						_SCREAMING_SNAKE_CASE     )
def        __lowercase       (				)			->       Dict:


      '''simple docstring'''


      with pytest.raises(_SCREAMING_SNAKE_CASE     ):
            cached_path("""https://huggingface.co"""     )



@patch("""datasets.config.HF_DATASETS_OFFLINE"""			,						_SCREAMING_SNAKE_CASE     )
def        __lowercase       (				_SCREAMING_SNAKE_CASE     )			->       Union[str, Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""     ) / """file.html"""
      with pytest.raises(_SCREAMING_SNAKE_CASE     ):
            http_get("""https://huggingface.co"""			,						temp_file=_SCREAMING_SNAKE_CASE     )
      with pytest.raises(_SCREAMING_SNAKE_CASE     ):
            http_head("""https://huggingface.co"""     )



@patch("""datasets.config.HF_DATASETS_OFFLINE"""			,						_SCREAMING_SNAKE_CASE     )
def        __lowercase       (				_SCREAMING_SNAKE_CASE     )			->       Optional[int]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""     ) / """file.html"""
      with pytest.raises(_SCREAMING_SNAKE_CASE     ):
            ftp_get("""ftp://huggingface.co"""			,						temp_file=_SCREAMING_SNAKE_CASE     )
      with pytest.raises(_SCREAMING_SNAKE_CASE     ):
            ftp_head("""ftp://huggingface.co"""     )



@patch("""datasets.config.HF_DATASETS_OFFLINE"""			,						_SCREAMING_SNAKE_CASE     )
def        __lowercase       (				_SCREAMING_SNAKE_CASE     )			->       int:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""     ) / """file.html"""
      with pytest.raises(_SCREAMING_SNAKE_CASE     ):
            fsspec_get("""s3://huggingface.co"""			,						temp_file=_SCREAMING_SNAKE_CASE     )
      with pytest.raises(_SCREAMING_SNAKE_CASE     ):
            fsspec_head("""s3://huggingface.co"""     )


