




from typing import Any, Callable, Dict, List, Optional, Union

import PIL.Image
import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionImgaImgPipeline,
    StableDiffusionInpaintPipelineLegacy,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.configuration_utils import FrozenDict
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import deprecate, logging


UpperCAmelCase__						:       Tuple							=    logging.get_logger(__name__)  # pylint: disable=invalid-name

class      UpperCAmelCase						(			A   ):
   __SCREAMING_SNAKE_CASE		:							Dict							=					["safety_checker", "feature_extractor"]






   def __init__(	self   :       Union[str, Any]						,					UpperCamelCase_   :       AutoencoderKL						,					UpperCamelCase_   :       CLIPTextModel						,					UpperCamelCase_   :       CLIPTokenizer						,					UpperCamelCase_   :       UNetaDConditionModel						,					UpperCamelCase_   :       Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]						,					UpperCamelCase_   :       StableDiffusionSafetyChecker						,					UpperCamelCase_   :       CLIPImageProcessor						,					UpperCamelCase_   :       bool = True						,					)->			List[str]:
     super().__init__()
     if hasattr(scheduler.config						,					'steps_offset'     ) and scheduler.config.steps_offset != 1:
       _lowercase: Union[str, Any]						     =    (
           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'						,					UpperCamelCase_						,					standard_warn=UpperCamelCase_     )
       _lowercase: List[str]						     =    dict(scheduler.config     )
       _lowercase: List[str]						     =    1
       _lowercase: Union[str, Any]						     =    FrozenDict(UpperCamelCase_     )

     self.register_modules(
         vae=UpperCamelCase_						,					text_encoder=UpperCamelCase_						,					tokenizer=UpperCamelCase_						,					unet=UpperCamelCase_						,					scheduler=UpperCamelCase_						,					safety_checker=UpperCamelCase_						,					feature_extractor=UpperCamelCase_						,					)
     self.register_to_config(requires_safety_checker=UpperCamelCase_     )






   @property
   def  _a						(	self   :       str     )->			Dict[str, Any]:
     return {k: getattr(self						,					UpperCamelCase_     ) for k in self.config.keys() if not k.startswith('_'     )}






   def  _a						(	self   :       Any						,					UpperCamelCase_   :       Optional[Union[str, int]] = "auto"     )->			Optional[Any]:
     if slice_size == "auto":
       # half the attention head size is usually a good trade-off between
       # speed and memory
       _lowercase: Any						     =    self.unet.config.attention_head_dim // 2
     self.unet.set_attention_slice(UpperCamelCase_     )






   def  _a						(	self   :       Any     )->			Any:
     self.enable_attention_slicing(UpperCamelCase_     )






   @torch.no_grad()
   def  _a						(	self   :       Any						,					UpperCamelCase_   :       Union[str, List[str]]						,					UpperCamelCase_   :       Union[torch.FloatTensor, PIL.Image.Image]						,					UpperCamelCase_   :       Union[torch.FloatTensor, PIL.Image.Image]						,					UpperCamelCase_   :       float = 0.8						,					UpperCamelCase_   :       Optional[int] = 50						,					UpperCamelCase_   :       Optional[float] = 7.5						,					UpperCamelCase_   :       Optional[Union[str, List[str]]] = None						,					UpperCamelCase_   :       Optional[int] = 1						,					UpperCamelCase_   :       Optional[float] = 0.0						,					UpperCamelCase_   :       Optional[torch.Generator] = None						,					UpperCamelCase_   :       Optional[str] = "pil"						,					UpperCamelCase_   :       bool = True						,					UpperCamelCase_   :       Optional[Callable[[int, int, torch.FloatTensor], None]] = None						,					UpperCamelCase_   :       int = 1						,					)->			List[str]:
     # For more information on how this function works, please see: https://huggingface.co/docs/diffusers/api/pipelines/stable_diffusion#diffusers.StableDiffusionImg2ImgPipeline
     return StableDiffusionInpaintPipelineLegacy(**self.components     )(
         prompt=UpperCamelCase_						,					image=UpperCamelCase_						,					mask_image=UpperCamelCase_						,					strength=UpperCamelCase_						,					num_inference_steps=UpperCamelCase_						,					guidance_scale=UpperCamelCase_						,					negative_prompt=UpperCamelCase_						,					num_images_per_prompt=UpperCamelCase_						,					eta=UpperCamelCase_						,					generator=UpperCamelCase_						,					output_type=UpperCamelCase_						,					return_dict=UpperCamelCase_						,					callback=UpperCamelCase_						,					)






   @torch.no_grad()
   def  _a						(	self   :       Any						,					UpperCamelCase_   :       Union[str, List[str]]						,					UpperCamelCase_   :       Union[torch.FloatTensor, PIL.Image.Image]						,					UpperCamelCase_   :       float = 0.8						,					UpperCamelCase_   :       Optional[int] = 50						,					UpperCamelCase_   :       Optional[float] = 7.5						,					UpperCamelCase_   :       Optional[Union[str, List[str]]] = None						,					UpperCamelCase_   :       Optional[int] = 1						,					UpperCamelCase_   :       Optional[float] = 0.0						,					UpperCamelCase_   :       Optional[torch.Generator] = None						,					UpperCamelCase_   :       Optional[str] = "pil"						,					UpperCamelCase_   :       bool = True						,					UpperCamelCase_   :       Optional[Callable[[int, int, torch.FloatTensor], None]] = None						,					UpperCamelCase_   :       int = 1						,					**UpperCamelCase_   :       Optional[int]						,					)->			List[Any]:
     # For more information on how this function works, please see: https://huggingface.co/docs/diffusers/api/pipelines/stable_diffusion#diffusers.StableDiffusionImg2ImgPipeline
     return StableDiffusionImgaImgPipeline(**self.components     )(
         prompt=UpperCamelCase_						,					image=UpperCamelCase_						,					strength=UpperCamelCase_						,					num_inference_steps=UpperCamelCase_						,					guidance_scale=UpperCamelCase_						,					negative_prompt=UpperCamelCase_						,					num_images_per_prompt=UpperCamelCase_						,					eta=UpperCamelCase_						,					generator=UpperCamelCase_						,					output_type=UpperCamelCase_						,					return_dict=UpperCamelCase_						,					callback=UpperCamelCase_						,					callback_steps=UpperCamelCase_						,					)






   @torch.no_grad()
   def  _a						(	self   :       Tuple						,					UpperCamelCase_   :       Union[str, List[str]]						,					UpperCamelCase_   :       int = 5_12						,					UpperCamelCase_   :       int = 5_12						,					UpperCamelCase_   :       int = 50						,					UpperCamelCase_   :       float = 7.5						,					UpperCamelCase_   :       Optional[Union[str, List[str]]] = None						,					UpperCamelCase_   :       Optional[int] = 1						,					UpperCamelCase_   :       float = 0.0						,					UpperCamelCase_   :       Optional[torch.Generator] = None						,					UpperCamelCase_   :       Optional[torch.FloatTensor] = None						,					UpperCamelCase_   :       Optional[str] = "pil"						,					UpperCamelCase_   :       bool = True						,					UpperCamelCase_   :       Optional[Callable[[int, int, torch.FloatTensor], None]] = None						,					UpperCamelCase_   :       int = 1						,					)->			Union[str, Any]:
     # For more information on how this function https://huggingface.co/docs/diffusers/api/pipelines/stable_diffusion#diffusers.StableDiffusionPipeline
     return StableDiffusionPipeline(**self.components     )(
         prompt=UpperCamelCase_						,					height=UpperCamelCase_						,					width=UpperCamelCase_						,					num_inference_steps=UpperCamelCase_						,					guidance_scale=UpperCamelCase_						,					negative_prompt=UpperCamelCase_						,					num_images_per_prompt=UpperCamelCase_						,					eta=UpperCamelCase_						,					generator=UpperCamelCase_						,					latents=UpperCamelCase_						,					output_type=UpperCamelCase_						,					return_dict=UpperCamelCase_						,					callback=UpperCamelCase_						,					callback_steps=UpperCamelCase_						,					)






import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin

class      UpperCAmelCase						(			A   ):
   __SCREAMING_SNAKE_CASE		:							Optional[Any]							=					"AutoFeatureExtractor"
   __SCREAMING_SNAKE_CASE		:							Optional[Any]							=					"Speech2Text2Tokenizer"






   def __init__(	self   :       Optional[Any]						,					UpperCamelCase_   :       Any						,					UpperCamelCase_   :       List[Any]     )->			Dict:
     super().__init__(UpperCamelCase_						,					UpperCamelCase_     )
     _lowercase: Optional[int]						     =    self.feature_extractor
     _lowercase: int						     =    False






   def __call__(	self   :       List[Any]						,					*UpperCamelCase_   :       Dict						,					**UpperCamelCase_   :       Optional[int]     )->			str:
     # For backward compatibility
     if self._in_target_context_manager:
       return self.current_processor(*UpperCamelCase_						,					**UpperCamelCase_     )

     if "raw_speech" in kwargs:
       warnings.warn('Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.'     )
       _lowercase: Tuple						     =    kwargs.pop('raw_speech'     )
     else:
       _lowercase: Optional[Any]						     =    kwargs.pop('audio'						,					UpperCamelCase_     )
     _lowercase: int						     =    kwargs.pop('sampling_rate'						,					UpperCamelCase_     )
     _lowercase: Dict						     =    kwargs.pop('text'						,					UpperCamelCase_     )
     if len(UpperCamelCase_     ) > 0:
       _lowercase: List[Any]						     =    args[0]
       _lowercase: List[Any]						     =    args[1:]

     if audio is None and text is None:
       raise ValueError('You need to specify either an `audio` or `text` input to process.'     )

     if audio is not None:
       _lowercase: Optional[int]						     =    self.feature_extractor(UpperCamelCase_						,					*UpperCamelCase_						,					sampling_rate=UpperCamelCase_						,					**UpperCamelCase_     )
     if text is not None:
       _lowercase: str						     =    self.tokenizer(UpperCamelCase_						,					**UpperCamelCase_     )

     if text is None:
       return inputs
     elif audio is None:
       return encodings
     else:
       _lowercase: int						     =    encodings['input_ids']
       return inputs






   def  _a						(	self   :       Optional[int]						,					*UpperCamelCase_   :       Tuple						,					**UpperCamelCase_   :       Dict     )->			List[str]:
     return self.tokenizer.batch_decode(*UpperCamelCase_						,					**UpperCamelCase_     )






   def  _a						(	self   :       Tuple						,					*UpperCamelCase_   :       List[Any]						,					**UpperCamelCase_   :       Union[str, Any]     )->			Optional[int]:
     return self.tokenizer.decode(*UpperCamelCase_						,					**UpperCamelCase_     )






   @contextmanager
   def  _a						(	self   :       Tuple     )->			Union[str, Any]:
     warnings.warn(
         '`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '
         'labels by using the argument `text` of the regular `__call__` method (either in the same call as '
         'your audio inputs, or in a separate call.'     )
     _lowercase: Any						     =    True
     _lowercase: Optional[Any]						     =    self.tokenizer
     yield
     _lowercase: Union[str, Any]						     =    self.feature_extractor
     _lowercase: str						     =    False
