




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


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




class 		UpperCamelCase__	(lowerCAmelCase__     ):

    '''simple docstring'''



    def __init__(    self     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  )     ->      Any:
           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     _lowercase     (    self     ,  UpperCamelCase__ = "auto"     )     ->      Optional[Any]:
           if slice_size == "auto":
                  lowerCamelCase   :      Union[str, Any]			 =					self.unet.config.attention_head_dim // 2
           self.unet.set_attention_slice(UpperCamelCase__     )







    def     _lowercase     (    self     )     ->      Union[str, Any]:
           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__     ,  )     ->      str:
           lowerCamelCase   :      Dict			 =					self.speech_processor.feature_extractor(
               UpperCamelCase__     ,  return_tensors="pt"     ,  sampling_rate=UpperCamelCase__     ).input_features.to(self.device     )
           lowerCamelCase   :      Any			 =					self.speech_model.generate(UpperCamelCase__     ,  max_length=48_0000     )

           lowerCamelCase   :      List[str]			 =					self.speech_processor.tokenizer.batch_decode(UpperCamelCase__     ,  skip_special_tokens=UpperCamelCase__     ,  normalize=UpperCamelCase__     )[
               0
           ]

           if isinstance(UpperCamelCase__     ,  UpperCamelCase__     ):
                  lowerCamelCase   :      str			 =					1
           elif isinstance(UpperCamelCase__     ,  UpperCamelCase__     ):
                  lowerCamelCase   :      Any			 =					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
           lowerCamelCase   :      Union[str, Any]			 =					self.tokenizer(
               UpperCamelCase__     ,  padding="max_length"     ,  max_length=self.tokenizer.model_max_length     ,  return_tensors="pt"     ,  )
           lowerCamelCase   :      List[str]			 =					text_inputs.input_ids

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

           # duplicate text embeddings for each generation per prompt, using mps friendly method
           lowerCamelCase ,      lowerCamelCase ,      lowerCamelCase   :      List[str]			 =					text_embeddings.shape
           lowerCamelCase   :      Dict			 =					text_embeddings.repeat(1     ,  UpperCamelCase__     ,  1     )
           lowerCamelCase   :      int			 =					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.
           lowerCamelCase   :      Union[str, Any]			 =					guidance_scale > 1.0
           # get unconditional embeddings for classifier free guidance
           if do_classifier_free_guidance:
                  lowerCamelCase   :      List[str]
                  if negative_prompt is None:
                         lowerCamelCase   :      Dict			 =					[""] * 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__     ):
                         lowerCamelCase   :      Union[str, Any]			 =					[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:
                         lowerCamelCase   :      Tuple			 =					negative_prompt

                  lowerCamelCase   :      Any			 =					text_input_ids.shape[-1]
                  lowerCamelCase   :      Dict			 =					self.tokenizer(
                      UpperCamelCase__     ,  padding="max_length"     ,  max_length=UpperCamelCase__     ,  truncation=UpperCamelCase__     ,  return_tensors="pt"     ,  )
                  lowerCamelCase   :      Optional[int]			 =					self.text_encoder(uncond_input.input_ids.to(self.device     )     )[0]

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

           # scale the initial noise by the standard deviation required by the scheduler
           lowerCamelCase   :      Dict			 =					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]
           lowerCamelCase   :      int			 =					"eta" in set(inspect.signature(self.scheduler.step     ).parameters.keys()     )
           lowerCamelCase   :      List[str]			 =					{}
           if accepts_eta:
                  lowerCamelCase   :      Tuple			 =					eta

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

                  # predict the noise residual
                  lowerCamelCase   :      Optional[int]			 =					self.unet(UpperCamelCase__     ,  UpperCamelCase__     ,  encoder_hidden_states=UpperCamelCase__     ).sample

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

                  # compute the previous noisy sample x_t -> x_t-1
                  lowerCamelCase   :      Optional[int]			 =					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__     )

           lowerCamelCase   :      List[str]			 =					1 / 0.18215 * latents
           lowerCamelCase   :      List[Any]			 =					self.vae.decode(UpperCamelCase__     ).sample

           lowerCamelCase   :      Tuple			 =					(image / 2 + 0.5).clamp(0     ,  1     )

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

           if output_type == "pil":
                  lowerCamelCase   :      Optional[int]			 =					self.numpy_to_pil(UpperCamelCase__     )

           if not return_dict:
                  return image

           return StableDiffusionPipelineOutput(images=UpperCamelCase__     ,  nsfw_content_detected=UpperCamelCase__     )








from __future__ import annotations




def 	_SCREAMING_SNAKE_CASE					(		_lowerCamelCase :			list[list[int]])					->    bool:






							'''simple docstring'''





							__UpperCamelCase	:   Any			   =					len(_lowerCamelCase)
							# We need to create solution object to save path.
							__UpperCamelCase	:   List[str]			   =					[[0 for _ in range(_lowerCamelCase)] for _ in range(_lowerCamelCase)]
							__UpperCamelCase	:   Optional[int]			   =					run_maze(_lowerCamelCase						,  0						,  0						,  _lowerCamelCase)
							if solved:
														print("\n".join(str(_lowerCamelCase) for row in solutions))
							else:
														print("No solution exists!")
							return solved




def 	_SCREAMING_SNAKE_CASE					(		_lowerCamelCase :			list[list[int]]						,  _lowerCamelCase :			int						,  _lowerCamelCase :			int						,  _lowerCamelCase :			list[list[int]])					->    bool:






							'''simple docstring'''





							__UpperCamelCase	:   Tuple			   =					len(_lowerCamelCase)
							# Final check point.
							if i == j == (size - 1):
														__UpperCamelCase	:   Optional[int]			   =					1
														return True

							__UpperCamelCase	:   List[Any]			   =					(not i < 0) and (not j < 0)  # Check lower bounds
							__UpperCamelCase	:   List[str]			   =					(i < size) and (j < size)  # Check upper bounds

							if lower_flag and upper_flag:
														# check for already visited and block points.
														__UpperCamelCase	:   int			   =					(not solutions[i][j]) and (not maze[i][j])
														if block_flag:
																					# check visited
																					__UpperCamelCase	:   Tuple			   =					1

																					# check for directions
																					if (
																					    run_maze(_lowerCamelCase						,  i + 1						,  _lowerCamelCase						,  _lowerCamelCase)
																					    or run_maze(_lowerCamelCase						,  _lowerCamelCase						,  j + 1						,  _lowerCamelCase)
																					    or run_maze(_lowerCamelCase						,  i - 1						,  _lowerCamelCase						,  _lowerCamelCase)
																					    or run_maze(_lowerCamelCase						,  _lowerCamelCase						,  j - 1						,  _lowerCamelCase)
																					):
																												return True

																					__UpperCamelCase	:   Tuple			   =					0
																					return False
							return False


if __name__ == "__main__":
		import doctest

		doctest.testmod()