




import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


UpperCamelCase__					    =    logging.get_logger(__name__)
class     a__					:
						_a							:					int      =					None







@experimental
def 			_a   (		SCREAMING_SNAKE_CASE_							:       List[Any]   ,   SCREAMING_SNAKE_CASE_							:       Tuple   ,   SCREAMING_SNAKE_CASE_							:       Dict   ,   SCREAMING_SNAKE_CASE_							:       List[str]   ,   SCREAMING_SNAKE_CASE_							:       Dict   ,   SCREAMING_SNAKE_CASE_							:       Any   ,   SCREAMING_SNAKE_CASE_							:       Optional[Any]  ):
					if ParallelBackendConfig.backend_name is None:
										return _map_with_multiprocessing_pool(
										    SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_  )

					return _map_with_joblib(SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_  )







def 			_a   (		SCREAMING_SNAKE_CASE_							:       Optional[int]   ,   SCREAMING_SNAKE_CASE_							:       int   ,   SCREAMING_SNAKE_CASE_							:       Optional[Any]   ,   SCREAMING_SNAKE_CASE_							:       List[Any]   ,   SCREAMING_SNAKE_CASE_							:       str   ,   SCREAMING_SNAKE_CASE_							:       List[str]   ,   SCREAMING_SNAKE_CASE_							:       List[str]  ):
					__lowerCAmelCase            = num_proc if num_proc <= len(SCREAMING_SNAKE_CASE_  ) else len(SCREAMING_SNAKE_CASE_  )
					__lowerCAmelCase            = []  # We organize the splits ourselve (contiguous splits)
					for index in range(SCREAMING_SNAKE_CASE_  ):
										__lowerCAmelCase            = len(SCREAMING_SNAKE_CASE_  ) // num_proc
										__lowerCAmelCase            = len(SCREAMING_SNAKE_CASE_  ) % num_proc
										__lowerCAmelCase            = div * index + min(SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_  )
										__lowerCAmelCase            = start + div + (1 if index < mod else 0)
										split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)  )

					if len(SCREAMING_SNAKE_CASE_  ) != sum(len(i[1]  ) for i in split_kwds  ):
										raise ValueError(
										    F"""Error dividing inputs iterable among processes. """
										    F"""Total number of objects {len(SCREAMING_SNAKE_CASE_  )}, """
										    F"""length: {sum(len(i[1]  ) for i in split_kwds  )}"""  )

					logger.info(
					    F"""Spawning {num_proc} processes for {len(SCREAMING_SNAKE_CASE_  )} objects in slices of {[len(i[1]  ) for i in split_kwds]}"""  )
					__lowerCAmelCase     ,							__lowerCAmelCase            = None, None
					if not disable_tqdm:
										__lowerCAmelCase     ,							__lowerCAmelCase            = (RLock(),), tqdm.set_lock
					with Pool(SCREAMING_SNAKE_CASE_   ,   initargs=SCREAMING_SNAKE_CASE_   ,   initializer=SCREAMING_SNAKE_CASE_  ) as pool:
										__lowerCAmelCase            = pool.map(SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_  )
					logger.info(F"""Finished {num_proc} processes"""  )
					__lowerCAmelCase            = [obj for proc_res in mapped for obj in proc_res]
					logger.info(F"""Unpacked {len(SCREAMING_SNAKE_CASE_  )} objects"""  )

					return mapped







def 			_a   (		SCREAMING_SNAKE_CASE_							:       Optional[int]   ,   SCREAMING_SNAKE_CASE_							:       Dict   ,   SCREAMING_SNAKE_CASE_							:       Union[str, Any]   ,   SCREAMING_SNAKE_CASE_							:       Any   ,   SCREAMING_SNAKE_CASE_							:       int   ,   SCREAMING_SNAKE_CASE_							:       Dict   ,   SCREAMING_SNAKE_CASE_							:       Dict  ):
					# progress bar is not yet supported for _map_with_joblib, because tqdm couldn't accurately be applied to joblib,
					# and it requires monkey-patching joblib internal classes which is subject to change
					import joblib

					with joblib.parallel_backend(ParallelBackendConfig.backend_name   ,   n_jobs=SCREAMING_SNAKE_CASE_  ):
										return joblib.Parallel()(
										    joblib.delayed(SCREAMING_SNAKE_CASE_  )((function, obj, types, None, True, None)  ) for obj in iterable  )







@experimental
@contextlib.contextmanager
def 			_a   (		SCREAMING_SNAKE_CASE_							:       str  ):
					__lowerCAmelCase            = backend_name

					if backend_name == "spark":
										from joblibspark import register_spark

										register_spark()

										# TODO: call create_cache_and_write_probe if "download" in steps
										# TODO: raise NotImplementedError when Dataset.map etc is called

					try:
										yield
					finally:
										__lowerCAmelCase            = None










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

UpperCamelCase__					    =    logging.get_logger(__name__)
@add_end_docstrings(snake_case__   )
class     a__					(      snake_case__   ):






						def __init__(     self	,    *_A	,    **_A    ):




											"""simple docstring"""



											super().__init__(*_A	,    **_A    )
											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        __SCREAMING_SNAKE_CASE(     self	,    _A=None    ):




											"""simple docstring"""



											__lowerCAmelCase            = {}
											if top_k is not None:
																__lowerCAmelCase            = top_k
											return {}, {}, postprocess_params






						def __call__(     self	,    _A	,    **_A    ):




											"""simple docstring"""



											return super().__call__(_A	,    **_A    )






						def        __SCREAMING_SNAKE_CASE(     self	,    _A    ):




											"""simple docstring"""



											__lowerCAmelCase            = load_image(_A    )
											__lowerCAmelCase            = self.image_processor(images=_A	,    return_tensors=self.framework    )
											return model_inputs






						def        __SCREAMING_SNAKE_CASE(     self	,    _A    ):




											"""simple docstring"""



											__lowerCAmelCase            = self.model(**_A    )
											return model_outputs






						def        __SCREAMING_SNAKE_CASE(     self	,    _A	,    _A=5    ):




											"""simple docstring"""



											if top_k > self.model.config.num_labels:
																__lowerCAmelCase            = self.model.config.num_labels

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

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




