

'''simple docstring'''

import warnings
from typing import Dict

import numpy as np

from ..utils import ExplicitEnum, add_end_docstrings, is_tf_available, is_torch_available
from .base import PIPELINE_INIT_ARGS, GenericTensor, Pipeline


if is_tf_available():
       from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING

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

def 					__UpperCamelCase				(						lowercase__    :	Tuple			):






     '''simple docstring'''





     return 1.0 / (1.0 + np.exp(-_outputs			))

def 					__UpperCamelCase				(						lowercase__    :	Optional[Any]			):






     '''simple docstring'''





     __lowercase      		=np.max(_outputs,       axis=-1,       keepdims=__snake_case			)
     __lowercase      		=np.exp(_outputs - maxes			)
     return shifted_exp / shifted_exp.sum(axis=-1,       keepdims=__snake_case			)


class 	lowerCAmelCase  (   a_ ):
       lowerCAmelCase_												=			"sigmoid"
       lowerCAmelCase_												=			"softmax"
       lowerCAmelCase_												=			"none"






@add_end_docstrings(
    a_ ,      r"\n        return_all_scores (`bool`, *optional*, defaults to `False`):\n            Whether to return all prediction scores or just the one of the predicted class.\n        function_to_apply (`str`, *optional*, defaults to `\"default\"`):\n            The function to apply to the model outputs in order to retrieve the scores. Accepts four different values:\n\n            - `\"default\"`: if the model has a single label, will apply the sigmoid function on the output. If the model\n              has several labels, will apply the softmax function on the output.\n            - `\"sigmoid\"`: Applies the sigmoid function on the output.\n            - `\"softmax\"`: Applies the softmax function on the output.\n            - `\"none\"`: Does not apply any function on the output.\n    " ,      )
class 	lowerCAmelCase  (   a_ ):
       lowerCAmelCase_												=			False
       lowerCAmelCase_												=			ClassificationFunction.NONE







       def __init__(				self      :			Dict    ,   **__lowercase      :			Dict   ):







            """simple docstring"""







            super().__init__(**_a   )

            self.check_model_type(
                TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING
                if self.framework == 'tf'
                else MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING   )







       def       snake_case (				self      :			Optional[int]    ,   __lowercase      :			List[str]=None    ,   __lowercase      :			Optional[Any]=None    ,   __lowercase      :			Optional[int]=""    ,   **__lowercase      :			Optional[Any]   ):







            """simple docstring"""







            __lowercase      		=tokenizer_kwargs

            __lowercase      		={}
            if hasattr(self.model.config    ,   'return_all_scores'   ) and return_all_scores is None:
                 __lowercase      		=self.model.config.return_all_scores

            if isinstance(_a    ,   _a   ) or top_k is None:
                 __lowercase      		=top_k
                 __lowercase      		=False
            elif return_all_scores is not None:
                 warnings.warn(
                     '`return_all_scores` is now deprecated,  if want a similar functionality use `top_k=None` instead of'
                     ' `return_all_scores=True` or `top_k=1` instead of `return_all_scores=False`.'    ,   _a    ,   )
                 if return_all_scores:
                      __lowercase      		=None
                 else:
                      __lowercase      		=1

            if isinstance(_a    ,   _a   ):
                 __lowercase      		=ClassificationFunction[function_to_apply.upper()]

            if function_to_apply is not None:
                 __lowercase      		=function_to_apply
            return preprocess_params, {}, postprocess_params







       def __call__(				self      :			Optional[int]    ,   *__lowercase      :			Tuple    ,   **__lowercase      :			Union[str, Any]   ):







            """simple docstring"""







            __lowercase      		=super().__call__(*_a    ,   **_a   )
            # TODO try and retrieve it in a nicer way from _sanitize_parameters.
            __lowercase      		="top_k" not in kwargs
            if isinstance(args[0]    ,   _a   ) and _legacy:
                 # This pipeline is odd, and return a list when single item is run
                 return [result]
            else:
                 return result







       def       snake_case (				self      :			Any    ,   __lowercase      :			Optional[Any]    ,   **__lowercase      :			List[Any]   ):







            """simple docstring"""







            __lowercase      		=self.framework
            if isinstance(_a    ,   _a   ):
                 return self.tokenizer(**_a    ,   return_tensors=_a    ,   **_a   )
            elif isinstance(_a    ,   _a   ) and len(_a   ) == 1 and isinstance(inputs[0]    ,   _a   ) and len(inputs[0]   ) == 2:
                 # It used to be valid to use a list of list of list for text pairs, keeping this path for BC
                 return self.tokenizer(
                     text=inputs[0][0]    ,   text_pair=inputs[0][1]    ,   return_tensors=_a    ,   **_a   )
            elif isinstance(_a    ,   _a   ):
                 # This is likely an invalid usage of the pipeline attempting to pass text pairs.
                 raise ValueError(
                     'The pipeline received invalid inputs, if you are trying to send text pairs, you can try to send a'
                     ' dictionary `{\"text\": \"My text\", \"text_pair\": \"My pair\"}` in order to send a text pair.'   )
            return self.tokenizer(_a    ,   return_tensors=_a    ,   **_a   )







       def       snake_case (				self      :			Union[str, Any]    ,   __lowercase      :			Tuple   ):







            """simple docstring"""







            return self.model(**_a   )







       def       snake_case (				self      :			Dict    ,   __lowercase      :			Tuple    ,   __lowercase      :			List[Any]=None    ,   __lowercase      :			Dict=1    ,   __lowercase      :			List[str]=True   ):







            """simple docstring"""







            if function_to_apply is None:
                 if self.model.config.problem_type == "multi_label_classification" or self.model.config.num_labels == 1:
                      __lowercase      		=ClassificationFunction.SIGMOID
                 elif self.model.config.problem_type == "single_label_classification" or self.model.config.num_labels > 1:
                      __lowercase      		=ClassificationFunction.SOFTMAX
                 elif hasattr(self.model.config    ,   'function_to_apply'   ) and function_to_apply is None:
                      __lowercase      		=self.model.config.function_to_apply
                 else:
                      __lowercase      		=ClassificationFunction.NONE

            __lowercase      		=model_outputs["logits"][0]
            __lowercase      		=outputs.numpy()

            if function_to_apply == ClassificationFunction.SIGMOID:
                 __lowercase      		=sigmoid(_a   )
            elif function_to_apply == ClassificationFunction.SOFTMAX:
                 __lowercase      		=softmax(_a   )
            elif function_to_apply == ClassificationFunction.NONE:
                 __lowercase      		=outputs
            else:
                 raise ValueError(f'''Unrecognized `function_to_apply` argument: {function_to_apply}'''   )

            if top_k == 1 and _legacy:
                 return {"label": self.model.config.idalabel[scores.argmax().item()], "score": scores.max().item()}

            __lowercase      		=[
                {"label": self.model.config.idalabel[i], "score": score.item()} for i, score in enumerate(_a   )
            ]
            if not _legacy:
                 dict_scores.sort(key=lambda __lowercase   : x["score"]    ,   reverse=_a   )
                 if top_k is not None:
                      __lowercase      		=dict_scores[:top_k]
            return dict_scores




"""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     :			List[str]  						=			get_logger(__name__)

class 						a__		:
   __lowerCAmelCase				=							"""dummy_data"""
   __lowerCAmelCase				=							"""datasets"""
   __lowerCAmelCase				=							False
   def __init__(     self      ,       _a      ,       _a      ,       _a      ,       _a = None      ,       _a = False      ,       _a = True      ,       _a = None      ,       ):
          lowercase       :							int									=				0
          lowercase       :							Optional[Any]									=				dataset_name
          lowercase       :							List[str]									=				cache_dir
          lowercase       :							Union[str, Any]									=				use_local_dummy_data
          lowercase       :							str									=				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       :							List[Any]									=				load_existing_dummy_data

          # TODO(PVP, QL) might need to make this more general
          lowercase       :							Tuple									=				str(_a							)
          # to be downloaded
          lowercase       :							Tuple									=				None
          lowercase       :							List[Any]									=				None
   @property
   def      __magic_name__     (     self							):
          if self._dummy_file is None:
                 lowercase       :							Optional[int]									=				self.download_dummy_data()
          return self._dummy_file
   @property
   def      __magic_name__     (     self							):
          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      __magic_name__     (     self							):
          return os.path.join(self.dummy_data_folder      ,       "dummy_data.zip"							)
   def      __magic_name__     (     self							):
          lowercase       :							Optional[Any]									=				(
              self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
          )

          lowercase       :							str									=				cached_path(
              _a      ,       cache_dir=self.cache_dir      ,       extract_compressed_file=_a      ,       force_extract=_a							)

          return os.path.join(_a      ,       self.dummy_file_name							)
   @property
   def      __magic_name__     (     self							):
          return os.path.join(self.datasets_scripts_dir      ,       self.dataset_name      ,       self.dummy_zip_file							)
   @property
   def      __magic_name__     (     self							):
          if self._bucket_url is None:
                 lowercase       :							Dict									=				hf_github_url(self.dataset_name      ,       self.dummy_zip_file.replace(os.sep      ,       "/"							)							)
          return self._bucket_url
   @property
   def      __magic_name__     (     self							):
          # 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      __magic_name__     (     self      ,       _a      ,       *_a							):
          if self.load_existing_dummy_data:
                 # dummy data is downloaded and tested
                 lowercase       :							Optional[int]									=				self.dummy_file
          else:
                 # dummy data cannot be downloaded and only the path to dummy file is returned
                 lowercase       :							Optional[Any]									=				self.dummy_file_name

          # special case when data_url is a dict
          if isinstance(_a      ,       _a							):
                 return self.create_dummy_data_dict(_a      ,       _a							)
          elif isinstance(_a      ,       (list, tuple)							):
                 return self.create_dummy_data_list(_a      ,       _a							)
          else:
                 return self.create_dummy_data_single(_a      ,       _a							)
   def      __magic_name__     (     self      ,       _a      ,       *_a							):
          return self.download_and_extract(_a							)
   def      __magic_name__     (     self      ,       _a      ,       _a							):
          return self.download_and_extract(_a							)
   def      __magic_name__     (     self      ,       _a      ,       *_a      ,       **_a							):
          return path
   def      __magic_name__     (     self							):
          return {}
   def      __magic_name__     (     self      ,       _a      ,       _a							):
          lowercase       :							List[str]									=				{}
          for key, single_urls in data_url.items():
                 for download_callback in self.download_callbacks:
                        if isinstance(_a      ,       _a							):
                               for single_url in single_urls:
                                      download_callback(_a							)
                        else:
                               lowercase       :							Union[str, Any]									=				single_urls
                               download_callback(_a							)
            # 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(_a      ,       _a							):
                        lowercase       :							Any									=				[os.path.join(_a      ,       urllib.parse.quote_plus(Path(_a							).name							)							) for x in single_urls]
                 else:
                        lowercase       :							int									=				single_urls
                        lowercase       :							Tuple									=				os.path.join(_a      ,       urllib.parse.quote_plus(Path(_a							).name							)							)
                 lowercase       :							List[str]									=				value

          # make sure that values are unique
          if all(isinstance(_a      ,       _a							) 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       :							str									=				{key: value + key for key, value in dummy_data_dict.items()}

          return dummy_data_dict
   def      __magic_name__     (     self      ,       _a      ,       _a							):
          lowercase       :							Union[str, Any]									=				[]
          # trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
          lowercase       :							Any									=				all(bool(re.findall("[0-9]{3,}-of-[0-9]{3,}"      ,       _a							)							) for url in data_url							)
          lowercase       :							List[Any]									=				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       :							Tuple									=				[data_url[0]] * len(_a							)
          for single_url in data_url:
                 for download_callback in self.download_callbacks:
                        download_callback(_a							)
                 # 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(_a      ,       urllib.parse.quote_plus(single_url.split("/"							)[-1]							)							)
                 dummy_data_list.append(_a							)
          return dummy_data_list
   def      __magic_name__     (     self      ,       _a      ,       _a							):
          for download_callback in self.download_callbacks:
                 download_callback(_a							)
          # 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       :							Tuple									=				os.path.join(_a      ,       urllib.parse.quote_plus(data_url.split("/"							)[-1]							)							)
          if os.path.exists(_a							) 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      __magic_name__     (     self							):
          pass
   def      __magic_name__     (     self							):
          pass
   def      __magic_name__     (     self      ,       _a							):
          def _iter_archive_members(_a							):
                 # this preserves the order of the members inside the ZIP archive
                 lowercase       :							Optional[int]									=				Path(self.dummy_file							).parent
                 lowercase       :							List[str]									=				path.relative_to(_a							)
                 with ZipFile(self.local_path_to_dummy_data							) as zip_file:
                        lowercase       :							List[str]									=				zip_file.namelist()
                 for member in members:
                        if member.startswith(relative_path.as_posix()							):
                               yield dummy_parent_path.joinpath(_a							)

          lowercase       :							Union[str, Any]									=				Path(_a							)
          lowercase       :							List[Any]									=				_iter_archive_members(_a							) 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(_a							).as_posix(), file_path.open("rb"							)

   def      __magic_name__     (     self      ,       _a							):
          if not isinstance(_a      ,       _a							):
                 lowercase       :							Any									=				[paths]
          for path in paths:
                 if os.path.isfile(_a							):
                        if os.path.basename(_a							).startswith((".", "__")							):
                               return
                        yield path
                 else:
                        for dirpath, dirnames, filenames in os.walk(_a							):
                               if os.path.basename(_a							).startswith((".", "__")							):
                                      continue
                               dirnames.sort()
                               for filename in sorted(_a							):
                                      if filename.startswith((".", "__")							):
                                             continue
                                      yield os.path.join(_a      ,       _a							)


