






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


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





class __lowerCamelCase (      snake_case__):





	"""simple docstring"""






	def __init__(							self       ,    UpperCAmelCase       ,    UpperCAmelCase       ,    UpperCAmelCase       ,    UpperCAmelCase       ,    UpperCAmelCase       ,    UpperCAmelCase       ,    UpperCAmelCase       ,    UpperCAmelCase       ,    UpperCAmelCase       ,    ):
							"""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=UpperCAmelCase       ,    speech_processor=UpperCAmelCase       ,    vae=UpperCAmelCase       ,    text_encoder=UpperCAmelCase       ,    tokenizer=UpperCAmelCase       ,    unet=UpperCAmelCase       ,    scheduler=UpperCAmelCase       ,    feature_extractor=UpperCAmelCase       ,    )

	def        UpperCamelCase    (							self       ,    UpperCAmelCase = "auto"    ):
							"""simple docstring"""







							if slice_size == "auto":
													_UpperCAmelCase									=			self.unet.config.attention_head_dim // 2
							self.unet.set_attention_slice(UpperCAmelCase    )

	def        UpperCamelCase    (							self    ):
							"""simple docstring"""







							self.enable_attention_slicing(UpperCAmelCase    )

	@torch.no_grad()
	def __call__(							self       ,    UpperCAmelCase       ,    UpperCAmelCase=1_6000       ,    UpperCAmelCase = 512       ,    UpperCAmelCase = 512       ,    UpperCAmelCase = 50       ,    UpperCAmelCase = 7.5       ,    UpperCAmelCase = None       ,    UpperCAmelCase = 1       ,    UpperCAmelCase = 0.0       ,    UpperCAmelCase = None       ,    UpperCAmelCase = None       ,    UpperCAmelCase = "pil"       ,    UpperCAmelCase = True       ,    UpperCAmelCase = None       ,    UpperCAmelCase = 1       ,    **UpperCAmelCase       ,    ):
							"""simple docstring"""







							_UpperCAmelCase									=			self.speech_processor.feature_extractor(
							    UpperCAmelCase       ,    return_tensors='pt'       ,    sampling_rate=UpperCAmelCase    ).input_features.to(self.device    )
							_UpperCAmelCase									=			self.speech_model.generate(UpperCAmelCase       ,    max_length=48_0000    )

							_UpperCAmelCase									=			self.speech_processor.tokenizer.batch_decode(UpperCAmelCase       ,    skip_special_tokens=UpperCAmelCase       ,    normalize=UpperCAmelCase    )[
							    0
							]

							if isinstance(UpperCAmelCase       ,    UpperCAmelCase    ):
													_UpperCAmelCase									=			1
							elif isinstance(UpperCAmelCase       ,    UpperCAmelCase    ):
													_UpperCAmelCase									=			len(UpperCAmelCase    )
							else:
													raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(UpperCAmelCase    )}"""    )

							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(UpperCAmelCase       ,    UpperCAmelCase    ) or callback_steps <= 0)
							):
													raise ValueError(
													    F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
													    F""" {type(UpperCAmelCase    )}."""    )

							# get prompt text embeddings
							_UpperCAmelCase									=			self.tokenizer(
							    UpperCAmelCase       ,    padding='max_length'       ,    max_length=self.tokenizer.model_max_length       ,    return_tensors='pt'       ,    )
							_UpperCAmelCase									=			text_inputs.input_ids

							if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
													_UpperCAmelCase									=			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}"""    )
													_UpperCAmelCase									=			text_input_ids[:, : self.tokenizer.model_max_length]
							_UpperCAmelCase									=			self.text_encoder(text_input_ids.to(self.device    )    )[0]

							# duplicate text embeddings for each generation per prompt, using mps friendly method
							_UpperCAmelCase					, _UpperCAmelCase					, _UpperCAmelCase									=			text_embeddings.shape
							_UpperCAmelCase									=			text_embeddings.repeat(1       ,    UpperCAmelCase       ,    1    )
							_UpperCAmelCase									=			text_embeddings.view(bs_embed * num_images_per_prompt       ,    UpperCAmelCase       ,    -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.
							_UpperCAmelCase									=			guidance_scale > 1.0
							# get unconditional embeddings for classifier free guidance
							if do_classifier_free_guidance:
													_UpperCAmelCase									=			42
													if negative_prompt is None:
																			_UpperCAmelCase									=			[''] * batch_size
													elif type(UpperCAmelCase    ) is not type(UpperCAmelCase    ):
																			raise TypeError(
																			    F"""`negative_prompt` should be the same type to `prompt`, but got {type(UpperCAmelCase    )} !="""
																			    F""" {type(UpperCAmelCase    )}."""    )
													elif isinstance(UpperCAmelCase       ,    UpperCAmelCase    ):
																			_UpperCAmelCase									=			[negative_prompt]
													elif batch_size != len(UpperCAmelCase    ):
																			raise ValueError(
																			    F"""`negative_prompt`: {negative_prompt} has batch size {len(UpperCAmelCase    )}, but `prompt`:"""
																			    F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
																			    ' the batch size of `prompt`.'    )
													else:
																			_UpperCAmelCase									=			negative_prompt

													_UpperCAmelCase									=			text_input_ids.shape[-1]
													_UpperCAmelCase									=			self.tokenizer(
													    UpperCAmelCase       ,    padding='max_length'       ,    max_length=UpperCAmelCase       ,    truncation=UpperCAmelCase       ,    return_tensors='pt'       ,    )
													_UpperCAmelCase									=			self.text_encoder(uncond_input.input_ids.to(self.device    )    )[0]

													# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
													_UpperCAmelCase									=			uncond_embeddings.shape[1]
													_UpperCAmelCase									=			uncond_embeddings.repeat(1       ,    UpperCAmelCase       ,    1    )
													_UpperCAmelCase									=			uncond_embeddings.view(batch_size * num_images_per_prompt       ,    UpperCAmelCase       ,    -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
													_UpperCAmelCase									=			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`.
							_UpperCAmelCase									=			(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
							_UpperCAmelCase									=			text_embeddings.dtype
							if latents is None:
													if self.device.type == "mps":
																			# randn does not exist on mps
																			_UpperCAmelCase									=			torch.randn(UpperCAmelCase       ,    generator=UpperCAmelCase       ,    device='cpu'       ,    dtype=UpperCAmelCase    ).to(
																			    self.device    )
													else:
																			_UpperCAmelCase									=			torch.randn(UpperCAmelCase       ,    generator=UpperCAmelCase       ,    device=self.device       ,    dtype=UpperCAmelCase    )
							else:
													if latents.shape != latents_shape:
																			raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""    )
													_UpperCAmelCase									=			latents.to(self.device    )

							# set timesteps
							self.scheduler.set_timesteps(UpperCAmelCase    )

							# Some schedulers like PNDM have timesteps as arrays
							# It's more optimized to move all timesteps to correct device beforehand
							_UpperCAmelCase									=			self.scheduler.timesteps.to(self.device    )

							# scale the initial noise by the standard deviation required by the scheduler
							_UpperCAmelCase									=			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]
							_UpperCAmelCase									=			'eta' in set(inspect.signature(self.scheduler.step    ).parameters.keys()    )
							_UpperCAmelCase									=			{}
							if accepts_eta:
													_UpperCAmelCase									=			eta

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

													# predict the noise residual
													_UpperCAmelCase									=			self.unet(UpperCAmelCase       ,    UpperCAmelCase       ,    encoder_hidden_states=UpperCAmelCase    ).sample

													# perform guidance
													if do_classifier_free_guidance:
																			_UpperCAmelCase					, _UpperCAmelCase									=			noise_pred.chunk(2    )
																			_UpperCAmelCase									=			noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

													# compute the previous noisy sample x_t -> x_t-1
													_UpperCAmelCase									=			self.scheduler.step(UpperCAmelCase       ,    UpperCAmelCase       ,    UpperCAmelCase       ,    **UpperCAmelCase    ).prev_sample

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

							_UpperCAmelCase									=			1 / 0.1_82_15 * latents
							_UpperCAmelCase									=			self.vae.decode(UpperCAmelCase    ).sample

							_UpperCAmelCase									=			(image / 2 + 0.5).clamp(0       ,    1    )

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

							if output_type == "pil":
													_UpperCAmelCase									=			self.numpy_to_pil(UpperCAmelCase    )

							if not return_dict:
													return image

							return StableDiffusionPipelineOutput(images=UpperCAmelCase       ,    nsfw_content_detected=UpperCAmelCase    )








"""simple docstring"""



def 						lowercase    (				__snake_case				: Optional[int]	):
		lowercase_ : int							       =   0
		lowercase_ : Optional[Any]							       =   len(__snake_case	)

		for i in range(n - 1	):
				for j in range(i + 1		,					__snake_case	):
						if arr[i] > arr[j]:
								num_inversions += 1

		return num_inversions



def 						lowercase    (				__snake_case				: str	):
		if len(__snake_case	) <= 1:
				return arr, 0
		lowercase_ : Optional[Any]							       =   len(__snake_case	) // 2
		lowercase_ : List[Any]							       =   arr[0:mid]
		lowercase_ : Union[str, Any]							       =   arr[mid:]

		lowercase_       ,					lowercase_ : Tuple							       =   count_inversions_recursive(__snake_case	)
		lowercase_       ,					lowercase_ : List[Any]							       =   count_inversions_recursive(__snake_case	)
		lowercase_       ,					lowercase_ : List[Any]							       =   _count_cross_inversions(__snake_case		,					__snake_case	)

		lowercase_ : List[Any]							       =   inversion_p + inversions_q + cross_inversions
		return c, num_inversions



def 						lowercase    (				__snake_case				: str		,					__snake_case				: Optional[int]	):
		lowercase_ : Optional[Any]							       =   []
		lowercase_ : Any							       =   0
		while i < len(__snake_case	) and j < len(__snake_case	):
				if p[i] > q[j]:
						# if P[1] > Q[j], then P[k] > Q[k] for all  i < k <= len(P)
						# These are all inversions. The claim emerges from the
						# property that P is sorted.
						num_inversion += len(__snake_case	) - i
						r.append(q[j]	)
						j += 1
				else:
						r.append(p[i]	)
						i += 1

		if i < len(__snake_case	):
				r.extend(p[i:]	)
		else:
				r.extend(q[j:]	)

		return r, num_inversion



def 						lowercase    (				):
		lowercase_ : Union[str, Any]							       =   [1_0, 2, 1, 5, 5, 2, 1_1]

		# this arr has 8 inversions:
		# (10, 2), (10, 1), (10, 5), (10, 5), (10, 2), (2, 1), (5, 2), (5, 2)

		lowercase_ : int							       =   count_inversions_bf(__snake_case	)
		lowercase_       ,					lowercase_ : Dict							       =   count_inversions_recursive(__snake_case	)

		assert num_inversions_bf == num_inversions_recursive == 8

		print('''number of inversions = '''		,					__snake_case	)

		# testing an array with zero inversion (a sorted arr_1)

		arr_a.sort()
		lowercase_ : Dict							       =   count_inversions_bf(__snake_case	)
		lowercase_       ,					lowercase_ : Dict							       =   count_inversions_recursive(__snake_case	)

		assert num_inversions_bf == num_inversions_recursive == 0
		print('''number of inversions = '''		,					__snake_case	)

		# an empty list should also have zero inversions
		lowercase_ : List[Any]							       =   []
		lowercase_ : Any							       =   count_inversions_bf(__snake_case	)
		lowercase_       ,					lowercase_ : List[str]							       =   count_inversions_recursive(__snake_case	)

		assert num_inversions_bf == num_inversions_recursive == 0
		print('''number of inversions = '''		,					__snake_case	)


if __name__ == "__main__":
						main()



