





from typing import Callable, List, Optional, Union

import PIL
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPSegForImageSegmentation,
    CLIPSegProcessor,
    CLIPTextModel,
    CLIPTokenizer,
)

from diffusers import DiffusionPipeline
from diffusers.configuration_utils import FrozenDict
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionInpaintPipeline
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import deprecate, is_accelerate_available, logging


a_    :						List[Any]    		=      logging.get_logger(__name__)  # pylint: disable=invalid-name




class      UpperCamelCase (     SCREAMING_SNAKE_CASE   ):

    def __init__(			self :      int							,    snake_case__ :      Union[str, Any]							,    snake_case__ :      int							,    snake_case__ :      Any							,    snake_case__ :      int							,    snake_case__ :      List[Any]							,    snake_case__ :      Tuple							,    snake_case__ :      str							,    snake_case__ :      int							,    snake_case__ :      int							,    ):

     """simple docstring"""






     super().__init__()

     if hasattr(scheduler.config							,    'steps_offset'    ) and scheduler.config.steps_offset != 1:
      SCREAMING_SNAKE_CASE								=			(
          F"""The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"""
          F""" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure """
          'to update the config accordingly as leaving `steps_offset` might led to incorrect results'
          ' in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,'
          ' it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`'
          ' file'
      )
      deprecate('steps_offset!=1'							,    '1.0.0'							,    __a							,    standard_warn=__a    )
      SCREAMING_SNAKE_CASE								=			dict(scheduler.config    )
      SCREAMING_SNAKE_CASE								=			1
      SCREAMING_SNAKE_CASE								=			FrozenDict(__a    )

     if hasattr(scheduler.config							,    'skip_prk_steps'    ) and scheduler.config.skip_prk_steps is False:
      SCREAMING_SNAKE_CASE								=			(
          F"""The configuration file of this scheduler: {scheduler} has not set the configuration"""
          ' `skip_prk_steps`. `skip_prk_steps` should be set to True in the configuration file. Please make'
          ' sure to update the config accordingly as not setting `skip_prk_steps` in the config might lead to'
          ' incorrect results in future versions. If you have downloaded this checkpoint from the Hugging Face'
          ' Hub, it would be very nice if you could open a Pull request for the'
          ' `scheduler/scheduler_config.json` file'
      )
      deprecate('skip_prk_steps not set'							,    '1.0.0'							,    __a							,    standard_warn=__a    )
      SCREAMING_SNAKE_CASE								=			dict(scheduler.config    )
      SCREAMING_SNAKE_CASE								=			True
      SCREAMING_SNAKE_CASE								=			FrozenDict(__a    )

     if safety_checker is None:
      logger.warning(
          F"""You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"""
          ' that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered'
          ' results in services or applications open to the public. Both the diffusers team and Hugging Face'
          ' strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling'
          ' it only for use-cases that involve analyzing network behavior or auditing its results. For more'
          ' information, please have a look at https://github.com/huggingface/diffusers/pull/254 .'    )

     self.register_modules(
         segmentation_model=__a							,    segmentation_processor=__a							,    vae=__a							,    text_encoder=__a							,    tokenizer=__a							,    unet=__a							,    scheduler=__a							,    safety_checker=__a							,    feature_extractor=__a							,    )

    def UpperCamelCase			(			self :      int							,    snake_case__ :      Optional[Any] = "auto"    ):

     """simple docstring"""






     if slice_size == "auto":
      # half the attention head size is usually a good trade-off between
      # speed and memory
      SCREAMING_SNAKE_CASE								=			self.unet.config.attention_head_dim // 2
     self.unet.set_attention_slice(__a    )

    def UpperCamelCase			(			self :      List[Any]    ):

     """simple docstring"""






     self.enable_attention_slicing(__a    )

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

     """simple docstring"""






     if is_accelerate_available():
      from accelerate import cpu_offload
     else:
      raise ImportError('Please install accelerate via `pip install accelerate`'    )

     SCREAMING_SNAKE_CASE								=			torch.device('cuda'    )

     for cpu_offloaded_model in [self.unet, self.text_encoder, self.vae, self.safety_checker]:
      if cpu_offloaded_model is not None:
       cpu_offload(__a							,    __a    )

    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def UpperCamelCase			(			self :      Dict    ):

     """simple docstring"""






     if self.device != torch.device('meta'    ) or not hasattr(self.unet							,    '_hf_hook'    ):
      return self.device
     for module in self.unet.modules():
      if (
          hasattr(__a							,    '_hf_hook'    )
          and hasattr(module._hf_hook							,    'execution_device'    )
          and module._hf_hook.execution_device is not None
      ):
       return torch.device(module._hf_hook.execution_device    )
     return self.device




    @torch.no_grad()
    def __call__(			self :      Any							,    snake_case__ :      List[Any]							,    snake_case__ :      Tuple							,    snake_case__ :      Tuple							,    snake_case__ :      str = 5_1_2							,    snake_case__ :      Union[str, Any] = 5_1_2							,    snake_case__ :      List[str] = 5_0							,    snake_case__ :      List[str] = 7.5							,    snake_case__ :      List[str] = None							,    snake_case__ :      int = 1							,    snake_case__ :      Optional[int] = 0.0							,    snake_case__ :      Union[str, Any] = None							,    snake_case__ :      int = None							,    snake_case__ :      Any = "pil"							,    snake_case__ :      Dict = True							,    snake_case__ :      Union[str, Any] = None							,    snake_case__ :      Optional[Any] = 1							,    **snake_case__ :      str							,    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			self.segmentation_processor(
         text=[text]							,    images=[image]							,    padding='max_length'							,    return_tensors='pt'    ).to(self.device    )
     SCREAMING_SNAKE_CASE								=			self.segmentation_model(**__a    )
     SCREAMING_SNAKE_CASE								=			torch.sigmoid(outputs.logits    ).cpu().detach().unsqueeze(-1    ).numpy()
     SCREAMING_SNAKE_CASE								=			self.numpy_to_pil(__a    )[0].resize(image.size    )

     # Run inpainting pipeline with the generated mask
     SCREAMING_SNAKE_CASE								=			StableDiffusionInpaintPipeline(
         vae=self.vae							,    text_encoder=self.text_encoder							,    tokenizer=self.tokenizer							,    unet=self.unet							,    scheduler=self.scheduler							,    safety_checker=self.safety_checker							,    feature_extractor=self.feature_extractor							,    )
     return inpainting_pipeline(
         prompt=__a							,    image=__a							,    mask_image=__a							,    height=__a							,    width=__a							,    num_inference_steps=__a							,    guidance_scale=__a							,    negative_prompt=__a							,    num_images_per_prompt=__a							,    eta=__a							,    generator=__a							,    latents=__a							,    output_type=__a							,    return_dict=__a							,    callback=__a							,    callback_steps=__a							,    )




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

