



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,
)


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

_lowercase: int      			=		'''
    Examples:
        ```py
        >>> import torch
        >>> import numpy as np

        >>> from diffusers import KandinskyV22PriorPipeline, KandinskyV22ControlnetPipeline
        >>> from transformers import pipeline
        >>> from diffusers.utils import load_image


        >>> def make_hint(image, depth_estimator):
        ...     image = depth_estimator(image)["depth"]
        ...     image = np.array(image)
        ...     image = image[:, :, None]
        ...     image = np.concatenate([image, image, image], axis=2)
        ...     detected_map = torch.from_numpy(image).float() / 255.0
        ...     hint = detected_map.permute(2, 0, 1)
        ...     return hint


        >>> depth_estimator = pipeline("depth-estimation")

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(
        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16
        ... )
        >>> pipe_prior = pipe_prior.to("cuda")

        >>> pipe = KandinskyV22ControlnetPipeline.from_pretrained(
        ...     "kandinsky-community/kandinsky-2-2-controlnet-depth", torch_dtype=torch.float16
        ... )
        >>> pipe = pipe.to("cuda")


        >>> img = load_image(
        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
        ...     "/kandinsky/cat.png"
        ... ).resize((768, 768))

        >>> hint = make_hint(img, depth_estimator).unsqueeze(0).half().to("cuda")

        >>> prompt = "A robot, 4k photo"
        >>> negative_prior_prompt = "lowres, text, error, cropped, worst quality, low quality, jpeg artifacts, ugly, duplicate, morbid, mutilated, out of frame, extra fingers, mutated hands, poorly drawn hands, poorly drawn face, mutation, deformed, blurry, dehydrated, bad anatomy, bad proportions, extra limbs, cloned face, disfigured, gross proportions, malformed limbs, missing arms, missing legs, extra arms, extra legs, fused fingers, too many fingers, long neck, username, watermark, signature"

        >>> generator = torch.Generator(device="cuda").manual_seed(43)

        >>> image_emb, zero_image_emb = pipe_prior(
        ...     prompt=prompt, negative_prompt=negative_prior_prompt, generator=generator
        ... ).to_tuple()

        >>> images = pipe(
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     hint=hint,
        ...     num_inference_steps=50,
        ...     generator=generator,
        ...     height=768,
        ...     width=768,
        ... ).images

        >>> images[0].save("robot_cat.png")
        ```
'''






def     a(      A       :    Any ,     A       :    int ,     A       :    Dict=8       )		->       Any:
					"""simple docstring"""





					a					     =   height // scale_factor**2
					if height % scale_factor**2 != 0:
										new_height += 1
					a					     =   width // scale_factor**2
					if width % scale_factor**2 != 0:
										new_width += 1
					return new_height * scale_factor, new_width * scale_factor







class 	_lowercase							(   _A	):





	"""simple docstring"""




	def __init__(self		,       lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_		,       ):
						"""simple docstring"""

						super().__init__()

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




	def      UpperCamelCase_       (self		,       lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_		):
						"""simple docstring"""

						if latents is None:
											a					     =   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}'''		)
											a					     =   latents.to(lowerCamelCase_		)

						a					     =   latents * scheduler.init_noise_sigma
						return latents




	def      UpperCamelCase_       (self		,       lowerCamelCase_=0		):
						"""simple docstring"""

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

						a					     =   torch.device(F'''cuda:{gpu_id}'''		)

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




	def      UpperCamelCase_       (self		,       lowerCamelCase_=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."		)

						a					     =   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)

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

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




	@property
	# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
	def      UpperCamelCase_       (self		):
						"""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		,       lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_ = 512		,       lowerCamelCase_ = 512		,       lowerCamelCase_ = 100		,       lowerCamelCase_ = 4.0		,       lowerCamelCase_ = 1		,       lowerCamelCase_ = None		,       lowerCamelCase_ = None		,       lowerCamelCase_ = "pil"		,       lowerCamelCase_ = True		,       ):
						"""simple docstring"""

						a					     =   self._execution_device

						a					     =   guidance_scale > 1.0

						if isinstance(lowerCamelCase_		,       lowerCamelCase_		):
											a					     =   torch.cat(lowerCamelCase_		,       dim=0		)
						if isinstance(lowerCamelCase_		,       lowerCamelCase_		):
											a					     =   torch.cat(lowerCamelCase_		,       dim=0		)
						if isinstance(lowerCamelCase_		,       lowerCamelCase_		):
											a					     =   torch.cat(lowerCamelCase_		,       dim=0		)

						a					     =   image_embeds.shape[0] * num_images_per_prompt

						if do_classifier_free_guidance:
											a					     =   image_embeds.repeat_interleave(lowerCamelCase_		,       dim=0		)
											a					     =   negative_image_embeds.repeat_interleave(lowerCamelCase_		,       dim=0		)
											a					     =   hint.repeat_interleave(lowerCamelCase_		,       dim=0		)

						a					     =   torch.cat([negative_image_embeds, image_embeds]		,       dim=0		).to(dtype=self.unet.dtype		,       device=lowerCamelCase_		)
						a					     =   torch.cat([hint, hint]		,       dim=0		).to(dtype=self.unet.dtype		,       device=lowerCamelCase_		)
						self.scheduler.set_timesteps(lowerCamelCase_		,       device=lowerCamelCase_		)
						a					     =   self.scheduler.timesteps

						a					     =   self.movq.config.latent_channels

						a					     =   downscale_height_and_width(lowerCamelCase_		,       lowerCamelCase_		,       self.movq_scale_factor		)

						# create initial latent
						a					     =   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
											a					     =   torch.cat([latents] * 2		) if do_classifier_free_guidance else latents

											a					     =   {'''image_embeds''': image_embeds, '''hint''': hint}
											a					     =   self.unet(
											    sample=lowerCamelCase_		,       timestep=lowerCamelCase_		,       encoder_hidden_states=lowerCamelCase_		,       added_cond_kwargs=lowerCamelCase_		,       return_dict=lowerCamelCase_		,       )[0]

											if do_classifier_free_guidance:
																a					     =   noise_pred.split(latents.shape[1]		,       dim=1		)
																a					     =   noise_pred.chunk(2		)
																a					     =   variance_pred.chunk(2		)
																a					     =   noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
																a					     =   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"]
											):
																a					     =   noise_pred.split(latents.shape[1]		,       dim=1		)

											# compute the previous noisy sample x_t -> x_t-1
											a					     =   self.scheduler.step(
											    lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_		,       generator=lowerCamelCase_		,       )[0]
						# post-processing
						a					     =   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"]:
											a					     =   image * 0.5 + 0.5
											a					     =   image.clamp(0		,       1		)
											a					     =   image.cpu().permute(0		,       2		,       3		,       1		).float().numpy()

						if output_type == "pil":
											a					     =   self.numpy_to_pil(lowerCamelCase_		)

						if not return_dict:
											return (image,)

						return ImagePipelineOutput(images=lowerCamelCase_		)








"""simple docstring"""



from __future__ import annotations

import inspect
import unittest

import numpy as np

from transformers import ResNetConfig
from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import cached_property, is_tf_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
						import tensorflow as tf

						from transformers import TFResNetForImageClassification, TFResNetModel
						from transformers.models.resnet.modeling_tf_resnet import TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
						from PIL import Image

						from transformers import AutoImageProcessor





class 					_UpperCAmelCase       :




					def __init__(				self     :						int    ,						A     :						Tuple    ,						A     :						int=3    ,						A     :						List[str]=32    ,						A     :						Dict=3    ,						A     :						Any=10    ,						A     :						Dict=[10, 20, 30, 40]    ,						A     :						Optional[Any]=[1, 1, 2, 1]    ,						A     :						Union[str, Any]=True    ,						A     :						Optional[Any]=True    ,						A     :						Any="relu"    ,						A     :						Optional[Any]=3    ,						A     :						Tuple=None    ,						)  ->  Dict:
							lowercase_ : str							       =   parent
							lowercase_ : List[Any]							       =   batch_size
							lowercase_ : Optional[int]							       =   image_size
							lowercase_ : int							       =   num_channels
							lowercase_ : int							       =   embeddings_size
							lowercase_ : str							       =   hidden_sizes
							lowercase_ : List[str]							       =   depths
							lowercase_ : Dict							       =   is_training
							lowercase_ : int							       =   use_labels
							lowercase_ : Any							       =   hidden_act
							lowercase_ : List[Any]							       =   num_labels
							lowercase_ : Tuple							       =   scope
							lowercase_ : Optional[Any]							       =   len(A   )




					def   A  (				self     :						str   )  ->  Tuple:
							lowercase_ : Optional[int]							       =   floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]   )

							lowercase_ : Union[str, Any]							       =   None
							if self.use_labels:
									lowercase_ : List[str]							       =   ids_tensor([self.batch_size]    ,						self.num_labels   )

							lowercase_ : Optional[int]							       =   self.get_config()

							return config, pixel_values, labels




					def   A  (				self     :						Dict   )  ->  int:
							return ResNetConfig(
							    num_channels=self.num_channels    ,						embeddings_size=self.embeddings_size    ,						hidden_sizes=self.hidden_sizes    ,						depths=self.depths    ,						hidden_act=self.hidden_act    ,						num_labels=self.num_labels    ,						image_size=self.image_size    ,						)




					def   A  (				self     :						str    ,						A     :						Tuple    ,						A     :						str    ,						A     :						str   )  ->  str:
							lowercase_ : str							       =   TFResNetModel(config=A   )
							lowercase_ : Union[str, Any]							       =   model(A   )
							# expected last hidden states: B, C, H // 32, W // 32
							self.parent.assertEqual(
							    result.last_hidden_state.shape    ,						(self.batch_size, self.hidden_sizes[-1], self.image_size // 32, self.image_size // 32)    ,						)




					def   A  (				self     :						Any    ,						A     :						int    ,						A     :						List[Any]    ,						A     :						Optional[Any]   )  ->  Optional[Any]:
							lowercase_ : Tuple							       =   self.num_labels
							lowercase_ : Union[str, Any]							       =   TFResNetForImageClassification(A   )
							lowercase_ : Tuple							       =   model(A    ,						labels=A   )
							self.parent.assertEqual(result.logits.shape    ,						(self.batch_size, self.num_labels)   )






					def   A  (				self     :						Union[str, Any]   )  ->  Tuple:
							lowercase_ : Tuple							       =   self.prepare_config_and_inputs()
							lowercase_       ,					lowercase_       ,					lowercase_ : Dict							       =   config_and_inputs
							lowercase_ : int							       =   {'''pixel_values''': pixel_values}
							return config, inputs_dict





@require_tf
class 					_UpperCAmelCase       ( _A   ,						_A   ,						unittest.TestCase							):
					SCREAMING_SNAKE_CASE_		:				Union[str, Any]							    =   (TFResNetModel, TFResNetForImageClassification) if is_tf_available() else ()
					SCREAMING_SNAKE_CASE_		:				List[Any]							    =   (
					    {"feature-extraction": TFResNetModel, "image-classification": TFResNetForImageClassification}
					    if is_tf_available()
					    else {}
					)

					SCREAMING_SNAKE_CASE_		:				Optional[int]							    =   False
					SCREAMING_SNAKE_CASE_		:				Optional[int]							    =   False
					SCREAMING_SNAKE_CASE_		:				str							    =   False
					SCREAMING_SNAKE_CASE_		:				Optional[int]							    =   False
					SCREAMING_SNAKE_CASE_		:				Any							    =   False




					def   A  (				self     :						Union[str, Any]   )  ->  List[Any]:
							lowercase_ : int							       =   TFResNetModelTester(self   )
							lowercase_ : str							       =   ConfigTester(self    ,						config_class=A    ,						has_text_modality=A   )




					def   A  (				self     :						Dict   )  ->  Optional[Any]:
							self.create_and_test_config_common_properties()
							self.config_tester.create_and_test_config_to_json_string()
							self.config_tester.create_and_test_config_to_json_file()
							self.config_tester.create_and_test_config_from_and_save_pretrained()
							self.config_tester.create_and_test_config_with_num_labels()
							self.config_tester.check_config_can_be_init_without_params()
							self.config_tester.check_config_arguments_init()




					def   A  (				self     :						Dict   )  ->  List[Any]:
							return




					@unittest.skip(reason='''ResNet does not use inputs_embeds'''   )
					def   A  (				self     :						Any   )  ->  Any:
							pass




					@unittest.skip(reason='''ResNet does not support input and output embeddings'''   )
					def   A  (				self     :						List[str]   )  ->  Optional[Any]:
							pass




					def   A  (				self     :						str   )  ->  Tuple:
							lowercase_       ,					lowercase_ : Optional[Any]							       =   self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
									lowercase_ : int							       =   model_class(A   )
									lowercase_ : str							       =   inspect.signature(model.call   )
									# signature.parameters is an OrderedDict => so arg_names order is deterministic
									lowercase_ : str							       =   [*signature.parameters.keys()]

									lowercase_ : str							       =   ['''pixel_values''']
									self.assertListEqual(arg_names[:1]    ,						A   )




					def   A  (				self     :						List[str]   )  ->  Tuple:
							lowercase_ : Optional[Any]							       =   self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*A   )




					def   A  (				self     :						List[Any]   )  ->  List[str]:
							def check_hidden_states_output(A     :						Union[str, Any]    ,						A     :						int    ,						A     :						List[Any]   ):
									lowercase_ : int							       =   model_class(A   )
									lowercase_ : Optional[Any]							       =   model(**self._prepare_for_class(A    ,						A   )   )

									lowercase_ : Union[str, Any]							       =   outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

									lowercase_ : Any							       =   self.model_tester.num_stages
									self.assertEqual(len(A   )    ,						expected_num_stages + 1   )

									# ResNet's feature maps are of shape (batch_size, num_channels, height, width)
									self.assertListEqual(
									    list(hidden_states[0].shape[-2:]   )    ,						[self.model_tester.image_size // 4, self.model_tester.image_size // 4]    ,						)

							lowercase_       ,					lowercase_ : List[Any]							       =   self.model_tester.prepare_config_and_inputs_for_common()
							lowercase_ : Union[str, Any]							       =   ['''basic''', '''bottleneck''']
							for model_class in self.all_model_classes:
									for layer_type in layers_type:
											lowercase_ : List[str]							       =   layer_type
											lowercase_ : Tuple							       =   True
											check_hidden_states_output(A    ,						A    ,						A   )

											# check that output_hidden_states also work using config
											del inputs_dict["output_hidden_states"]
											lowercase_ : Optional[Any]							       =   True

											check_hidden_states_output(A    ,						A    ,						A   )




					def   A  (				self     :						Optional[int]   )  ->  Tuple:
							lowercase_ : Optional[int]							       =   self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_image_classification(*A   )






					@slow
					def   A  (				self     :						List[str]   )  ->  Optional[int]:
							for model_name in TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
									lowercase_ : Tuple							       =   TFResNetModel.from_pretrained(A   )
									self.assertIsNotNone(A   )



def 						lowercase    (				):
		lowercase_ : List[str]							       =   Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''	)
		return image






@require_tf
@require_vision
class 					_UpperCAmelCase       ( unittest.TestCase							):




					@cached_property
					def   A  (				self     :						Any   )  ->  Optional[int]:
							return (
							    AutoImageProcessor.from_pretrained(TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]   )
							    if is_vision_available()
							    else None
							)






					@slow
					def   A  (				self     :						Any   )  ->  Optional[int]:
							lowercase_ : Optional[int]							       =   TFResNetForImageClassification.from_pretrained(TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]   )

							lowercase_ : List[Any]							       =   self.default_image_processor
							lowercase_ : Dict							       =   prepare_img()
							lowercase_ : List[str]							       =   image_processor(images=A    ,						return_tensors='''tf'''   )

							# forward pass
							lowercase_ : Tuple							       =   model(**A   )

							# verify the logits
							lowercase_ : Optional[int]							       =   tf.TensorShape((1, 10_00)   )
							self.assertEqual(outputs.logits.shape    ,						A   )

							lowercase_ : Optional[Any]							       =   tf.constant([-11.1069, -9.7877, -8.3777]   )

							self.assertTrue(np.allclose(outputs.logits[0, :3].numpy()    ,						A    ,						atol=1e-4   )   )



