





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,
)


a_    :						str    		=      "\\n    Text data.\n    Second line of data."

a_    :						List[Any]    		=      "file"


@pytest.fixture(scope='session'	)
def        __lowerCAmelCase    (       _UpperCamelCase		:  Optional[int]	)		->							Tuple:
 '''simple docstring'''


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


@pytest.fixture
def        __lowerCAmelCase    (       _UpperCamelCase		:  Tuple	)		->							int:
 '''simple docstring'''


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


@pytest.mark.parametrize('compression_format'		,							['gzip', 'xz', 'zstd']	)
def        __lowerCAmelCase    (       _UpperCamelCase		:  Tuple		,							_UpperCamelCase		:  Optional[int]		,							_UpperCamelCase		:  str		,							_UpperCamelCase		:  Optional[Any]		,							_UpperCamelCase		:  Dict		,							_UpperCamelCase		:  List[str]	)		->							List[str]:
 '''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=__lowerCAmelCase		,							extract_compressed_file=__lowerCAmelCase	)
 SCREAMING_SNAKE_CASE								=			cached_path(__lowerCAmelCase		,							download_config=__lowerCAmelCase	)
 with open(__lowerCAmelCase	) as f:
  SCREAMING_SNAKE_CASE								=			f.read()
 with open(__lowerCAmelCase	) 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        __lowerCAmelCase    (       _UpperCamelCase		:  Optional[int]		,							_UpperCamelCase		:  Dict		,							_UpperCamelCase		:  int		,							_UpperCamelCase		:  str		,							_UpperCamelCase		:  int	)		->							Union[str, Any]:
 '''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'		,							__lowerCAmelCase	)
  monkeypatch.setattr('datasets.config.EXTRACTED_DATASETS_PATH'		,							str(__lowerCAmelCase	)	)
  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=__lowerCAmelCase	)
     if default_cache_dir
     else DownloadConfig(cache_dir=tmp_path / custom_cache_dir		,							extract_compressed_file=__lowerCAmelCase	)
 )
 SCREAMING_SNAKE_CASE								=			cached_path(__lowerCAmelCase		,							download_config=__lowerCAmelCase	)
 assert Path(__lowerCAmelCase	).parent.parts[-2:] == expected


def        __lowerCAmelCase    (       _UpperCamelCase		:  Optional[int]	)		->							Any:
 '''simple docstring'''


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


def        __lowerCAmelCase    (       _UpperCamelCase		:  Optional[Any]	)		->							Union[str, Any]:
 '''simple docstring'''


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


def        __lowerCAmelCase    (       _UpperCamelCase		:  Dict	)		->							Any:
 '''simple docstring'''


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


@patch('datasets.config.HF_DATASETS_OFFLINE'		,							__lowerCAmelCase	)
def        __lowerCAmelCase    (       )		->							List[Any]:
 '''simple docstring'''


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


@patch('datasets.config.HF_DATASETS_OFFLINE'		,							__lowerCAmelCase	)
def        __lowerCAmelCase    (       _UpperCamelCase		:  Dict	)		->							Optional[Any]:
 '''simple docstring'''


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


@patch('datasets.config.HF_DATASETS_OFFLINE'		,							__lowerCAmelCase	)
def        __lowerCAmelCase    (       _UpperCamelCase		:  str	)		->							Optional[int]:
 '''simple docstring'''


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




@patch('datasets.config.HF_DATASETS_OFFLINE'		,							__lowerCAmelCase	)
def        __lowerCAmelCase    (       _UpperCamelCase		:  List[Any]	)		->							str:
 '''simple docstring'''


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




from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_vision_available():
      from PIL import Image

      from ..image_utils import load_image

if is_torch_available():
      import torch

      from transformers.modeling_outputs import BaseModelOutput

      from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

a_    :						Dict    		=      logging.get_logger(__name__)




@add_end_docstrings(SCREAMING_SNAKE_CASE   )
class      UpperCamelCase (     SCREAMING_SNAKE_CASE   ):

    def __init__(			self :      Any							,    **snake_case__ :      Optional[int]    ):

     """simple docstring"""






     super().__init__(**snake_case__    )

     if self.framework == "tf":
      raise ValueError(F"""The {self.__class__} is only available in PyTorch."""    )

     requires_backends(self							,    'vision'    )
     self.check_model_type(snake_case__    )

    def __call__(			self :      List[Any]							,    snake_case__ :      Union[str, "Image.Image", List[Dict[str, Any]]]							,    snake_case__ :      Union[str, List[str]] = None							,    **snake_case__ :      Union[str, Any]							,    ):

     """simple docstring"""






     if "text_queries" in kwargs:
      SCREAMING_SNAKE_CASE								=			kwargs.pop('text_queries'    )

     if isinstance(snake_case__							,    (str, Image.Image)    ):
      SCREAMING_SNAKE_CASE								=			{'image': image, 'candidate_labels': candidate_labels}
     else:
      SCREAMING_SNAKE_CASE								=			image
     SCREAMING_SNAKE_CASE								=			super().__call__(snake_case__							,    **snake_case__    )
     return results

    def UpperCamelCase			(			self :      Union[str, Any]							,    **snake_case__ :      Optional[Any]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			{}
     if "threshold" in kwargs:
      SCREAMING_SNAKE_CASE								=			kwargs['threshold']
     if "top_k" in kwargs:
      SCREAMING_SNAKE_CASE								=			kwargs['top_k']
     return {}, {}, postprocess_params

    def UpperCamelCase			(			self :      List[Any]							,    snake_case__ :      Optional[int]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			load_image(inputs['image']    )
     SCREAMING_SNAKE_CASE								=			inputs['candidate_labels']
     if isinstance(snake_case__							,    snake_case__    ):
      SCREAMING_SNAKE_CASE								=			candidate_labels.split(','    )

     SCREAMING_SNAKE_CASE								=			torch.tensor([[image.height, image.width]]							,    dtype=torch.intaa    )
     for i, candidate_label in enumerate(snake_case__    ):
      SCREAMING_SNAKE_CASE								=			self.tokenizer(snake_case__							,    return_tensors=self.framework    )
      SCREAMING_SNAKE_CASE								=			self.image_processor(snake_case__							,    return_tensors=self.framework    )
      yield {
          "is_last": i == len(snake_case__    ) - 1,
          "target_size": target_size,
          "candidate_label": candidate_label,
          **text_inputs,
          **image_features,
      }

    def UpperCamelCase			(			self :      Any							,    snake_case__ :      Dict    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			model_inputs.pop('target_size'    )
     SCREAMING_SNAKE_CASE								=			model_inputs.pop('candidate_label'    )
     SCREAMING_SNAKE_CASE								=			model_inputs.pop('is_last'    )

     SCREAMING_SNAKE_CASE								=			self.model(**snake_case__    )

     SCREAMING_SNAKE_CASE								=			{'target_size': target_size, 'candidate_label': candidate_label, 'is_last': is_last, **outputs}
     return model_outputs

    def UpperCamelCase			(			self :      Union[str, Any]							,    snake_case__ :      Tuple							,    snake_case__ :      str=0.1							,    snake_case__ :      Union[str, Any]=None    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			[]
     for model_output in model_outputs:
      SCREAMING_SNAKE_CASE								=			model_output['candidate_label']
      SCREAMING_SNAKE_CASE								=			BaseModelOutput(snake_case__    )
      SCREAMING_SNAKE_CASE								=			self.image_processor.post_process_object_detection(
          outputs=snake_case__							,    threshold=snake_case__							,    target_sizes=model_output['target_size']    )[0]

      for index in outputs["scores"].nonzero():
       SCREAMING_SNAKE_CASE								=			outputs['scores'][index].item()
       SCREAMING_SNAKE_CASE								=			self._get_bounding_box(outputs['boxes'][index][0]    )

       SCREAMING_SNAKE_CASE								=			{'score': score, 'label': label, 'box': box}
       results.append(snake_case__    )

     SCREAMING_SNAKE_CASE								=			sorted(snake_case__							,    key=lambda snake_case__    : x["score"]							,    reverse=snake_case__    )
     if top_k:
      SCREAMING_SNAKE_CASE								=			results[:top_k]

     return results




    def UpperCamelCase			(			self :      List[Any]							,    snake_case__ :      "torch.Tensor"    ):

     """simple docstring"""






     if self.framework != "pt":
      raise ValueError('The ZeroShotObjectDetectionPipeline is only available in PyTorch.'    )
     SCREAMING_SNAKE_CASE						,	SCREAMING_SNAKE_CASE						,	SCREAMING_SNAKE_CASE						,	SCREAMING_SNAKE_CASE								=			box.int().tolist()
     SCREAMING_SNAKE_CASE								=			{
         'xmin': xmin,
         'ymin': ymin,
         'xmax': xmax,
         'ymax': ymax,
     }
     return bbox

