


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


snake_case							:	List[str]					      =				logging.get_logger(__name__)  # pylint: disable=invalid-name







class 	snake_case_       (lowerCamelCase_				):
					def __init__(				self       :Any      ,__snake_case       :WhisperForConditionalGeneration      ,__snake_case       :WhisperProcessor      ,__snake_case       :AutoencoderKL      ,__snake_case       :CLIPTextModel      ,__snake_case       :CLIPTokenizer      ,__snake_case       :UNetaDConditionModel      ,__snake_case       :Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]      ,__snake_case       :StableDiffusionSafetyChecker      ,__snake_case       :CLIPImageProcessor      ,) ->   Tuple:
												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 			lowerCamelCase__(				self       :str      ,__snake_case       :Optional[Union[str, int]] = "auto"				) ->   str:
												if slice_size == "auto":
																			a__					=					self.unet.config.attention_head_dim // 2
												self.unet.set_attention_slice(__snake_case				)
					def 			lowerCamelCase__(				self       :Optional[int]				) ->   Union[str, Any]:
												self.enable_attention_slicing(__snake_case				)



					@torch.no_grad()
					def __call__(				self       :int      ,__snake_case       :Any      ,__snake_case       :List[str]=1_60_00      ,__snake_case       :int = 5_12      ,__snake_case       :int = 5_12      ,__snake_case       :int = 50      ,__snake_case       :float = 7.5      ,__snake_case       :Optional[Union[str, List[str]]] = None      ,__snake_case       :Optional[int] = 1      ,__snake_case       :float = 0.0      ,__snake_case       :Optional[torch.Generator] = None      ,__snake_case       :Optional[torch.FloatTensor] = None      ,__snake_case       :Optional[str] = "pil"      ,__snake_case       :bool = True      ,__snake_case       :Optional[Callable[[int, int, torch.FloatTensor], None]] = None      ,__snake_case       :int = 1      ,**__snake_case       :List[str]      ,) ->   List[Any]:
												a__					=					self.speech_processor.feature_extractor(
												    __snake_case      ,return_tensors='pt'      ,sampling_rate=__snake_case				).input_features.to(self.device				)
												a__					=					self.speech_model.generate(__snake_case      ,max_length=48_00_00				)

												a__					=					self.speech_processor.tokenizer.batch_decode(__snake_case      ,skip_special_tokens=__snake_case      ,normalize=__snake_case				)[
												    0
												]

												if isinstance(__snake_case      ,__snake_case				):
																			a__					=					1
												elif isinstance(__snake_case      ,__snake_case				):
																			a__					=					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
												a__					=					self.tokenizer(
												    __snake_case      ,padding='max_length'      ,max_length=self.tokenizer.model_max_length      ,return_tensors='pt'      ,)
												a__					=					text_inputs.input_ids

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

												# duplicate text embeddings for each generation per prompt, using mps friendly method
												a__				, a__				, a__					=					text_embeddings.shape
												a__					=					text_embeddings.repeat(1      ,__snake_case      ,1				)
												a__					=					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.
												a__					=					guidance_scale > 1.0
												# get unconditional embeddings for classifier free guidance
												if do_classifier_free_guidance:
																			a__					=					42
																			if negative_prompt is None:
																										a__					=					[''] * 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				):
																										a__					=					[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:
																										a__					=					negative_prompt

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

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

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

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

																			# predict the noise residual
																			a__					=					self.unet(__snake_case      ,__snake_case      ,encoder_hidden_states=__snake_case				).sample

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

																			# compute the previous noisy sample x_t -> x_t-1
																			a__					=					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				)

												a__					=					1 / 0.1_82_15 * latents
												a__					=					self.vae.decode(__snake_case				).sample

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

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

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

												if not return_dict:
																			return image

												return StableDiffusionPipelineOutput(images=__snake_case      ,nsfw_content_detected=__snake_case				)








def 							__lowercase       (			__lowerCAmelCase							:    Optional[Any]       ):
							return [
							    {
							        0: [1, 2],
							        1: [0, 2],
							        2: [0, 1, 3, 5],
							        3: [2, 4],
							        4: [3],
							        5: [2, 6, 8],
							        6: [5, 7],
							        7: [6, 8],
							        8: [5, 7],
							    },
							    {
							        0: [6],
							        1: [9],
							        2: [4, 5],
							        3: [4],
							        4: [2, 3],
							        5: [2],
							        6: [0, 7],
							        7: [6],
							        8: [],
							        9: [1],
							    },
							    {
							        0: [4],
							        1: [6],
							        2: [],
							        3: [5, 6, 7],
							        4: [0, 6],
							        5: [3, 8, 9],
							        6: [1, 3, 4, 7],
							        7: [3, 6, 8, 9],
							        8: [5, 7],
							        9: [5, 7],
							    },
							    {
							        0: [1, 3],
							        1: [0, 2, 4],
							        2: [1, 3, 4],
							        3: [0, 2, 4],
							        4: [1, 2, 3],
							    },
							][index]






def 							__lowercase       (			__lowerCAmelCase							:    dict[int, list[int]]       ):
							a__					=					0
							a__					=					len(__lowerCAmelCase       )  # No of vertices in graph
							a__					=					[0] * n
							a__					=					[False] * n

							def dfs(__lowerCAmelCase							:    int ,   __lowerCAmelCase							:    Any ,   __lowerCAmelCase							:    Dict ,   __lowerCAmelCase							:    Tuple       ):
														a__					=					True
														a__					=					id_
														id_ += 1
														for to in graph[at]:
																					if to == parent:
																												pass
																					elif not visited[to]:
																												dfs(__lowerCAmelCase ,   __lowerCAmelCase ,   __lowerCAmelCase ,   id_       )
																												a__					=					min(low[at] ,   low[to]       )
																												if id_ <= low[to]:
																																			bridges.append((at, to) if at < to else (to, at)       )
																					else:
																												# This edge is a back edge and cannot be a bridge
																												a__					=					min(low[at] ,   low[to]       )

							a__					=					[]
							for i in range(__lowerCAmelCase       ):
														if not visited[i]:
																					dfs(__lowerCAmelCase ,   -1 ,   __lowerCAmelCase ,   id_       )
							return bridges


if __name__ == "__main__":
						import doctest

						doctest.testmod()




