
import importlib.metadata
import warnings
from copy import deepcopy

from packaging import version

from ..utils import logging
from .import_utils import is_accelerate_available, is_bitsandbytes_available


if is_bitsandbytes_available():
	import bitsandbytes as bnb
	import torch
	import torch.nn as nn

	from ..pytorch_utils import ConvaD

if is_accelerate_available():
	from accelerate import init_empty_weights
	from accelerate.utils import find_tied_parameters

UpperCAmelCase	      = logging.get_logger(__name__)






def 							UpperCAmelCase_		(							__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE=None ,							__SCREAMING_SNAKE_CASE=None							):
	# Recurse if needed
	if "." in tensor_name:
		lowercase       =    tensor_name.split('.'							)
		for split in splits[:-1]:
			lowercase       =    getattr(__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE							)
			if new_module is None:
				raise ValueError(F'''{module} has no attribute {split}.'''							)
			lowercase       =    new_module
		lowercase       =    splits[-1]

	if tensor_name not in module._parameters and tensor_name not in module._buffers:
		raise ValueError(F'''{module} does not have a parameter or a buffer named {tensor_name}.'''							)
	lowercase       =    tensor_name in module._buffers
	lowercase       =    getattr(__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE							)

	if old_value.device == torch.device('meta'							) and device not in ["meta", torch.device('meta'							)] and value is None:
		raise ValueError(F'''{tensor_name} is on the meta device, we need a `value` to put in on {device}.'''							)

	lowercase       =    False
	lowercase       =    False
	if is_buffer or not is_bitsandbytes_available():
		lowercase       =    False
		lowercase       =    False
	else:
		lowercase       =    hasattr(bnb.nn ,							'Params4bit'							) and isinstance(module._parameters[tensor_name] ,							bnb.nn.Paramsabit							)
		lowercase       =    isinstance(module._parameters[tensor_name] ,							bnb.nn.IntaParams							)

	if is_abit or is_abit:
		lowercase       =    module._parameters[tensor_name]
		if param.device.type != "cuda":
			if value is None:
				lowercase       =    old_value.to(__SCREAMING_SNAKE_CASE							)
			elif isinstance(__SCREAMING_SNAKE_CASE ,							torch.Tensor							):
				lowercase       =    value.to('cpu'							)
				if value.dtype == torch.inta:
					lowercase       =    version.parse(importlib.metadata.version('bitsandbytes'							)							) > version.parse(
					    '0.37.2'							)
					if not is_abit_serializable:
						raise ValueError(
						    'Detected int8 weights but the version of bitsandbytes is not compatible with int8 serialization. '
						    'Make sure to download the latest `bitsandbytes` version. `pip install --upgrade bitsandbytes`.'							)
			else:
				lowercase       =    torch.tensor(__SCREAMING_SNAKE_CASE ,							device='cpu'							)

			# Support models using `Conv1D` in place of `nn.Linear` (e.g. gpt2) by transposing the weight matrix prior to quantization.
			# Since weights are saved in the correct "orientation", we skip transposing when loading.
			if issubclass(module.source_cls ,							__SCREAMING_SNAKE_CASE							) and fpaa_statistics is None:
				lowercase       =    new_value.T

			lowercase       =    old_value.__dict__
			if is_abit:
				lowercase       =    bnb.nn.IntaParams(__SCREAMING_SNAKE_CASE ,							requires_grad=__SCREAMING_SNAKE_CASE ,							**__SCREAMING_SNAKE_CASE							).to(__SCREAMING_SNAKE_CASE							)
			elif is_abit:
				lowercase       =    bnb.nn.Paramsabit(__SCREAMING_SNAKE_CASE ,							requires_grad=__SCREAMING_SNAKE_CASE ,							**__SCREAMING_SNAKE_CASE							).to(__SCREAMING_SNAKE_CASE							)

			lowercase       =    new_value
			if fpaa_statistics is not None:
				setattr(module.weight ,							'SCB' ,							fpaa_statistics.to(__SCREAMING_SNAKE_CASE							)							)

	else:
		if value is None:
			lowercase       =    old_value.to(__SCREAMING_SNAKE_CASE							)
		elif isinstance(__SCREAMING_SNAKE_CASE ,							torch.Tensor							):
			lowercase       =    value.to(__SCREAMING_SNAKE_CASE							)
		else:
			lowercase       =    torch.tensor(__SCREAMING_SNAKE_CASE ,							device=__SCREAMING_SNAKE_CASE							)

		if is_buffer:
			lowercase       =    new_value
		else:
			lowercase       =    nn.Parameter(__SCREAMING_SNAKE_CASE ,							requires_grad=old_value.requires_grad							)
			lowercase       =    new_value






def 							UpperCAmelCase_		(							__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE=None ,							__SCREAMING_SNAKE_CASE=None ,							__SCREAMING_SNAKE_CASE=None ,							__SCREAMING_SNAKE_CASE=False							):
	for name, module in model.named_children():
		if current_key_name is None:
			lowercase       =    []
		current_key_name.append(__SCREAMING_SNAKE_CASE							)

		if (isinstance(__SCREAMING_SNAKE_CASE ,							nn.Linear							) or isinstance(__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE							)) and name not in modules_to_not_convert:
			# Check if the current key is not in the `modules_to_not_convert`
			if not any(key in '.'.join(__SCREAMING_SNAKE_CASE							) for key in modules_to_not_convert							):
				with init_empty_weights():
					if isinstance(__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE							):
						lowercase  ,					lowercase       =    module.weight.shape
					else:
						lowercase       =    module.in_features
						lowercase       =    module.out_features

					if quantization_config.quantization_method() == "llm_int8":
						lowercase       =    bnb.nn.LinearabitLt(
						    __SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE ,							module.bias is not None ,							has_fpaa_weights=quantization_config.llm_inta_has_fpaa_weight ,							threshold=quantization_config.llm_inta_threshold ,							)
						lowercase       =    True
					else:
						if (
						    quantization_config.llm_inta_skip_modules is not None
						    and name in quantization_config.llm_inta_skip_modules
						):
							pass
						else:
							lowercase       =    bnb.nn.Linearabit(
							    __SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE ,							module.bias is not None ,							quantization_config.bnb_abit_compute_dtype ,							compress_statistics=quantization_config.bnb_abit_use_double_quant ,							quant_type=quantization_config.bnb_abit_quant_type ,							)
							lowercase       =    True
                    # Store the module class in case we need to transpose the weight later
					lowercase       =    type(__SCREAMING_SNAKE_CASE							)
					# Force requires grad to False to avoid unexpected errors
					model._modules[name].requires_grad_(__SCREAMING_SNAKE_CASE							)
		if len(list(module.children()							)							) > 0:
			lowercase  ,					lowercase       =    _replace_with_bnb_linear(
			    __SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE ,							has_been_replaced=__SCREAMING_SNAKE_CASE ,							)
		# Remove the last key for recursion
		current_key_name.pop(-1							)
	return model, has_been_replaced






def 							UpperCAmelCase_		(							__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE=None ,							__SCREAMING_SNAKE_CASE=None ,							__SCREAMING_SNAKE_CASE=None							):
	lowercase       =    ['lm_head'] if modules_to_not_convert is None else modules_to_not_convert
	lowercase  ,					lowercase       =    _replace_with_bnb_linear(
	    __SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE							)

	if not has_been_replaced:
		logger.warning(
		    'You are loading your model in 8bit or 4bit but no linear modules were found in your model.'
		    ' Please double check your model architecture, or submit an issue on github if you think this is'
		    ' a bug.'							)

	return model






def 							UpperCAmelCase_		(							*__SCREAMING_SNAKE_CASE ,							**__SCREAMING_SNAKE_CASE							):
	warnings.warn(
	    '`replace_8bit_linear` will be deprecated in a future version, please use `replace_with_bnb_linear` instead' ,							__SCREAMING_SNAKE_CASE ,							)
	return replace_with_bnb_linear(*__SCREAMING_SNAKE_CASE ,							**__SCREAMING_SNAKE_CASE							)






def 							UpperCAmelCase_		(							*__SCREAMING_SNAKE_CASE ,							**__SCREAMING_SNAKE_CASE							):
	warnings.warn(
	    '`set_module_8bit_tensor_to_device` will be deprecated in a future version, please use `set_module_quantized_tensor_to_device` instead' ,							__SCREAMING_SNAKE_CASE ,							)
	return set_module_quantized_tensor_to_device(*__SCREAMING_SNAKE_CASE ,							**__SCREAMING_SNAKE_CASE							)






def 							UpperCAmelCase_		(							__SCREAMING_SNAKE_CASE							):
	lowercase       =    deepcopy(__SCREAMING_SNAKE_CASE							)  # this has 0 cost since it is done inside `init_empty_weights` context manager`
	tied_model.tie_weights()

	lowercase       =    find_tied_parameters(__SCREAMING_SNAKE_CASE							)
	# For compatibility with Accelerate < 0.18
	if isinstance(__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE							):
		lowercase       =    sum(list(tied_params.values()							) ,							[]							) + list(tied_params.keys()							)
	else:
		lowercase       =    sum(__SCREAMING_SNAKE_CASE ,							[]							)
	lowercase       =    len(__SCREAMING_SNAKE_CASE							) > 0

	# Check if it is a base model
	lowercase       =    not hasattr(__SCREAMING_SNAKE_CASE ,							model.base_model_prefix							)

	# Ignore this for base models (BertModel, GPT2Model, etc.)
	if (not has_tied_params) and is_base_model:
		return []

	# otherwise they have an attached head
	lowercase       =    list(model.named_children()							)
	lowercase       =    [list_modules[-1][0]]

	# add last module together with tied weights
	lowercase       =    set(__SCREAMING_SNAKE_CASE							) - set(__SCREAMING_SNAKE_CASE							)
	lowercase       =    list(set(__SCREAMING_SNAKE_CASE							)							) + list(__SCREAMING_SNAKE_CASE							)

	# remove ".weight" from the keys
	lowercase       =    ['.weight', '.bias']
	lowercase       =    []
	for name in list_untouched:
		for name_to_remove in names_to_remove:
			if name_to_remove in name:
				lowercase       =    name.replace(__SCREAMING_SNAKE_CASE ,							''							)
		filtered_module_names.append(__SCREAMING_SNAKE_CASE							)

	return filtered_module_names



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


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







class        A_   (			__lowerCamelCase   ):




	'''simple docstring'''


	def __init__(       self		,    snake_case		,    snake_case		,    snake_case		,    snake_case		,    snake_case		,    snake_case		,    snake_case		,    snake_case		,    snake_case		,    ):
		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 	SCREAMING_SNAKE_CASE__						(       self		,    snake_case = "auto"       ):
		if slice_size == "auto":
			lowercase       =    self.unet.config.attention_head_dim // 2
		self.unet.set_attention_slice(snake_case       )





	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		self.enable_attention_slicing(snake_case       )





	@torch.no_grad()
	def __call__(       self		,    snake_case		,    snake_case=1_6000		,    snake_case = 512		,    snake_case = 512		,    snake_case = 50		,    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		,    ):
		lowercase       =    self.speech_processor.feature_extractor(
		    snake_case		,    return_tensors='pt'		,    sampling_rate=snake_case       ).input_features.to(self.device       )
		lowercase       =    self.speech_model.generate(snake_case		,    max_length=48_0000       )

		lowercase       =    self.speech_processor.tokenizer.batch_decode(snake_case		,    skip_special_tokens=snake_case		,    normalize=snake_case       )[
		    0
		]

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

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

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

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

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

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

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

			# predict the noise residual
			lowercase       =    self.unet(snake_case		,    snake_case		,    encoder_hidden_states=snake_case       ).sample

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

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

		lowercase       =    1 / 0.18_215 * latents
		lowercase       =    self.vae.decode(snake_case       ).sample

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

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

		if output_type == "pil":
			lowercase       =    self.numpy_to_pil(snake_case       )

		if not return_dict:
			return image

		return StableDiffusionPipelineOutput(images=snake_case		,    nsfw_content_detected=snake_case       )

