





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

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


@add_end_docstrings(lowerCamelCase__				)
class        __magic_name__	(				lowerCamelCase__				):





     """simple docstring"""



     def __init__(				self							:str      ,   **snake_case							:Optional[Any]	):





        '''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							:Tuple      ,   snake_case							:Union[str, "Image.Image", List[Dict[str, Any]]]      ,   snake_case							:Union[str, List[str]] = None      ,   **snake_case							:Tuple      ,   ):





        '''simple docstring'''




        if "text_queries" in kwargs:
           A_       :					Any        =      kwargs.pop("text_queries"	)

        if isinstance(snake_case      ,   (str, Image.Image)	):
           A_       :					Tuple        =      {"image": image, "candidate_labels": candidate_labels}
        else:
           A_       :					Dict        =      image
        A_       :					List[Any]        =      super().__call__(snake_case      ,   **snake_case	)
        return results



     def     SCREAMING_SNAKE_CASE		(				self							:Dict      ,   **snake_case							:Optional[Any]	):





        '''simple docstring'''




        A_       :					int        =      {}
        if "threshold" in kwargs:
           A_       :					Union[str, Any]        =      kwargs["threshold"]
        if "top_k" in kwargs:
           A_       :					List[str]        =      kwargs["top_k"]
        return {}, {}, postprocess_params



     def     SCREAMING_SNAKE_CASE		(				self							:str      ,   snake_case							:Optional[int]	):





        '''simple docstring'''




        A_       :					Tuple        =      load_image(inputs["image"]	)
        A_       :					Optional[int]        =      inputs["candidate_labels"]
        if isinstance(snake_case      ,   snake_case	):
           A_       :					List[Any]        =      candidate_labels.split(","	)

        A_       :					str        =      torch.tensor([[image.height, image.width]]      ,   dtype=torch.intaa	)
        for i, candidate_label in enumerate(snake_case	):
           A_       :					List[str]        =      self.tokenizer(snake_case      ,   return_tensors=self.framework	)
           A_       :					Union[str, Any]        =      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     SCREAMING_SNAKE_CASE		(				self							:Dict      ,   snake_case							:Optional[int]	):





        '''simple docstring'''




        A_       :					Tuple        =      model_inputs.pop("target_size"	)
        A_       :					Optional[Any]        =      model_inputs.pop("candidate_label"	)
        A_       :					List[str]        =      model_inputs.pop("is_last"	)

        A_       :					Optional[int]        =      self.model(**snake_case	)

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



     def     SCREAMING_SNAKE_CASE		(				self							:List[Any]      ,   snake_case							:Dict      ,   snake_case							:List[Any]=0.1      ,   snake_case							:Dict=None	):





        '''simple docstring'''




        A_       :					Optional[int]        =      []
        for model_output in model_outputs:
           A_       :					Any        =      model_output["candidate_label"]
           A_       :					Optional[int]        =      BaseModelOutput(snake_case	)
           A_       :					Optional[int]        =      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():
              A_       :					Optional[int]        =      outputs["scores"][index].item()
              A_       :					List[str]        =      self._get_bounding_box(outputs["boxes"][index][0]	)

              A_       :					Dict        =      {"score": score, "label": label, "box": box}
              results.append(snake_case	)

        A_       :					Dict        =      sorted(snake_case      ,   key=lambda snake_case	: x["score"]      ,   reverse=snake_case	)
        if top_k:
           A_       :					List[str]        =      results[:top_k]

        return results





     def     SCREAMING_SNAKE_CASE		(				self							:List[str]      ,   snake_case							:"torch.Tensor"	):





        '''simple docstring'''




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







import argparse

import intel_extension_for_pytorch as ipex
import torch

from diffusers import DPMSolverMultistepScheduler, StableDiffusionPipeline


_lowerCAmelCase				:   Any						       =						argparse.ArgumentParser('''Stable Diffusion script with intel optimization''', add_help=False)
parser.add_argument('''--dpm''', action='''store_true''', help='''Enable DPMSolver or not''')
parser.add_argument('''--steps''', default=None, type=int, help='''Num inference steps''')
_lowerCAmelCase				:   Tuple						       =						parser.parse_args()


_lowerCAmelCase				:   Any						       =						'''cpu'''
_lowerCAmelCase				:   Dict						       =						'''a lovely <dicoo> in red dress and hat, in the snowly and brightly night, with many brighly buildings'''

_lowerCAmelCase				:   Union[str, Any]						       =						'''path-to-your-trained-model'''
_lowerCAmelCase				:   Tuple						       =						StableDiffusionPipeline.from_pretrained(model_id)
if args.dpm:
 _lowerCAmelCase				:   List[str]						       =						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
_lowerCAmelCase				:   List[Any]						       =						pipe.to(device)

# to channels last
_lowerCAmelCase				:   Any						       =						pipe.unet.to(memory_format=torch.channels_last)
_lowerCAmelCase				:   int						       =						pipe.vae.to(memory_format=torch.channels_last)
_lowerCAmelCase				:   Optional[Any]						       =						pipe.text_encoder.to(memory_format=torch.channels_last)
if pipe.requires_safety_checker:
 _lowerCAmelCase				:   Optional[int]						       =						pipe.safety_checker.to(memory_format=torch.channels_last)

# optimize with ipex
_lowerCAmelCase				:   Tuple						       =						torch.randn(2, 4, 64, 64)
_lowerCAmelCase				:   List[str]						       =						torch.rand(1) * 999
_lowerCAmelCase				:   List[Any]						       =						torch.randn(2, 77, 768)
_lowerCAmelCase				:   int						       =						(sample, timestep, encoder_hidden_status)
try:
 _lowerCAmelCase				:   Tuple						       =						ipex.optimize(pipe.unet.eval(), dtype=torch.bfloataa, inplace=True, sample_input=input_example)
except Exception:
 _lowerCAmelCase				:   Optional[Any]						       =						ipex.optimize(pipe.unet.eval(), dtype=torch.bfloataa, inplace=True)
_lowerCAmelCase				:   Dict						       =						ipex.optimize(pipe.vae.eval(), dtype=torch.bfloataa, inplace=True)
_lowerCAmelCase				:   Dict						       =						ipex.optimize(pipe.text_encoder.eval(), dtype=torch.bfloataa, inplace=True)
if pipe.requires_safety_checker:
 _lowerCAmelCase				:   Tuple						       =						ipex.optimize(pipe.safety_checker.eval(), dtype=torch.bfloataa, inplace=True)

# compute
_lowerCAmelCase				:   Dict						       =						666
_lowerCAmelCase				:   Tuple						       =						torch.Generator(device).manual_seed(seed)
_lowerCAmelCase				:   Tuple						       =						{'''generator''': generator}
if args.steps is not None:
 _lowerCAmelCase				:   Tuple						       =						args.steps

with torch.cpu.amp.autocast(enabled=True, dtype=torch.bfloataa):
 _lowerCAmelCase				:   int						       =						pipe(prompt, **generate_kwargs).images[0]

# save image
image.save('''generated.png''')
