






"""simple docstring"""
# 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.
import copy
import importlib.metadata
import json
import os
from dataclasses import dataclass
from typing import Any, Dict, Union

from packaging import version

from ..utils import is_torch_available, logging


if is_torch_available():
			import torch


lowerCamelCase_							      = logging.get_logger(__name__)







@dataclass
class      UpperCamelCase_			:





	def __init__( self   :					Union[str, Any]      ,      lowerCAmelCase_   :					Optional[Any]=False      ,      lowerCAmelCase_   :					Optional[Any]=False      ,      lowerCAmelCase_   :					str=6.0      ,      lowerCAmelCase_   :					int=None      ,      lowerCAmelCase_   :					Dict=False      ,      lowerCAmelCase_   :					Union[str, Any]=False      ,      lowerCAmelCase_   :					str=None      ,      lowerCAmelCase_   :					int="fp4"      ,      lowerCAmelCase_   :					List[Any]=False      ,      **lowerCAmelCase_   :					str      ,      )		->     Optional[int]:
				UpperCAmelCase_	:					Union[str, Any]     	=   load_in_abit
				UpperCAmelCase_	:					Optional[Any]     	=   load_in_abit
				UpperCAmelCase_	:					Union[str, Any]     	=   llm_inta_threshold
				UpperCAmelCase_	:					Optional[Any]     	=   llm_inta_skip_modules
				UpperCAmelCase_	:					Union[str, Any]     	=   llm_inta_enable_fpaa_cpu_offload
				UpperCAmelCase_	:					Optional[int]     	=   llm_inta_has_fpaa_weight
				UpperCAmelCase_	:					Union[str, Any]     	=   bnb_abit_quant_type
				UpperCAmelCase_	:					Dict     	=   bnb_abit_use_double_quant

				if bnb_abit_compute_dtype is None:
							UpperCAmelCase_	:					List[Any]     	=   torch.floataa
				elif isinstance(lowerCAmelCase_      ,      lowerCAmelCase_		):
							UpperCAmelCase_	:					str     	=   getattr(lowerCAmelCase_      ,      lowerCAmelCase_		)
				elif isinstance(lowerCAmelCase_      ,      torch.dtype		):
							UpperCAmelCase_	:					str     	=   bnb_abit_compute_dtype
				else:
							raise ValueError("bnb_4bit_compute_dtype must be a string or a torch.dtype"		)

				self.post_init()





	def 	_SCREAMING_SNAKE_CASE ( self   :					List[str]		)		->     Optional[Any]:
				if not isinstance(self.llm_inta_threshold      ,      lowerCAmelCase_		):
							raise ValueError("llm_int8_threshold must be a float"		)

				if self.llm_inta_skip_modules is not None and not isinstance(self.llm_inta_skip_modules      ,      lowerCAmelCase_		):
							raise ValueError("llm_int8_skip_modules must be a list of strings"		)
				if not isinstance(self.llm_inta_enable_fpaa_cpu_offload      ,      lowerCAmelCase_		):
							raise ValueError("llm_int8_enable_fp32_cpu_offload must be a boolean"		)

				if not isinstance(self.llm_inta_has_fpaa_weight      ,      lowerCAmelCase_		):
							raise ValueError("llm_int8_has_fp16_weight must be a boolean"		)

				if self.bnb_abit_compute_dtype is not None and not isinstance(self.bnb_abit_compute_dtype      ,      torch.dtype		):
							raise ValueError("bnb_4bit_compute_dtype must be torch.dtype"		)

				if not isinstance(self.bnb_abit_quant_type      ,      lowerCAmelCase_		):
							raise ValueError("bnb_4bit_quant_type must be a string"		)

				if not isinstance(self.bnb_abit_use_double_quant      ,      lowerCAmelCase_		):
							raise ValueError("bnb_4bit_use_double_quant must be a boolean"		)

				if self.load_in_abit and not version.parse(importlib.metadata.version("bitsandbytes"		)		) >= version.parse(
				    "0.39.0"		):
							raise ValueError(
							    "4 bit quantization requires bitsandbytes>=0.39.0 - please upgrade your bitsandbytes version"		)





	def 	_SCREAMING_SNAKE_CASE ( self   :					str		)		->     List[str]:
				return self.load_in_abit or self.load_in_abit





	def 	_SCREAMING_SNAKE_CASE ( self   :					List[str]		)		->     Tuple:
				if self.load_in_abit:
							return "llm_int8"
				elif self.load_in_abit and self.bnb_abit_quant_type == "fp4":
							return "fp4"
				elif self.load_in_abit and self.bnb_abit_quant_type == "nf4":
							return "nf4"
				else:
							return None





	@classmethod
	def 	_SCREAMING_SNAKE_CASE ( cls   :					List[Any]      ,      lowerCAmelCase_   :					Optional[int]      ,      lowerCAmelCase_   :					str      ,      **lowerCAmelCase_   :					Optional[Any]		)		->     Any:
				UpperCAmelCase_	:					str     	=   cls(**lowerCAmelCase_		)

				UpperCAmelCase_	:					int     	=   []
				for key, value in kwargs.items():
							if hasattr(lowerCAmelCase_      ,      lowerCAmelCase_		):
										setattr(lowerCAmelCase_      ,      lowerCAmelCase_      ,      lowerCAmelCase_		)
										to_remove.append(lowerCAmelCase_		)
				for key in to_remove:
							kwargs.pop(lowerCAmelCase_      ,      lowerCAmelCase_		)

				if return_unused_kwargs:
							return config, kwargs
				else:
							return config





	def 	_SCREAMING_SNAKE_CASE ( self   :					Tuple      ,      lowerCAmelCase_   :					Union[str, os.PathLike]		)		->     Optional[int]:
				with open(lowerCAmelCase_      ,      "w"      ,      encoding="utf-8"		) as writer:
							UpperCAmelCase_	:					Any     	=   self.to_dict()
							UpperCAmelCase_	:					Optional[int]     	=   json.dumps(lowerCAmelCase_      ,      indent=2      ,      sort_keys=lowerCAmelCase_		) + "\n"

							writer.write(lowerCAmelCase_		)





	def 	_SCREAMING_SNAKE_CASE ( self   :					Union[str, Any]		)		->     Dict[str, Any]:
				UpperCAmelCase_	:					Optional[int]     	=   copy.deepcopy(self.__dict__		)
				UpperCAmelCase_	:					Union[str, Any]     	=   str(output["bnb_4bit_compute_dtype"]		).split("."		)[1]

				return output





	def __repr__( self   :					List[str]		)		->     List[Any]:
				return f"""{self.__class__.__name__} {self.to_json_string()}"""





	def 	_SCREAMING_SNAKE_CASE ( self   :					str      ,      lowerCAmelCase_   :					bool = True		)		->     str:
				if use_diff is True:
							UpperCAmelCase_	:					Dict     	=   self.to_diff_dict()
				else:
							UpperCAmelCase_	:					Optional[Any]     	=   self.to_dict()
				return json.dumps(lowerCAmelCase_      ,      indent=2      ,      sort_keys=lowerCAmelCase_		) + "\n"





	def 	_SCREAMING_SNAKE_CASE ( self   :					str		)		->     Dict[str, Any]:
				UpperCAmelCase_	:					Dict     	=   self.to_dict()

				# get the default config dict
				UpperCAmelCase_	:					int     	=   BitsAndBytesConfig().to_dict()

				UpperCAmelCase_	:					Dict     	=   {}

				# only serialize values that differ from the default config
				for key, value in config_dict.items():
							if value != default_config_dict[key]:
										UpperCAmelCase_	:					Any     	=   value

				return serializable_config_dict











"""simple docstring"""
from typing import Dict, List, Optional, Tuple, Union

import torch

from ...models import AutoencoderKL, TransformeraDModel
from ...schedulers import KarrasDiffusionSchedulers
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput







class      UpperCamelCase_			(__A	):





	def __init__( self   :					List[Any]      ,      lowerCAmelCase_   :					TransformeraDModel      ,      lowerCAmelCase_   :					AutoencoderKL      ,      lowerCAmelCase_   :					KarrasDiffusionSchedulers      ,      lowerCAmelCase_   :					Optional[Dict[int, str]] = None      ,      )		->     List[Any]:
				super().__init__()
				self.register_modules(transformer=lowerCAmelCase_      ,      vae=lowerCAmelCase_      ,      scheduler=lowerCAmelCase_		)

				# create a imagenet -> id dictionary for easier use
				UpperCAmelCase_	:					str     	=   {}
				if idalabel is not None:
							for key, value in idalabel.items():
										for label in value.split(","		):
													UpperCAmelCase_	:					Union[str, Any]     	=   int(lowerCAmelCase_		)
							UpperCAmelCase_	:					Optional[int]     	=   dict(sorted(self.labels.items()		)		)





	def 	_SCREAMING_SNAKE_CASE ( self   :					str      ,      lowerCAmelCase_   :					Union[str, List[str]]		)		->     List[int]:

				if not isinstance(lowerCAmelCase_      ,      lowerCAmelCase_		):
							UpperCAmelCase_	:					str     	=   list(lowerCAmelCase_		)

				for l in label:
							if l not in self.labels:
										raise ValueError(
										    f"""{l} does not exist. Please make sure to select one of the following labels: \n {self.labels}."""		)

				return [self.labels[l] for l in label]





	@torch.no_grad()
	def __call__( self   :					List[str]      ,      lowerCAmelCase_   :					List[int]      ,      lowerCAmelCase_   :					float = 4.0      ,      lowerCAmelCase_   :					Optional[Union[torch.Generator, List[torch.Generator]]] = None      ,      lowerCAmelCase_   :					int = 50      ,      lowerCAmelCase_   :					Optional[str] = "pil"      ,      lowerCAmelCase_   :					bool = True      ,      )		->     Union[ImagePipelineOutput, Tuple]:
				UpperCAmelCase_	:					Optional[int]     	=   len(lowerCAmelCase_		)
				UpperCAmelCase_	:					int     	=   self.transformer.config.sample_size
				UpperCAmelCase_	:					Optional[Any]     	=   self.transformer.config.in_channels

				UpperCAmelCase_	:					Tuple     	=   randn_tensor(
				    shape=(batch_size, latent_channels, latent_size, latent_size)      ,      generator=lowerCAmelCase_      ,      device=self.device      ,      dtype=self.transformer.dtype      ,      )
				UpperCAmelCase_	:					str     	=   torch.cat([latents] * 2		) if guidance_scale > 1 else latents

				UpperCAmelCase_	:					List[Any]     	=   torch.tensor(lowerCAmelCase_      ,      device=self.device		).reshape(-1		)
				UpperCAmelCase_	:					Optional[Any]     	=   torch.tensor([1_000] * batch_size      ,      device=self.device		)
				UpperCAmelCase_	:					int     	=   torch.cat([class_labels, class_null]      ,      0		) if guidance_scale > 1 else class_labels

				# set step values
				self.scheduler.set_timesteps(lowerCAmelCase_		)

				for t in self.progress_bar(self.scheduler.timesteps		):
							if guidance_scale > 1:
										UpperCAmelCase_	:					Optional[Any]     	=   latent_model_input[: len(lowerCAmelCase_		) // 2]
										UpperCAmelCase_	:					int     	=   torch.cat([half, half]      ,      dim=0		)
							UpperCAmelCase_	:					Optional[int]     	=   self.scheduler.scale_model_input(lowerCAmelCase_      ,      lowerCAmelCase_		)

							UpperCAmelCase_	:					Union[str, Any]     	=   t
							if not torch.is_tensor(lowerCAmelCase_		):
										# TODO: this requires sync between CPU and GPU. So try to pass timesteps as tensors if you can
										# This would be a good case for the `match` statement (Python 3.10+)
										UpperCAmelCase_	:					List[Any]     	=   latent_model_input.device.type == "mps"
										if isinstance(lowerCAmelCase_      ,      lowerCAmelCase_		):
													UpperCAmelCase_	:					Optional[int]     	=   torch.floataa if is_mps else torch.floataa
										else:
													UpperCAmelCase_	:					int     	=   torch.intaa if is_mps else torch.intaa
										UpperCAmelCase_	:					Optional[int]     	=   torch.tensor([timesteps]      ,      dtype=lowerCAmelCase_      ,      device=latent_model_input.device		)
							elif len(timesteps.shape		) == 0:
										UpperCAmelCase_	:					List[Any]     	=   timesteps[None].to(latent_model_input.device		)
							# broadcast to batch dimension in a way that's compatible with ONNX/Core ML
							UpperCAmelCase_	:					List[Any]     	=   timesteps.expand(latent_model_input.shape[0]		)
							# predict noise model_output
							UpperCAmelCase_	:					List[Any]     	=   self.transformer(
							    lowerCAmelCase_      ,      timestep=lowerCAmelCase_      ,      class_labels=lowerCAmelCase_		).sample

							# perform guidance
							if guidance_scale > 1:
										UpperCAmelCase_	,      UpperCAmelCase_	:					Optional[int]     	=   noise_pred[:, :latent_channels], noise_pred[:, latent_channels:]
										UpperCAmelCase_	,      UpperCAmelCase_	:					Union[str, Any]     	=   torch.split(lowerCAmelCase_      ,      len(lowerCAmelCase_		) // 2      ,      dim=0		)

										UpperCAmelCase_	:					Optional[Any]     	=   uncond_eps + guidance_scale * (cond_eps - uncond_eps)
										UpperCAmelCase_	:					Tuple     	=   torch.cat([half_eps, half_eps]      ,      dim=0		)

										UpperCAmelCase_	:					Optional[Any]     	=   torch.cat([eps, rest]      ,      dim=1		)

							# learned sigma
							if self.transformer.config.out_channels // 2 == latent_channels:
										UpperCAmelCase_	,      UpperCAmelCase_	:					List[str]     	=   torch.split(lowerCAmelCase_      ,      lowerCAmelCase_      ,      dim=1		)
							else:
										UpperCAmelCase_	:					Optional[Any]     	=   noise_pred

							# compute previous image: x_t -> x_t-1
							UpperCAmelCase_	:					Union[str, Any]     	=   self.scheduler.step(lowerCAmelCase_      ,      lowerCAmelCase_      ,      lowerCAmelCase_		).prev_sample

				if guidance_scale > 1:
							UpperCAmelCase_	,      UpperCAmelCase_	:					Optional[Any]     	=   latent_model_input.chunk(2      ,      dim=0		)
				else:
							UpperCAmelCase_	:					Optional[int]     	=   latent_model_input

				UpperCAmelCase_	:					Dict     	=   1 / self.vae.config.scaling_factor * latents
				UpperCAmelCase_	:					Dict     	=   self.vae.decode(lowerCAmelCase_		).sample

				UpperCAmelCase_	:					Tuple     	=   (samples / 2 + 0.5).clamp(0      ,      1		)

				# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
				UpperCAmelCase_	:					Optional[Any]     	=   samples.cpu().permute(0      ,      2      ,      3      ,      1		).float().numpy()

				if output_type == "pil":
							UpperCAmelCase_	:					Optional[int]     	=   self.numpy_to_pil(lowerCAmelCase_		)

				if not return_dict:
							return (samples,)

				return ImagePipelineOutput(images=lowerCAmelCase_		)



