
'''simple docstring'''


from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


__SCREAMING_SNAKE_CASE						=      logging.get_logger(__name__)




class     __snake_case			(    _a	):
		a_          =					["input_values", "attention_mask"]




		def __init__(  self					:List[Any]			,    _SCREAMING_SNAKE_CASE					:Union[str, Any]=1			,    _SCREAMING_SNAKE_CASE					:Tuple=1_6_0_0_0			,    _SCREAMING_SNAKE_CASE					:Optional[Any]=1_2_8			,    _SCREAMING_SNAKE_CASE					:Optional[Any]=1_0_2_4			,    _SCREAMING_SNAKE_CASE					:str=0.0			,    _SCREAMING_SNAKE_CASE					:Tuple=True			,    _SCREAMING_SNAKE_CASE					:Optional[Any]=-4.2677393			,    _SCREAMING_SNAKE_CASE					:str=4.5689974			,    _SCREAMING_SNAKE_CASE					:Optional[int]=False			,    **_SCREAMING_SNAKE_CASE					:Union[str, Any]			,    )					->			int:






								'''simple docstring'''






								super().__init__(feature_size=_SCREAMING_SNAKE_CASE			,    sampling_rate=_SCREAMING_SNAKE_CASE			,    padding_value=_SCREAMING_SNAKE_CASE			,    **_SCREAMING_SNAKE_CASE)
								UpperCamelCase_     :	Union[str, Any]					     =     num_mel_bins
								UpperCamelCase_     :	Tuple					     =     max_length
								UpperCamelCase_     :	Optional[Any]					     =     do_normalize
								UpperCamelCase_     :	List[Any]					     =     mean
								UpperCamelCase_     :	List[Any]					     =     std
								UpperCamelCase_     :	str					     =     return_attention_mask




		def 	_snake_case	(  self					:List[str]			,    _SCREAMING_SNAKE_CASE					:np.ndarray			,    _SCREAMING_SNAKE_CASE					:int			,    )					->			np.ndarray:






								'''simple docstring'''






								UpperCamelCase_     :	Any					     =     torch.from_numpy(_SCREAMING_SNAKE_CASE).unsqueeze(0)
								UpperCamelCase_     :	Dict					     =     ta_kaldi.fbank(
								    _SCREAMING_SNAKE_CASE			,    htk_compat=_SCREAMING_SNAKE_CASE			,    sample_frequency=self.sampling_rate			,    use_energy=_SCREAMING_SNAKE_CASE			,    window_type="""hanning"""			,    num_mel_bins=self.num_mel_bins			,    dither=0.0			,    frame_shift=1_0			,    )

								UpperCamelCase_     :	Optional[int]					     =     fbank.shape[0]
								UpperCamelCase_     :	List[str]					     =     max_length - n_frames

								# pad or truncate, depending on difference
								if difference > 0:
														UpperCamelCase_     :	Dict					     =     torch.nn.ZeroPadad((0, 0, 0, difference))
														UpperCamelCase_     :	Tuple					     =     pad_module(_SCREAMING_SNAKE_CASE)
								elif difference < 0:
														UpperCamelCase_     :	int					     =     fbank[0:max_length, :]

								UpperCamelCase_     :	str					     =     fbank.numpy()

								return fbank




		def 	_snake_case	(  self					:Tuple			,    _SCREAMING_SNAKE_CASE					:np.ndarray)					->			np.ndarray:






								'''simple docstring'''






								return (input_values - (self.mean)) / (self.std * 2)





		def __call__(  self					:Dict			,    _SCREAMING_SNAKE_CASE					:Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]			,    _SCREAMING_SNAKE_CASE					:Optional[int] = None			,    _SCREAMING_SNAKE_CASE					:Optional[Union[str, TensorType]] = None			,    **_SCREAMING_SNAKE_CASE					:Optional[Any]			,    )					->			BatchFeature:






								'''simple docstring'''






								if sampling_rate is not None:
														if sampling_rate != self.sampling_rate:
																				raise ValueError(
																				    F"""The model corresponding to this feature extractor: {self} was trained using a sampling rate of"""
																				    F""" {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with"""
																				    F""" {self.sampling_rate} and not {sampling_rate}.""")
								else:
														logger.warning(
														    """It is strongly recommended to pass the `sampling_rate` argument to this function. """
														    """Failing to do so can result in silent errors that might be hard to debug.""")

								UpperCamelCase_     :	Any					     =     isinstance(_SCREAMING_SNAKE_CASE			,    np.ndarray) and len(raw_speech.shape) > 1
								if is_batched_numpy and len(raw_speech.shape) > 2:
														raise ValueError(F"""Only mono-channel audio is supported for input to {self}""")
								UpperCamelCase_     :	List[Any]					     =     is_batched_numpy or (
								    isinstance(_SCREAMING_SNAKE_CASE			,    (list, tuple)) and (isinstance(raw_speech[0]			,    (np.ndarray, tuple, list)))
								)

								if is_batched:
														UpperCamelCase_     :	Dict					     =     [np.asarray(_SCREAMING_SNAKE_CASE			,    dtype=np.floataa) for speech in raw_speech]
								elif not is_batched and not isinstance(_SCREAMING_SNAKE_CASE			,    np.ndarray):
														UpperCamelCase_     :	Any					     =     np.asarray(_SCREAMING_SNAKE_CASE			,    dtype=np.floataa)
								elif isinstance(_SCREAMING_SNAKE_CASE			,    np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
														UpperCamelCase_     :	str					     =     raw_speech.astype(np.floataa)

								# always return batch
								if not is_batched:
														UpperCamelCase_     :	Union[str, Any]					     =     [raw_speech]

								# extract fbank features and pad/truncate to max_length
								UpperCamelCase_     :	List[Any]					     =     [self._extract_fbank_features(_SCREAMING_SNAKE_CASE			,    max_length=self.max_length) for waveform in raw_speech]

								# convert into BatchFeature
								UpperCamelCase_     :	List[str]					     =     BatchFeature({"""input_values""": features})

								# make sure list is in array format
								UpperCamelCase_     :	Tuple					     =     padded_inputs.get("""input_values""")
								if isinstance(input_values[0]			,    _SCREAMING_SNAKE_CASE):
														UpperCamelCase_     :	List[Any]					     =     [np.asarray(_SCREAMING_SNAKE_CASE			,    dtype=np.floataa) for feature in input_values]

								# normalization
								if self.do_normalize:
														UpperCamelCase_     :	Optional[int]					     =     [self.normalize(_SCREAMING_SNAKE_CASE) for feature in input_values]

								if return_tensors is not None:
														UpperCamelCase_     :	Optional[int]					     =     padded_inputs.convert_to_tensors(_SCREAMING_SNAKE_CASE)

								return padded_inputs






'''simple docstring'''


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


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




class     __snake_case			(    _a	):
		a_          =					["safety_checker", "feature_extractor"]




		def __init__(  self					:str			,    _SCREAMING_SNAKE_CASE					:AutoencoderKL			,    _SCREAMING_SNAKE_CASE					:CLIPTextModel			,    _SCREAMING_SNAKE_CASE					:CLIPTokenizer			,    _SCREAMING_SNAKE_CASE					:UNetaDConditionModel			,    _SCREAMING_SNAKE_CASE					:Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]			,    _SCREAMING_SNAKE_CASE					:StableDiffusionSafetyChecker			,    _SCREAMING_SNAKE_CASE					:CLIPImageProcessor			,    _SCREAMING_SNAKE_CASE					:bool = True			,    )					->			int:






								'''simple docstring'''






								super().__init__()
								if hasattr(scheduler.config			,    """steps_offset""") and scheduler.config.steps_offset != 1:
														UpperCamelCase_     :	Optional[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"""			,    _SCREAMING_SNAKE_CASE			,    standard_warn=_SCREAMING_SNAKE_CASE)
														UpperCamelCase_     :	Union[str, Any]					     =     dict(scheduler.config)
														UpperCamelCase_     :	Optional[Any]					     =     1
														UpperCamelCase_     :	int					     =     FrozenDict(_SCREAMING_SNAKE_CASE)

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




		@property
		def 	_snake_case	(  self					:Tuple)					->			Dict[str, Any]:






								'''simple docstring'''






								return {k: getattr(self			,    _SCREAMING_SNAKE_CASE) for k in self.config.keys() if not k.startswith("""_""")}




		def 	_snake_case	(  self					:Optional[int]			,    _SCREAMING_SNAKE_CASE					:Optional[Union[str, int]] = "auto")					->			Union[str, Any]:






								'''simple docstring'''






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




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






								'''simple docstring'''






								self.enable_attention_slicing(_SCREAMING_SNAKE_CASE)




		@torch.no_grad()
		def 	_snake_case	(  self					:Union[str, Any]			,    _SCREAMING_SNAKE_CASE					:Union[str, List[str]]			,    _SCREAMING_SNAKE_CASE					:Union[torch.FloatTensor, PIL.Image.Image]			,    _SCREAMING_SNAKE_CASE					:Union[torch.FloatTensor, PIL.Image.Image]			,    _SCREAMING_SNAKE_CASE					:float = 0.8			,    _SCREAMING_SNAKE_CASE					:Optional[int] = 5_0			,    _SCREAMING_SNAKE_CASE					:Optional[float] = 7.5			,    _SCREAMING_SNAKE_CASE					:Optional[Union[str, List[str]]] = None			,    _SCREAMING_SNAKE_CASE					:Optional[int] = 1			,    _SCREAMING_SNAKE_CASE					:Optional[float] = 0.0			,    _SCREAMING_SNAKE_CASE					:Optional[torch.Generator] = None			,    _SCREAMING_SNAKE_CASE					:Optional[str] = "pil"			,    _SCREAMING_SNAKE_CASE					:bool = True			,    _SCREAMING_SNAKE_CASE					:Optional[Callable[[int, int, torch.FloatTensor], None]] = None			,    _SCREAMING_SNAKE_CASE					:int = 1			,    )					->			List[str]:






								'''simple docstring'''






								return StableDiffusionInpaintPipelineLegacy(**self.components)(
								    prompt=_SCREAMING_SNAKE_CASE			,    image=_SCREAMING_SNAKE_CASE			,    mask_image=_SCREAMING_SNAKE_CASE			,    strength=_SCREAMING_SNAKE_CASE			,    num_inference_steps=_SCREAMING_SNAKE_CASE			,    guidance_scale=_SCREAMING_SNAKE_CASE			,    negative_prompt=_SCREAMING_SNAKE_CASE			,    num_images_per_prompt=_SCREAMING_SNAKE_CASE			,    eta=_SCREAMING_SNAKE_CASE			,    generator=_SCREAMING_SNAKE_CASE			,    output_type=_SCREAMING_SNAKE_CASE			,    return_dict=_SCREAMING_SNAKE_CASE			,    callback=_SCREAMING_SNAKE_CASE			,    )




		@torch.no_grad()
		def 	_snake_case	(  self					:List[str]			,    _SCREAMING_SNAKE_CASE					:Union[str, List[str]]			,    _SCREAMING_SNAKE_CASE					:Union[torch.FloatTensor, PIL.Image.Image]			,    _SCREAMING_SNAKE_CASE					:float = 0.8			,    _SCREAMING_SNAKE_CASE					:Optional[int] = 5_0			,    _SCREAMING_SNAKE_CASE					:Optional[float] = 7.5			,    _SCREAMING_SNAKE_CASE					:Optional[Union[str, List[str]]] = None			,    _SCREAMING_SNAKE_CASE					:Optional[int] = 1			,    _SCREAMING_SNAKE_CASE					:Optional[float] = 0.0			,    _SCREAMING_SNAKE_CASE					:Optional[torch.Generator] = None			,    _SCREAMING_SNAKE_CASE					:Optional[str] = "pil"			,    _SCREAMING_SNAKE_CASE					:bool = True			,    _SCREAMING_SNAKE_CASE					:Optional[Callable[[int, int, torch.FloatTensor], None]] = None			,    _SCREAMING_SNAKE_CASE					:int = 1			,    **_SCREAMING_SNAKE_CASE					:int			,    )					->			Any:






								'''simple docstring'''






								return StableDiffusionImgaImgPipeline(**self.components)(
								    prompt=_SCREAMING_SNAKE_CASE			,    image=_SCREAMING_SNAKE_CASE			,    strength=_SCREAMING_SNAKE_CASE			,    num_inference_steps=_SCREAMING_SNAKE_CASE			,    guidance_scale=_SCREAMING_SNAKE_CASE			,    negative_prompt=_SCREAMING_SNAKE_CASE			,    num_images_per_prompt=_SCREAMING_SNAKE_CASE			,    eta=_SCREAMING_SNAKE_CASE			,    generator=_SCREAMING_SNAKE_CASE			,    output_type=_SCREAMING_SNAKE_CASE			,    return_dict=_SCREAMING_SNAKE_CASE			,    callback=_SCREAMING_SNAKE_CASE			,    callback_steps=_SCREAMING_SNAKE_CASE			,    )





		@torch.no_grad()
		def 	_snake_case	(  self					:Any			,    _SCREAMING_SNAKE_CASE					:Union[str, List[str]]			,    _SCREAMING_SNAKE_CASE					:int = 5_1_2			,    _SCREAMING_SNAKE_CASE					:int = 5_1_2			,    _SCREAMING_SNAKE_CASE					:int = 5_0			,    _SCREAMING_SNAKE_CASE					:float = 7.5			,    _SCREAMING_SNAKE_CASE					:Optional[Union[str, List[str]]] = None			,    _SCREAMING_SNAKE_CASE					:Optional[int] = 1			,    _SCREAMING_SNAKE_CASE					:float = 0.0			,    _SCREAMING_SNAKE_CASE					:Optional[torch.Generator] = None			,    _SCREAMING_SNAKE_CASE					:Optional[torch.FloatTensor] = None			,    _SCREAMING_SNAKE_CASE					:Optional[str] = "pil"			,    _SCREAMING_SNAKE_CASE					:bool = True			,    _SCREAMING_SNAKE_CASE					:Optional[Callable[[int, int, torch.FloatTensor], None]] = None			,    _SCREAMING_SNAKE_CASE					:int = 1			,    )					->			List[Any]:






								'''simple docstring'''






								return StableDiffusionPipeline(**self.components)(
								    prompt=_SCREAMING_SNAKE_CASE			,    height=_SCREAMING_SNAKE_CASE			,    width=_SCREAMING_SNAKE_CASE			,    num_inference_steps=_SCREAMING_SNAKE_CASE			,    guidance_scale=_SCREAMING_SNAKE_CASE			,    negative_prompt=_SCREAMING_SNAKE_CASE			,    num_images_per_prompt=_SCREAMING_SNAKE_CASE			,    eta=_SCREAMING_SNAKE_CASE			,    generator=_SCREAMING_SNAKE_CASE			,    latents=_SCREAMING_SNAKE_CASE			,    output_type=_SCREAMING_SNAKE_CASE			,    return_dict=_SCREAMING_SNAKE_CASE			,    callback=_SCREAMING_SNAKE_CASE			,    callback_steps=_SCREAMING_SNAKE_CASE			,    )




