
"""simple docstring"""



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




@add_end_docstrings(_UpperCAmelCase						)
class 		a_    (			_UpperCAmelCase						):







    def __init__(      self      :							Optional[Any]  ,       **__UpperCamelCase      :							Optional[Any]  )		->Tuple:






        '''simple docstring'''






        super().__init__(**__UpperCamelCase  )

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

        requires_backends(self  ,       """vision"""  )
        self.check_model_type(__UpperCamelCase  )







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






        '''simple docstring'''






        if "text_queries" in kwargs:
            _UpperCAmelCase        = kwargs.pop("""text_queries"""  )

        if isinstance(__UpperCamelCase  ,       (str, Image.Image)  ):
            _UpperCAmelCase        = {"""image""": image, """candidate_labels""": candidate_labels}
        else:
            _UpperCAmelCase        = image
        _UpperCAmelCase        = super().__call__(__UpperCamelCase  ,       **__UpperCamelCase  )
        return results







    def   _snake_case       (      self      :							int  ,       **__UpperCamelCase      :							int  )		->str:






        '''simple docstring'''






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







    def   _snake_case       (      self      :							Dict  ,       __UpperCamelCase      :							Optional[int]  )		->Optional[Any]:






        '''simple docstring'''






        _UpperCAmelCase        = load_image(inputs["""image"""]  )
        _UpperCAmelCase        = inputs["""candidate_labels"""]
        if isinstance(__UpperCamelCase  ,       __UpperCamelCase  ):
            _UpperCAmelCase        = candidate_labels.split(""","""  )

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







    def   _snake_case       (      self      :							List[str]  ,       __UpperCamelCase      :							Optional[int]  )		->List[Any]:






        '''simple docstring'''






        _UpperCAmelCase        = model_inputs.pop("""target_size"""  )
        _UpperCAmelCase        = model_inputs.pop("""candidate_label"""  )
        _UpperCAmelCase        = model_inputs.pop("""is_last"""  )

        _UpperCAmelCase        = self.model(**__UpperCamelCase  )

        _UpperCAmelCase        = {"""target_size""": target_size, """candidate_label""": candidate_label, """is_last""": is_last, **outputs}
        return model_outputs







    def   _snake_case       (      self      :							Optional[int]  ,       __UpperCamelCase      :							Dict  ,       __UpperCamelCase      :							Dict=0.1  ,       __UpperCamelCase      :							str=None  )		->Union[str, Any]:






        '''simple docstring'''






        _UpperCAmelCase        = []
        for model_output in model_outputs:
            _UpperCAmelCase        = model_output["""candidate_label"""]
            _UpperCAmelCase        = BaseModelOutput(__UpperCamelCase  )
            _UpperCAmelCase        = self.image_processor.post_process_object_detection(
                outputs=__UpperCamelCase  ,       threshold=__UpperCamelCase  ,       target_sizes=model_output["""target_size"""]  )[0]

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

                _UpperCAmelCase        = {"""score""": score, """label""": label, """box""": box}
                results.append(__UpperCamelCase  )

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

        return results







    def   _snake_case       (      self      :							Dict  ,       __UpperCamelCase      :							"torch.Tensor"  )		->Dict[str, int]:






        '''simple docstring'''






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

"""simple docstring"""



import os
import warnings
from typing import List, Optional

from ...tokenization_utils_base import BatchEncoding
from ...utils import logging
from .configuration_rag import RagConfig


a       :						List[str]       		=		logging.get_logger(__name__)




class 		a_    :







    def __init__(      self      :							List[str]  ,       __UpperCamelCase      :							Any  ,       __UpperCamelCase      :							Tuple  )		->Tuple:






        '''simple docstring'''






        _UpperCAmelCase        = question_encoder
        _UpperCAmelCase        = generator
        _UpperCAmelCase        = self.question_encoder







    def   _snake_case       (      self      :							int  ,       __UpperCamelCase      :							str  )		->Dict:






        '''simple docstring'''






        if os.path.isfile(__UpperCamelCase  ):
            raise ValueError(f"""Provided path ({save_directory}) should be a directory, not a file"""  )
        os.makedirs(__UpperCamelCase  ,       exist_ok=__UpperCamelCase  )
        _UpperCAmelCase        = os.path.join(__UpperCamelCase  ,       """question_encoder_tokenizer"""  )
        _UpperCAmelCase        = os.path.join(__UpperCamelCase  ,       """generator_tokenizer"""  )
        self.question_encoder.save_pretrained(__UpperCamelCase  )
        self.generator.save_pretrained(__UpperCamelCase  )







    @classmethod
    def   _snake_case       (      cls      :							Optional[Any]  ,       __UpperCamelCase      :							Tuple  ,       **__UpperCamelCase      :							Optional[Any]  )		->Tuple:






        '''simple docstring'''






        from ..auto.tokenization_auto import AutoTokenizer

        _UpperCAmelCase        = kwargs.pop("""config"""  ,       __UpperCamelCase  )

        if config is None:
            _UpperCAmelCase        = RagConfig.from_pretrained(__UpperCamelCase  )

        _UpperCAmelCase        = AutoTokenizer.from_pretrained(
            __UpperCamelCase  ,       config=config.question_encoder  ,       subfolder="""question_encoder_tokenizer"""  )
        _UpperCAmelCase        = AutoTokenizer.from_pretrained(
            __UpperCamelCase  ,       config=config.generator  ,       subfolder="""generator_tokenizer"""  )
        return cls(question_encoder=__UpperCamelCase  ,       generator=__UpperCamelCase  )







    def __call__(      self      :							Any  ,       *__UpperCamelCase      :							List[Any]  ,       **__UpperCamelCase      :							Optional[Any]  )		->Optional[Any]:






        '''simple docstring'''






        return self.current_tokenizer(*__UpperCamelCase  ,       **__UpperCamelCase  )







    def   _snake_case       (      self      :							Union[str, Any]  ,       *__UpperCamelCase      :							List[Any]  ,       **__UpperCamelCase      :							Tuple  )		->Any:






        '''simple docstring'''






        return self.generator.batch_decode(*__UpperCamelCase  ,       **__UpperCamelCase  )







    def   _snake_case       (      self      :							Optional[int]  ,       *__UpperCamelCase      :							Optional[int]  ,       **__UpperCamelCase      :							List[str]  )		->Union[str, Any]:






        '''simple docstring'''






        return self.generator.decode(*__UpperCamelCase  ,       **__UpperCamelCase  )







    def   _snake_case       (      self      :							Union[str, Any]  )		->Dict:






        '''simple docstring'''






        _UpperCAmelCase        = self.question_encoder







    def   _snake_case       (      self      :							Tuple  )		->Optional[int]:






        '''simple docstring'''






        _UpperCAmelCase        = self.generator







    def   _snake_case       (      self      :							List[Any]  ,       __UpperCamelCase      :							List[str]  ,       __UpperCamelCase      :							Optional[List[str]] = None  ,       __UpperCamelCase      :							Optional[int] = None  ,       __UpperCamelCase      :							Optional[int] = None  ,       __UpperCamelCase      :							str = "longest"  ,       __UpperCamelCase      :							str = None  ,       __UpperCamelCase      :							bool = True  ,       **__UpperCamelCase      :							Tuple  ,       )		->BatchEncoding:






        '''simple docstring'''






        warnings.warn(
            """`prepare_seq2seq_batch` is deprecated and will be removed in version 5 of 🤗 Transformers. Use the """
            """regular `__call__` method to prepare your inputs and the tokenizer under the `with_target_tokenizer` """
            """context manager to prepare your targets. See the documentation of your specific tokenizer for more """
            """details"""  ,       __UpperCamelCase  ,       )
        if max_length is None:
            _UpperCAmelCase        = self.current_tokenizer.model_max_length
        _UpperCAmelCase        = self(
            __UpperCamelCase  ,       add_special_tokens=__UpperCamelCase  ,       return_tensors=__UpperCamelCase  ,       max_length=__UpperCamelCase  ,       padding=__UpperCamelCase  ,       truncation=__UpperCamelCase  ,       **__UpperCamelCase  ,       )
        if tgt_texts is None:
            return model_inputs
        # Process tgt_texts
        if max_target_length is None:
            _UpperCAmelCase        = self.current_tokenizer.model_max_length
        _UpperCAmelCase        = self(
            text_target=__UpperCamelCase  ,       add_special_tokens=__UpperCamelCase  ,       return_tensors=__UpperCamelCase  ,       padding=__UpperCamelCase  ,       max_length=__UpperCamelCase  ,       truncation=__UpperCamelCase  ,       **__UpperCamelCase  ,       )
        _UpperCAmelCase        = labels["""input_ids"""]
        return model_inputs