

'''simple docstring'''
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,
)


_UpperCAmelCase  :		int  =		'''\
    Text data.
    Second line of data.'''

_UpperCAmelCase  :		Optional[int]  =		'''file'''




@pytest.fixture(scope='''session'''	)
def 	UpperCamelCase   (       lowercase_						:    str	)		->	int:




  '''simple docstring'''



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




@pytest.fixture
def 	UpperCamelCase   (       lowercase_						:    Optional[int]	)		->	List[Any]:




  '''simple docstring'''



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




@pytest.mark.parametrize('''compression_format'''     ,					['''gzip''', '''xz''', '''zstd''']	)
def 	UpperCamelCase   (       lowercase_						:    str     ,					lowercase_						:    Dict     ,					lowercase_						:    Optional[Any]     ,					lowercase_						:    Any     ,					lowercase_						:    Optional[Any]     ,					lowercase_						:    Any	)		->	List[Any]:




  '''simple docstring'''



  lowercase									={'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
  lowercase									=input_paths[compression_format]
  lowercase									=tmp_path / '''cache'''
  lowercase									=DownloadConfig(cache_dir=lowercase_     ,					extract_compressed_file=lowercase_	)
  lowercase									=cached_path(lowercase_     ,					download_config=lowercase_	)
  with open(lowercase_	) as f:
    lowercase									=f.read()
  with open(lowercase_	) as f:
    lowercase									=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 	UpperCamelCase   (       lowercase_						:    List[Any]     ,					lowercase_						:    Optional[int]     ,					lowercase_						:    Optional[int]     ,					lowercase_						:    str     ,					lowercase_						:    List[Any]	)		->	Union[str, Any]:




  '''simple docstring'''



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

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




def 	UpperCamelCase   (       lowercase_						:    List[str]	)		->	Any:




  '''simple docstring'''



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




def 	UpperCamelCase   (       lowercase_						:    int	)		->	Dict:




  '''simple docstring'''



  lowercase									=str(tmp_path.resolve() / '''__missing_file__.txt'''	)
  with pytest.raises(lowercase_	):
    cached_path(lowercase_	)
  # relative path
  lowercase									='''./__missing_file__.txt'''
  with pytest.raises(lowercase_	):
    cached_path(lowercase_	)




def 	UpperCamelCase   (       lowercase_						:    List[str]	)		->	Tuple:




  '''simple docstring'''



  lowercase									=get_from_cache(f'tmp://{tmpfs_file}'	)
  with open(lowercase_	) as f:
    lowercase									=f.read()
  assert output_file_content == FILE_CONTENT




@patch('''datasets.config.HF_DATASETS_OFFLINE'''     ,					lowercase_	)
def 	UpperCamelCase   (       )		->	Union[str, Any]:




  '''simple docstring'''



  with pytest.raises(lowercase_	):
    cached_path('''https://huggingface.co'''	)




@patch('''datasets.config.HF_DATASETS_OFFLINE'''     ,					lowercase_	)
def 	UpperCamelCase   (       lowercase_						:    Dict	)		->	int:




  '''simple docstring'''



  lowercase									=tmp_path_factory.mktemp('''data'''	) / '''file.html'''
  with pytest.raises(lowercase_	):
    http_get('''https://huggingface.co'''     ,					temp_file=lowercase_	)
  with pytest.raises(lowercase_	):
    http_head('''https://huggingface.co'''	)




@patch('''datasets.config.HF_DATASETS_OFFLINE'''     ,					lowercase_	)
def 	UpperCamelCase   (       lowercase_						:    Tuple	)		->	Union[str, Any]:




  '''simple docstring'''



  lowercase									=tmp_path_factory.mktemp('''data'''	) / '''file.html'''
  with pytest.raises(lowercase_	):
    ftp_get('''ftp://huggingface.co'''     ,					temp_file=lowercase_	)
  with pytest.raises(lowercase_	):
    ftp_head('''ftp://huggingface.co'''	)




@patch('''datasets.config.HF_DATASETS_OFFLINE'''     ,					lowercase_	)
def 	UpperCamelCase   (       lowercase_						:    Dict	)		->	Any:




  '''simple docstring'''



  lowercase									=tmp_path_factory.mktemp('''data'''	) / '''file.html'''
  with pytest.raises(lowercase_	):
    fsspec_get('''s3://huggingface.co'''     ,					temp_file=lowercase_	)
  with pytest.raises(lowercase_	):
    fsspec_head('''s3://huggingface.co'''	)




'''simple docstring'''






from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
 from PIL import Image

 from ..image_utils import load_image

if is_tf_available():
 import tensorflow as tf

 from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
 from ..tf_utils import stable_softmax

if is_torch_available():
 from ..models.auto.modeling_auto import MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING

lowercase				=logging.get_logger(__name__)






@add_end_docstrings(lowerCAmelCase							)
class    __magic_name__				(						lowerCAmelCase							):






       def __init__(     self    ,				*snake_case    ,				**snake_case)		->    Dict:




          '''simple docstring'''
          super().__init__(*snake_case    ,				**snake_case)
          requires_backends(self    ,				'vision')
          self.check_model_type(
              TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
              if self.framework == 'tf'
              else MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING)






       def      lowerCAmelCase						(     self    ,				snake_case=None)		->    str:




          '''simple docstring'''
          _UpperCAmelCase  :  Any				       ={}
          if top_k is not None:
             _UpperCAmelCase  :  Optional[int]				       =top_k
          return {}, {}, postprocess_params






       def __call__(     self    ,				snake_case    ,				**snake_case)		->    List[str]:




          '''simple docstring'''
          return super().__call__(snake_case    ,				**snake_case)






       def      lowerCAmelCase						(     self    ,				snake_case)		->    List[str]:




          '''simple docstring'''
          _UpperCAmelCase  :  int				       =load_image(snake_case)
          _UpperCAmelCase  :  Tuple				       =self.image_processor(images=snake_case    ,				return_tensors=self.framework)
          return model_inputs






       def      lowerCAmelCase						(     self    ,				snake_case)		->    Optional[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  List[str]				       =self.model(**snake_case)
          return model_outputs






       def      lowerCAmelCase						(     self    ,				snake_case    ,				snake_case=5)		->    Any:




          '''simple docstring'''
          if top_k > self.model.config.num_labels:
             _UpperCAmelCase  :  Optional[Any]				       =self.model.config.num_labels

          if self.framework == "pt":
             _UpperCAmelCase  :  List[Any]				       =model_outputs.logits.softmax(-1)[0]
             _UpperCAmelCase       ,			_UpperCAmelCase  :  Optional[Any]				       =probs.topk(snake_case)
          elif self.framework == "tf":
             _UpperCAmelCase  :  int				       =stable_softmax(model_outputs.logits    ,				axis=-1)[0]
             _UpperCAmelCase  :  List[Any]				       =tf.math.top_k(snake_case    ,				k=snake_case)
             _UpperCAmelCase       ,			_UpperCAmelCase  :  List[str]				       =topk.values.numpy(), topk.indices.numpy()
          else:
             raise ValueError(f"Unsupported framework: {self.framework}")

          _UpperCAmelCase  :  int				       =scores.tolist()
          _UpperCAmelCase  :  Tuple				       =ids.tolist()
          return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(snake_case    ,				snake_case)]

