



from typing import List, Optional, Union

import torch

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)


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

snake_case__					:     List[Any]    		=   """
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline
        >>> import torch

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\"kandinsky-community/kandinsky-2-2-prior\")
        >>> pipe_prior.to(\"cuda\")
        >>> prompt = \"red cat, 4k photo\"
        >>> out = pipe_prior(prompt)
        >>> image_emb = out.image_embeds
        >>> zero_image_emb = out.negative_image_embeds
        >>> pipe = KandinskyV22Pipeline.from_pretrained(\"kandinsky-community/kandinsky-2-2-decoder\")
        >>> pipe.to(\"cuda\")
        >>> image = pipe(
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=50,
        ... ).images
        >>> image[0].save(\"cat.png\")
        ```
"""



def    snake_case_				(    _SCREAMING_SNAKE_CASE  ,	_SCREAMING_SNAKE_CASE  ,	_SCREAMING_SNAKE_CASE=8   ):
	__lowercase		       =			height // scale_factor**2
	if height % scale_factor**2 != 0:
		new_height += 1
	__lowercase		       =			width // scale_factor**2
	if width % scale_factor**2 != 0:
		new_width += 1
	return new_height * scale_factor, new_width * scale_factor

class   _A   ( _lowercase	):


						'''simple docstring'''





						def __init__(	self	:							Optional[int]							,      lowerCamelCase	:							UNetaDConditionModel							,      lowerCamelCase	:							DDPMScheduler							,      lowerCamelCase	:							VQModel							,      ):






							'''simple docstring'''



							super().__init__()

							self.register_modules(
							    unet=lowerCamelCase							,      scheduler=lowerCamelCase							,      movq=lowerCamelCase							,      )
							__lowercase		       =			2 ** (len(self.movq.config.block_out_channels	) - 1)



						def     _snake_case	(	self	:							Tuple							,      lowerCamelCase	:							Optional[Any]							,      lowerCamelCase	:							Union[str, Any]							,      lowerCamelCase	:							Tuple							,      lowerCamelCase	:							Tuple							,      lowerCamelCase	:							List[Any]							,      lowerCamelCase	:							List[Any]	):






							'''simple docstring'''



							if latents is None:
								__lowercase		       =			randn_tensor(lowerCamelCase							,      generator=lowerCamelCase							,      device=lowerCamelCase							,      dtype=lowerCamelCase	)
							else:
								if latents.shape != shape:
									raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {shape}"""	)
								__lowercase		       =			latents.to(lowerCamelCase	)

							__lowercase		       =			latents * scheduler.init_noise_sigma
							return latents



						def     _snake_case	(	self	:							List[Any]							,      lowerCamelCase	:							List[Any]=0	):






							'''simple docstring'''



							if is_accelerate_available():
								from accelerate import cpu_offload
							else:
								raise ImportError("Please install accelerate via `pip install accelerate`"	)

							__lowercase		       =			torch.device(f"""cuda:{gpu_id}"""	)

							__lowercase		       =			[
							    self.unet,
							    self.movq,
							]
							for cpu_offloaded_model in models:
								if cpu_offloaded_model is not None:
									cpu_offload(lowerCamelCase							,      lowerCamelCase	)



						def     _snake_case	(	self	:							Tuple							,      lowerCamelCase	:							Any=0	):






							'''simple docstring'''



							if is_accelerate_available() and is_accelerate_version(">="							,      "0.17.0.dev0"	):
								from accelerate import cpu_offload_with_hook
							else:
								raise ImportError("`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher."	)

							__lowercase		       =			torch.device(f"""cuda:{gpu_id}"""	)

							if self.device.type != "cpu":
								self.to("cpu"							,      silence_dtype_warnings=lowerCamelCase	)
								torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

							__lowercase		       =			None
							for cpu_offloaded_model in [self.unet, self.movq]:
								__lowercase       ,    __lowercase		       =			cpu_offload_with_hook(lowerCamelCase							,      lowerCamelCase							,      prev_module_hook=lowerCamelCase	)

							# We'll offload the last model manually.
							__lowercase		       =			hook



						@property
						# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
						def     _snake_case	(	self	:							List[Any]	):






							'''simple docstring'''



							if not hasattr(self.unet							,      "_hf_hook"	):
								return self.device
							for module in self.unet.modules():
								if (
								    hasattr(lowerCamelCase							,      "_hf_hook"	)
								    and hasattr(module._hf_hook							,      "execution_device"	)
								    and module._hf_hook.execution_device is not None
								):
									return torch.device(module._hf_hook.execution_device	)
							return self.device



						@torch.no_grad()
						@replace_example_docstring(lowerCamelCase	)
						def __call__(	self	:							Any							,      lowerCamelCase	:							Union[torch.FloatTensor, List[torch.FloatTensor]]							,      lowerCamelCase	:							Union[torch.FloatTensor, List[torch.FloatTensor]]							,      lowerCamelCase	:							int = 512							,      lowerCamelCase	:							int = 512							,      lowerCamelCase	:							int = 100							,      lowerCamelCase	:							float = 4.0							,      lowerCamelCase	:							int = 1							,      lowerCamelCase	:							Optional[Union[torch.Generator, List[torch.Generator]]] = None							,      lowerCamelCase	:							Optional[torch.FloatTensor] = None							,      lowerCamelCase	:							Optional[str] = "pil"							,      lowerCamelCase	:							bool = True							,      ):






							'''simple docstring'''



							__lowercase		       =			self._execution_device

							__lowercase		       =			guidance_scale > 1.0

							if isinstance(lowerCamelCase							,      lowerCamelCase	):
								__lowercase		       =			torch.cat(lowerCamelCase							,      dim=0	)
							__lowercase		       =			image_embeds.shape[0] * num_images_per_prompt
							if isinstance(lowerCamelCase							,      lowerCamelCase	):
								__lowercase		       =			torch.cat(lowerCamelCase							,      dim=0	)

							if do_classifier_free_guidance:
								__lowercase		       =			image_embeds.repeat_interleave(lowerCamelCase							,      dim=0	)
								__lowercase		       =			negative_image_embeds.repeat_interleave(lowerCamelCase							,      dim=0	)

							__lowercase		       =			torch.cat([negative_image_embeds, image_embeds]							,      dim=0	).to(dtype=self.unet.dtype							,      device=lowerCamelCase	)

							self.scheduler.set_timesteps(lowerCamelCase							,      device=lowerCamelCase	)
							__lowercase		       =			self.scheduler.timesteps

							__lowercase		       =			self.unet.config.in_channels

							__lowercase       ,    __lowercase		       =			downscale_height_and_width(lowerCamelCase							,      lowerCamelCase							,      self.movq_scale_factor	)

							# create initial latent
							__lowercase		       =			self.prepare_latents(
							    (batch_size, num_channels_latents, height, width)							,      image_embeds.dtype							,      lowerCamelCase							,      lowerCamelCase							,      lowerCamelCase							,      self.scheduler							,      )

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

								__lowercase		       =			{"image_embeds": image_embeds}
								__lowercase		       =			self.unet(
								    sample=lowerCamelCase							,      timestep=lowerCamelCase							,      encoder_hidden_states=lowerCamelCase							,      added_cond_kwargs=lowerCamelCase							,      return_dict=lowerCamelCase							,      )[0]

								if do_classifier_free_guidance:
									__lowercase       ,    __lowercase		       =			noise_pred.split(latents.shape[1]							,      dim=1	)
									__lowercase       ,    __lowercase		       =			noise_pred.chunk(2	)
									__lowercase       ,    __lowercase		       =			variance_pred.chunk(2	)
									__lowercase		       =			noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
									__lowercase		       =			torch.cat([noise_pred, variance_pred_text]							,      dim=1	)

								if not (
								    hasattr(self.scheduler.config							,      "variance_type"	)
								    and self.scheduler.config.variance_type in ["learned", "learned_range"]
								):
									__lowercase       ,    __lowercase		       =			noise_pred.split(latents.shape[1]							,      dim=1	)

								# compute the previous noisy sample x_t -> x_t-1
								__lowercase		       =			self.scheduler.step(
								    lowerCamelCase							,      lowerCamelCase							,      lowerCamelCase							,      generator=lowerCamelCase							,      )[0]
							# post-processing
							__lowercase		       =			self.movq.decode(lowerCamelCase							,      force_not_quantize=lowerCamelCase	)["sample"]

							if output_type not in ["pt", "np", "pil"]:
								raise ValueError(f"""Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}"""	)

							if output_type in ["np", "pil"]:
								__lowercase		       =			image * 0.5 + 0.5
								__lowercase		       =			image.clamp(0							,      1	)
								__lowercase		       =			image.cpu().permute(0							,      2							,      3							,      1	).float().numpy()

							if output_type == "pil":
								__lowercase		       =			self.numpy_to_pil(lowerCamelCase	)

							if not return_dict:
								return (image,)

							return ImagePipelineOutput(images=lowerCamelCase	)








# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ..models.whisper import WhisperForConditionalGeneration, WhisperProcessor
from .base import PipelineTool

class   _A   ( _lowercase	):


						'''simple docstring'''





						_snake_case			:     Dict      							=      """openai/whisper-base"""
						_snake_case			:     Union[str, Any]      							=      (
						    """This is a tool that transcribes an audio into text. It takes an input named `audio` and returns the """
						    """transcribed text."""
						)
						_snake_case			:     Any      							=      """transcriber"""
						_snake_case			:     Any      							=      WhisperProcessor
						_snake_case			:     Optional[int]      							=      WhisperForConditionalGeneration

						_snake_case			:     str      							=      ["""audio"""]
						_snake_case			:     Optional[int]      							=      ["""text"""]



						def     _snake_case	(	self	:							List[str]							,      lowerCamelCase	:							Optional[int]	):






							'''simple docstring'''



							return self.pre_processor(lowerCamelCase							,      return_tensors="pt"	).input_features



						def     _snake_case	(	self	:							str							,      lowerCamelCase	:							List[Any]	):






							'''simple docstring'''



							return self.model.generate(inputs=lowerCamelCase	)



						def     _snake_case	(	self	:							List[str]							,      lowerCamelCase	:							Optional[Any]	):






							'''simple docstring'''



							return self.pre_processor.batch_decode(lowerCamelCase							,      skip_special_tokens=lowerCamelCase	)[0]



