





import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextModel,
    CLIPTokenizer,
    WhisperForConditionalGeneration,
    WhisperProcessor,
)

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import logging


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






class        _a(    __A    ):







		def __init__(    self ,						__snake_case ,						__snake_case ,						__snake_case ,						__snake_case ,						__snake_case ,						__snake_case ,						__snake_case ,						__snake_case ,						__snake_case ,						)      ->      Union[str, Any]:


							'''simple docstring'''

							super().__init__()

							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(
							    speech_model=__snake_case ,						speech_processor=__snake_case ,						vae=__snake_case ,						text_encoder=__snake_case ,						tokenizer=__snake_case ,						unet=__snake_case ,						scheduler=__snake_case ,						feature_extractor=__snake_case ,						)







		def        lowercase       (    self ,						__snake_case = "auto"				)      ->      int:


							'''simple docstring'''

							if slice_size == "auto":
												_snake_case       :					List[str]				       =   self.unet.config.attention_head_dim // 2
							self.unet.set_attention_slice(__snake_case				)







		def        lowercase       (    self				)      ->      Optional[int]:


							'''simple docstring'''

							self.enable_attention_slicing(__snake_case				)







		@torch.no_grad()
		def __call__(    self ,						__snake_case ,						__snake_case=1_6_0_0_0 ,						__snake_case = 5_1_2 ,						__snake_case = 5_1_2 ,						__snake_case = 5_0 ,						__snake_case = 7.5 ,						__snake_case = None ,						__snake_case = 1 ,						__snake_case = 0.0 ,						__snake_case = None ,						__snake_case = None ,						__snake_case = "pil" ,						__snake_case = True ,						__snake_case = None ,						__snake_case = 1 ,						**__snake_case ,						)      ->      Optional[int]:


							'''simple docstring'''

							_snake_case       :					Any				       =   self.speech_processor.feature_extractor(
							    __snake_case ,						return_tensors="pt" ,						sampling_rate=__snake_case				).input_features.to(self.device				)
							_snake_case       :					Dict				       =   self.speech_model.generate(__snake_case ,						max_length=4_8_0_0_0_0				)

							_snake_case       :					Optional[Any]				       =   self.speech_processor.tokenizer.batch_decode(__snake_case ,						skip_special_tokens=__snake_case ,						normalize=__snake_case				)[
							    0
							]

							if isinstance(__snake_case ,						__snake_case				):
												_snake_case       :					Tuple				       =   1
							elif isinstance(__snake_case ,						__snake_case				):
												_snake_case       :					Any				       =   len(__snake_case				)
							else:
												raise ValueError(f"""`prompt` has to be of type `str` or `list` but is {type(__snake_case				)}"""				)

							if height % 8 != 0 or width % 8 != 0:
												raise ValueError(f"""`height` and `width` have to be divisible by 8 but are {height} and {width}."""				)

							if (callback_steps is None) or (
							    callback_steps is not None and (not isinstance(__snake_case ,						__snake_case				) or callback_steps <= 0)
							):
												raise ValueError(
												    f"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
												    f""" {type(__snake_case				)}."""				)

							# get prompt text embeddings
							_snake_case       :					Dict				       =   self.tokenizer(
							    __snake_case ,						padding="max_length" ,						max_length=self.tokenizer.model_max_length ,						return_tensors="pt" ,						)
							_snake_case       :					Union[str, Any]				       =   text_inputs.input_ids

							if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
												_snake_case       :					int				       =   self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]				)
												logger.warning(
												    "The following part of your input was truncated because CLIP can only handle sequences up to"
												    f""" {self.tokenizer.model_max_length} tokens: {removed_text}"""				)
												_snake_case       :					Union[str, Any]				       =   text_input_ids[:, : self.tokenizer.model_max_length]
							_snake_case       :					int				       =   self.text_encoder(text_input_ids.to(self.device				)				)[0]

							# duplicate text embeddings for each generation per prompt, using mps friendly method
							_snake_case       :					int				       =   text_embeddings.shape
							_snake_case       :					List[str]				       =   text_embeddings.repeat(1 ,						__snake_case ,						1				)
							_snake_case       :					List[str]				       =   text_embeddings.view(bs_embed * num_images_per_prompt ,						__snake_case ,						-1				)

							# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
							# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
							# corresponds to doing no classifier free guidance.
							_snake_case       :					Union[str, Any]				       =   guidance_scale > 1.0
							# get unconditional embeddings for classifier free guidance
							if do_classifier_free_guidance:
												_snake_case       :					List[str]
												if negative_prompt is None:
																	_snake_case       :					Optional[int]				       =   [""] * batch_size
												elif type(__snake_case				) is not type(__snake_case				):
																	raise TypeError(
																	    f"""`negative_prompt` should be the same type to `prompt`, but got {type(__snake_case				)} !="""
																	    f""" {type(__snake_case				)}."""				)
												elif isinstance(__snake_case ,						__snake_case				):
																	_snake_case       :					Optional[Any]				       =   [negative_prompt]
												elif batch_size != len(__snake_case				):
																	raise ValueError(
																	    f"""`negative_prompt`: {negative_prompt} has batch size {len(__snake_case				)}, but `prompt`:"""
																	    f""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
																	    " the batch size of `prompt`."				)
												else:
																	_snake_case       :					Tuple				       =   negative_prompt

												_snake_case       :					Optional[Any]				       =   text_input_ids.shape[-1]
												_snake_case       :					List[Any]				       =   self.tokenizer(
												    __snake_case ,						padding="max_length" ,						max_length=__snake_case ,						truncation=__snake_case ,						return_tensors="pt" ,						)
												_snake_case       :					Tuple				       =   self.text_encoder(uncond_input.input_ids.to(self.device				)				)[0]

												# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
												_snake_case       :					Optional[Any]				       =   uncond_embeddings.shape[1]
												_snake_case       :					str				       =   uncond_embeddings.repeat(1 ,						__snake_case ,						1				)
												_snake_case       :					Dict				       =   uncond_embeddings.view(batch_size * num_images_per_prompt ,						__snake_case ,						-1				)

												# For classifier free guidance, we need to do two forward passes.
												# Here we concatenate the unconditional and text embeddings into a single batch
												# to avoid doing two forward passes
												_snake_case       :					List[str]				       =   torch.cat([uncond_embeddings, text_embeddings]				)

							# get the initial random noise unless the user supplied it

							# Unlike in other pipelines, latents need to be generated in the target device
							# for 1-to-1 results reproducibility with the CompVis implementation.
							# However this currently doesn't work in `mps`.
							_snake_case       :					str				       =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
							_snake_case       :					Dict				       =   text_embeddings.dtype
							if latents is None:
												if self.device.type == "mps":
																	# randn does not exist on mps
																	_snake_case       :					Union[str, Any]				       =   torch.randn(__snake_case ,						generator=__snake_case ,						device="cpu" ,						dtype=__snake_case				).to(
																	    self.device				)
												else:
																	_snake_case       :					Optional[int]				       =   torch.randn(__snake_case ,						generator=__snake_case ,						device=self.device ,						dtype=__snake_case				)
							else:
												if latents.shape != latents_shape:
																	raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""				)
												_snake_case       :					List[Any]				       =   latents.to(self.device				)

							# set timesteps
							self.scheduler.set_timesteps(__snake_case				)

							# Some schedulers like PNDM have timesteps as arrays
							# It's more optimized to move all timesteps to correct device beforehand
							_snake_case       :					Optional[Any]				       =   self.scheduler.timesteps.to(self.device				)

							# scale the initial noise by the standard deviation required by the scheduler
							_snake_case       :					List[str]				       =   latents * self.scheduler.init_noise_sigma

							# prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
							# eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
							# eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
							# and should be between [0, 1]
							_snake_case       :					Any				       =   "eta" in set(inspect.signature(self.scheduler.step				).parameters.keys()				)
							_snake_case       :					str				       =   {}
							if accepts_eta:
												_snake_case       :					Any				       =   eta

							for i, t in enumerate(self.progress_bar(__snake_case				)				):
												# expand the latents if we are doing classifier free guidance
												_snake_case       :					List[Any]				       =   torch.cat([latents] * 2				) if do_classifier_free_guidance else latents
												_snake_case       :					Optional[Any]				       =   self.scheduler.scale_model_input(__snake_case ,						__snake_case				)

												# predict the noise residual
												_snake_case       :					List[Any]				       =   self.unet(__snake_case ,						__snake_case ,						encoder_hidden_states=__snake_case				).sample

												# perform guidance
												if do_classifier_free_guidance:
																	_snake_case       :					Optional[int]				       =   noise_pred.chunk(2				)
																	_snake_case       :					Any				       =   noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

												# compute the previous noisy sample x_t -> x_t-1
												_snake_case       :					Dict				       =   self.scheduler.step(__snake_case ,						__snake_case ,						__snake_case ,						**__snake_case				).prev_sample

												# call the callback, if provided
												if callback is not None and i % callback_steps == 0:
																	callback(__snake_case ,						__snake_case ,						__snake_case				)

							_snake_case       :					List[str]				       =   1 / 0.1_82_15 * latents
							_snake_case       :					List[Any]				       =   self.vae.decode(__snake_case				).sample

							_snake_case       :					List[str]				       =   (image / 2 + 0.5).clamp(0 ,						1				)

							# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
							_snake_case       :					Any				       =   image.cpu().permute(0 ,						2 ,						3 ,						1				).float().numpy()

							if output_type == "pil":
												_snake_case       :					str				       =   self.numpy_to_pil(__snake_case				)

							if not return_dict:
												return image

							return StableDiffusionPipelineOutput(images=__snake_case ,						nsfw_content_detected=__snake_case				)






import numpy as np

__lowerCAmelCase :Dict     =    [
    ['a', 'b', 'c', 'd', 'e'],
    ['f', 'g', 'h', 'i', 'k'],
    ['l', 'm', 'n', 'o', 'p'],
    ['q', 'r', 's', 't', 'u'],
    ['v', 'w', 'x', 'y', 'z'],
]






class        _a:







		def __init__(    self				)      ->      None:


							'''simple docstring'''

							_snake_case       :					Optional[Any]				       =   np.array(__snake_case				)







		def        lowercase       (    self ,						__snake_case				)      ->      np.ndarray:


							'''simple docstring'''

							_snake_case       , _snake_case       :					List[Any]				       =   np.where(letter == self.SQUARE				)
							_snake_case       :					Any				       =   np.concatenate([indexa + 1, indexa + 1]				)
							return indexes







		def        lowercase       (    self ,						__snake_case ,						__snake_case				)      ->      str:


							'''simple docstring'''

							_snake_case       :					int				       =   self.SQUARE[indexa - 1, indexa - 1]
							return letter







		def        lowercase       (    self ,						__snake_case				)      ->      str:


							'''simple docstring'''

							_snake_case       :					Optional[Any]				       =   message.lower()
							_snake_case       :					List[Any]				       =   message.replace(" " ,						""				)
							_snake_case       :					Any				       =   message.replace("j" ,						"i"				)

							_snake_case       :					Tuple				       =   np.empty((2, len(__snake_case				))				)
							for letter_index in range(len(__snake_case				)				):
												_snake_case       :					Dict				       =   self.letter_to_numbers(message[letter_index]				)

												_snake_case       :					Tuple				       =   numbers[0]
												_snake_case       :					List[Any]				       =   numbers[1]

							_snake_case       :					Any				       =   first_step.reshape(2 * len(__snake_case				)				)
							_snake_case       :					Optional[Any]				       =   ""
							for numbers_index in range(len(__snake_case				)				):
												_snake_case       :					Optional[int]				       =   int(second_step[numbers_index * 2]				)
												_snake_case       :					List[str]				       =   int(second_step[(numbers_index * 2) + 1]				)
												_snake_case       :					Any				       =   self.numbers_to_letter(__snake_case ,						__snake_case				)
												_snake_case       :					Any				       =   encoded_message + letter

							return encoded_message







		def        lowercase       (    self ,						__snake_case				)      ->      str:


							'''simple docstring'''

							_snake_case       :					Union[str, Any]				       =   message.lower()
							message.replace(" " ,						""				)
							_snake_case       :					Tuple				       =   np.empty(2 * len(__snake_case				)				)
							for letter_index in range(len(__snake_case				)				):
												_snake_case       :					Union[str, Any]				       =   self.letter_to_numbers(message[letter_index]				)
												_snake_case       :					Tuple				       =   numbers[0]
												_snake_case       :					Any				       =   numbers[1]

							_snake_case       :					Optional[Any]				       =   first_step.reshape((2, len(__snake_case				))				)
							_snake_case       :					Union[str, Any]				       =   ""
							for numbers_index in range(len(__snake_case				)				):
												_snake_case       :					Dict				       =   int(second_step[0, numbers_index]				)
												_snake_case       :					Optional[Any]				       =   int(second_step[1, numbers_index]				)
												_snake_case       :					Union[str, Any]				       =   self.numbers_to_letter(__snake_case ,						__snake_case				)
												_snake_case       :					str				       =   decoded_message + letter

							return decoded_message